Unify runtime rpc handling between different platforms

Author: djblue

src/portal/runtime/clr/server.clj

@@ -2,10 +2,10 @@
   (:require [clojure.edn :as edn]
             [clojure.string :as str]
             [portal.runtime :as rt]
-            [portal.runtime.clr.client :as c]
             [portal.runtime.fs :as fs]
             [portal.runtime.index :as index]
-            [portal.runtime.json :as json])
+            [portal.runtime.json :as json]
+            [portal.runtime.rpc :as rpc])
   (:import (clojure.lang RT)
            (System Environment Guid)
            (System.IO Path)
@@ -16,11 +16,6 @@
 
 (defmulti route (juxt :request-method :uri))
 
-(def ^:private ops (merge c/ops rt/ops))
-
-(defn- not-found [_request done]
-  (done {:status :not-found}))
-
 (defmacro array-segment [& args]
   `(new ~(RT/classForName "System.ArraySegment`1[System.Byte]") ~@args))
 
@@ -70,26 +65,15 @@
               ws    (.WebSocket (.Result task))
               send! (fn [message]
                       (send-message ws (rt/write message session)))]
-          (swap! rt/connections assoc (:session-id session) send!)
-          (when-let [f (get-in session [:options :on-load])]
-            (f))
+          (rpc/on-open session send!)
           (while (= (.State ws) WebSocketState/Open)
             (when-let [message (not-empty (recieve-messgae ws))]
-              (let [body     (rt/read message session)
-                    id       (:portal.rpc/id body)
-                    op       (get ops (:op body) not-found)]
-                (binding [rt/*session* session]
-                  (op body (fn [response]
-                             (send!
-                              (assoc response
-                                     :portal.rpc/id id
-                                     :op :portal.rpc/response)))))))))
+              (rpc/on-receive session message))))
         (catch Exception e
           (tap> (Throwable->map e)))
         (finally
           (close-debug debug)
-          (rt/reset-session session)
-          (swap! rt/connections dissoc (:session-id session)))))))
+          (rpc/on-close session))))))
 
 (defn- send-resource [content-type resource]
   {:status  200

src/portal/runtime/jvm/server.clj

@@ -9,9 +9,9 @@
             [portal.runtime.fs :as fs]
             [portal.runtime.index :as index]
             [portal.runtime.json :as json]
-            [portal.runtime.jvm.client :as c]
             [portal.runtime.npm :as npm]
-            [portal.runtime.remote.socket :as socket])
+            [portal.runtime.remote.socket :as socket]
+            [portal.runtime.rpc :as rpc])
   (:import [java.io File PushbackReader]
            [java.util UUID]))
 
@@ -25,9 +25,6 @@
 
 (defmulti route (juxt :request-method :uri))
 
-(defn- not-found [_request done]
-  (done {:status :not-found}))
-
 (defn- rpc-handler-remote [request]
   (let [conn (socket/open (:session request))]
     (server/as-channel
@@ -45,8 +42,6 @@
       (fn [_ch _status]
         (socket/close conn))})))
 
-(def ^:private ops (merge c/ops rt/ops))
-
 (defn- open-debug [{:keys [options] :as session}]
   (try
     (when (= :server (:debug options))
@@ -61,33 +56,14 @@
 
 (defn- rpc-handler-local [request]
   (let [session (rt/open-session (:session request))
-        send!   (fn send! [ch message]
-                  (server/send! ch (rt/write message session)))
         debug   (open-debug session)]
     (server/as-channel
      request
-     {:on-receive
-      (fn [ch message]
-        (let [body  (rt/read message session)
-              id    (:portal.rpc/id body)
-              op    (get ops (:op body) not-found)]
-          (binding [rt/*session* session]
-            (op body (fn [response]
-                       (send!
-                        ch
-                        (assoc response
-                               :portal.rpc/id id
-                               :op :portal.rpc/response)))))))
-      :on-open
-      (fn [ch]
-        (swap! rt/connections assoc (:session-id session) (partial send! ch))
-        (when-let [f (get-in session [:options :on-load])]
-          (f)))
-      :on-close
-      (fn [_ch _status]
-        (close-debug debug)
-        (rt/reset-session session)
-        (swap! rt/connections dissoc (:session-id session)))})))
+     {:on-receive (fn [_ch message] (rpc/on-receive session message))
+      :on-open    (fn [ch] (rpc/on-open session #(server/send! ch %)))
+      :on-close   (fn [_ch _status]
+                    (close-debug debug)
+                    (rpc/on-close session))})))
 
 (defmethod route [:get "/rpc"] [request]
   (if (get-in request [:session :options :runtime])

src/portal/runtime/node/server.cljs

@@ -7,15 +7,12 @@
             [portal.runtime.fs :as fs]
             [portal.runtime.index :as index]
             [portal.runtime.json :as json]
-            [portal.runtime.node.client :as c]
+            [portal.runtime.rpc :as rpc]
             [portal.runtime.transit :as transit]))
 
 (defn- get-header [^js req k]
   (-> req .-headers (aget k)))
 
-(defn- not-found [_request done]
-  (done {:status :not-found}))
-
 (defn- require-string [src file-name]
   (let [Module (js/require "module")
         ^js m (Module. file-name (some-> js/module .-parent))]
@@ -27,8 +24,6 @@
 
 (def Server (.-Server (require-string ws-code "portal/ws.js")))
 
-(def ops (merge c/ops rt/ops))
-
 (defn- get-session-id [^js req]
   (some->
    (or (second (str/split (.-url req) #"\?"))
@@ -54,27 +49,10 @@
      (.-socket req)
      (.-headers req)
      (fn [^js ws]
-       (let [session (rt/open-session session)
-             send!
-             (fn send! [message]
-               (.send ws (rt/write message session)))]
-         (swap! rt/connections assoc (:session-id session) send!)
-         (when-let [f (get-in session [:options :on-load])]
-           (f))
-         (.on ws "message"
-              (fn [message]
-                (a/let [req  (rt/read message session)
-                        id   (:portal.rpc/id req)
-                        op   (get ops (get req :op) not-found)
-                        done #(send! (assoc %
-                                            :portal.rpc/id id
-                                            :op :portal.rpc/response))]
-                  (binding [rt/*session* session]
-                    (op req done)))))
-         (.on ws "close"
-              (fn []
-                (rt/reset-session session)
-                (swap! rt/connections dissoc (:session-id session)))))))))
+       (let [session (rt/open-session session)]
+         (rpc/on-open session #(.send ws %))
+         (.on ws "message" (fn [message] (rpc/on-receive session message)))
+         (.on ws "close"   (fn [] (rpc/on-close session))))))))
 
 (defn- send-resource [^js res content-type body]
   (-> res

src/portal/runtime/rpc.cljc

@@ -0,0 +1,38 @@
+(ns ^:no-doc portal.runtime.rpc
+  (:require #?(:clj  [portal.runtime.jvm.client :as c]
+               :cljs [portal.runtime.node.client :as c]
+               :cljr [portal.runtime.clr.client :as c])
+            [portal.runtime :as rt]))
+
+(defn on-open [session send!]
+  (swap! rt/connections
+         assoc (:session-id session)
+         (fn [message]
+           (send! (rt/write message session))))
+  (when-let [f (get-in session [:options :on-load])]
+    (try
+      (f)
+      (catch #?(:cljs :default :default Exception) e
+        (tap> e)))))
+
+(def ^:private ops (merge c/ops rt/ops))
+
+(defn- not-found [_request done]
+  (done {:status :not-found}))
+
+(defn on-receive [session message]
+  (let [send! (get @rt/connections (:session-id session))
+        body  (rt/read message session)
+        id    (:portal.rpc/id body)
+        op    (get ops (:op body) not-found)
+        done  (fn on-done [response]
+                (send!
+                 (assoc response
+                        :portal.rpc/id id
+                        :op :portal.rpc/response)))]
+    (binding [rt/*session* session]
+      (op body done))))
+
+(defn on-close [session]
+  (swap! rt/connections dissoc (:session-id session))
+  (rt/reset-session session))