Skip to content

Commit

Permalink
Unify runtime rpc handling between different platforms
Browse files Browse the repository at this point in the history
  • Loading branch information
@djblue
djblue committed Aug 9, 2024
1 parent 2c5cb8d commit 392360f
Show file tree
Hide file tree
Showing 4 changed files with 58 additions and 84 deletions.
34 changes: 8 additions & 26 deletions src/portal/runtime/clr/server.clj
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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))

Expand All @@ -33,7 +28,7 @@
true
CancellationToken/None)))

(defn- recieve-messgae [^WebSocket ws]
(defn- receive-message [^WebSocket ws]
(let [max-size (* 50 1024 1024)
buffer (byte-array max-size)]
(loop [receive-count 0]
Expand Down Expand Up @@ -67,29 +62,16 @@
(let [^HttpListenerContext context (:context request)
task (.AcceptWebSocketAsync context nil)
_ (.Wait task)
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))
ws (.WebSocket (.Result task))]
(rpc/on-open session #(send-message ws %))
(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)))))))))
(when-let [message (not-empty (receive-message ws))]
(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
Expand Down
38 changes: 7 additions & 31 deletions src/portal/runtime/jvm/server.clj
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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]))

Expand All @@ -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
Expand All @@ -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))
Expand All @@ -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])
Expand Down
32 changes: 5 additions & 27 deletions src/portal/runtime/node/server.cljs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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))]
Expand All @@ -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) #"\?"))
Expand All @@ -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
Expand Down
38 changes: 38 additions & 0 deletions src/portal/runtime/rpc.cljc
View file
Original file line number Diff line number Diff line change
@@ -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))

0 comments on commit 392360f

Please sign in to comment.