🚧 Initial work on websocket communication.

backend/src/uxbox/http.clj

@@ -44,7 +44,7 @@
                       interceptors/format-response-body
                       (vxi/errors errors/handle)]
 
-        routes [["/sub/:page-id" {:interceptors [(vxi/cookies)
+        routes [["/sub/:file-id" {:interceptors [(vxi/cookies)
                                                  (vxi/cors cors-opts)
                                                  (session/auth)]
                                   :get ws/handler}]

backend/src/uxbox/http/interceptors.clj

@@ -7,7 +7,7 @@
 (ns uxbox.http.interceptors
   (:require
    [vertx.web :as vw]
-   [uxbox.util.blob :as blob]
+   [uxbox.util.transit :as t]
    [uxbox.util.exceptions :as ex])
   (:import
    io.vertx.ext.web.RoutingContext
@@ -20,7 +20,7 @@
                   mtype (get-in request [:headers "content-type"])]
               (if (= "application/transit+json" mtype)
                 (try
-                  (let [params (blob/decode-from-json body)]
+                  (let [params (t/decode (t/buffer->bytes body))]
                     (update data :request assoc :body-params params))
                   (catch Exception e
                     (ex/raise :type :parse
@@ -35,7 +35,7 @@
                 (coll? body)
                 (-> data
                     (assoc-in [:response :body]
-                              (blob/encode-with-json body true))
+                              (t/bytes->buffer (t/encode body)))
                     (update-in [:response :headers]
                                assoc "content-type" "application/transit+json"))
 

backend/src/uxbox/http/ws.clj

@@ -7,6 +7,7 @@
 (ns uxbox.http.ws
   "Web Socket handlers"
   (:require
+   [clojure.tools.logging :as log]
    [promesa.core :as p]
    [uxbox.emails :as emails]
    [uxbox.http.session :as session]
@@ -14,6 +15,7 @@
    [uxbox.services.mutations :as sm]
    [uxbox.services.queries :as sq]
    [uxbox.util.uuid :as uuid]
+   [uxbox.util.transit :as t]
    [uxbox.util.blob :as blob]
    [vertx.http :as vh]
    [vertx.web :as vw]
@@ -31,41 +33,69 @@
 
 (declare ws-websocket)
 (declare ws-send!)
-(declare ws-on-message!)
-(declare ws-on-close!)
 
-;; --- Public API
+;; --- State Management
 
-(declare on-message)
-(declare on-close)
-(declare on-eventbus-message)
+(defonce state
+  (atom {}))
 
-(def state (atom {}))
+(defn send!
+  [ws message]
+  (ws-send! ws (-> (t/encode message)
+                   (t/bytes->str))))
+
+(defmulti handle-message
+  (fn [ws message] (:type message)))
+
+(defmethod handle-message :connect
+  [ws {:keys [file-id user-id] :as message}]
+  (let [local (swap! state assoc-in [file-id user-id] ws)
+        sessions (get local file-id)
+        message {:type :who :users (set (keys sessions))}]
+    (run! #(send! % message) (vals sessions))))
+
+(defmethod handle-message :disconnect
+  [{:keys [user-id] :as ws} {:keys [file-id] :as message}]
+  (swap! state update file-id dissoc user-id)
+  nil)
+
+(defmethod handle-message :who
+  [{:keys [file-id] :as ws} message]
+  (let [users (keys (get @state file-id))]
+    (send! ws {:type :who :users (set users)})))
+
+;; --- Handler
+
+(declare start-eventbus-consumer!)
 
 (defn handler
   [{:keys [user] :as req}]
   (letfn [(on-init [ws]
             (let [vsm (::vw/execution-context req)
-                  tpc "test.foobar"
-                  pid (get-in req [:path-params :page-id])
-                  sem (ve/consumer vsm tpc #(on-eventbus-message ws %2))]
-              (swap! state update pid (fnil conj #{}) user)
-              (assoc ws ::sem sem)))
+                  fid (get-in req [:path-params :file-id])
+                  sem (start-eventbus-consumer! vsm ws fid)]
+
+              (handle-message ws {:type :connect :file-id fid :user-id user})
+              (assoc ws
+                     ::sem sem
+                     :user-id user
+                     :file-id fid)))
 
           (on-message [ws message]
-            (let [pid (get-in req [:path-params :page-id])]
-              (ws-send! ws (str (::counter ws 0)))
-              (update ws ::counter (fnil inc 0))))
+            (try
+              (->> (t/str->bytes message)
+                   (t/decode)
+                   (handle-message ws))
+              (catch Throwable err
+                (log/error "Unexpected exception:\n"
+                           (with-out-str
+                             (.printStackTrace err (java.io.PrintWriter. *out*)))))))
 
           (on-close [ws]
-            (let [pid (get-in req [:path-params :page-id])]
-              (swap! state update pid disj user)
+            (let [fid (get-in req [:path-params :file-id])]
+              (handle-message ws {:type :disconnect :file-id fid})
               (.unregister (::sem ws))))]
 
-    ;; (ws-websocket :on-init on-init
-    ;;               :on-message on-message
-    ;;               :on-close on-close)))
-
     (-> (ws-websocket)
         (assoc :on-init on-init
                :on-message on-message
@@ -73,8 +103,14 @@
 
 (defn- on-eventbus-message
   [ws {:keys [body] :as message}]
+  ;; TODO
   (ws-send! ws body))
 
+(defn- start-eventbus-consumer!
+  [vsm ws fid]
+  (let [topic (str "internal.uxbox.file." fid)]
+    (ve/consumer vsm topic #(on-eventbus-message ws %2))))
+
 ;; --- Internal (vertx api) (experimental)
 
 (defrecord WebSocket [on-init on-message on-close]

backend/src/uxbox/util/blob.clj

@@ -29,38 +29,6 @@
   String
   (->bytes [data] (.getBytes ^String data "UTF-8")))
 
-(defn str->bytes
-  "Convert string to byte array."
-  ([^String s]
-   (str->bytes s "UTF-8"))
-  ([^String s, ^String encoding]
-   (.getBytes s encoding)))
-
-(defn bytes->str
-  "Convert byte array to String."
-  ([^bytes data]
-   (bytes->str data "UTF-8"))
-  ([^bytes data, ^String encoding]
-   (String. data encoding)))
-
-(defn buffer
-  [^bytes data]
-  (Buffer/buffer data))
-
-(defn encode-with-json
-  "A function used for encode data for transfer it to frontend."
-  ([data] (encode-with-json data false))
-  ([data verbose?]
-   (let [type (if verbose? :json-verbose :json)]
-     (-> (t/encode data {:type type})
-         (Buffer/buffer)))))
-
-(defn decode-from-json
-  "A function used for parse data coming from frontend."
-  [data]
-  (-> (->bytes data)
-      (t/decode {:type :json})))
-
 (defn encode
   "A function used for encode data for persist in the database."
   [data]
@@ -73,7 +41,7 @@
       (.writeInt dos (int data-len))
       (.write dos ^bytes cdata (int 0) (alength cdata))
       (-> (.toByteArray baos)
-          (buffer)))))
+          (t/bytes->buffer)))))
 
 (declare decode-v1)
 

backend/src/uxbox/util/transit.clj

@@ -57,20 +57,8 @@
   ([data]
    (decode data nil))
   ([data opts]
-   (cond
-     (instance? Buffer data)
-     (decode (.getBytes ^Buffer data) opts)
-
-     (bytes? data)
-     (with-open [input (ByteArrayInputStream. data)]
-       (read! (reader input opts)))
-
-     (string? data)
-     (decode (.getBytes data "UTF-8") opts)
-
-     :else
-     (with-open [input (io/input-stream data)]
-       (read! (reader input opts))))))
+   (with-open [input (ByteArrayInputStream. ^bytes data)]
+     (read! (reader input opts)))))
 
 (defn encode
   ([data]
@@ -80,3 +68,29 @@
      (let [w (writer out opts)]
        (write! w data)
        (.toByteArray out)))))
+
+;; --- Helpers
+
+(defn str->bytes
+  "Convert string to byte array."
+  ([^String s]
+   (str->bytes s "UTF-8"))
+  ([^String s, ^String encoding]
+   (.getBytes s encoding)))
+
+(defn bytes->str
+  "Convert byte array to String."
+  ([^bytes data]
+   (bytes->str data "UTF-8"))
+  ([^bytes data, ^String encoding]
+   (String. data encoding)))
+
+(defn bytes->buffer
+  [^bytes data]
+  (Buffer/buffer data))
+
+(defn buffer->bytes
+  [^Buffer data]
+  (.getBytes data))
+
+