luciano/pods
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

[#2] Experimental socket support

Browse source
This commit is contained in:
Michiel Borkent 2020-10-13 22:34:01 +02:00
parent 7f804baca8
commit 225f9239d8
3 changed files with 49 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

44
src/babashka/pods/impl.clj
View file

@ -219,26 +219,38 @@
[^Socket socket]
(.close socket))
(defn gobbler [^java.io.InputStream is]
(future
(loop []
(let [v (.read is)]
(when-not (= -1 v)
(print (char v))
(recur))))))
(defn read-port [pid]
1888 #_(loop []
(let [f (io/file (str ".babashka/pods/" pid ".port"))]
(if (.exists f)
(edn/read-string (slurp f))
(recur)))))
(defn load-pod
([pod-spec] (load-pod pod-spec nil))
([pod-spec {:keys [:remove-ns :resolve]}]
([pod-spec {:keys [:remove-ns :resolve :socket :inherit-io]}]
(let [pod-spec (if (string? pod-spec) [pod-spec] pod-spec)
pb (ProcessBuilder. ^java.util.List pod-spec)
_ (.redirectError pb java.lang.ProcessBuilder$Redirect/INHERIT)
_ (if inherit-io
(.inheritIO pb)
(.redirectError pb java.lang.ProcessBuilder$Redirect/INHERIT))
_ (doto (.environment pb)
(.put "BABASHKA_POD" "true"))
p (.start pb)
stdin (.getOutputStream p)
stdout (.getInputStream p)
stdout (java.io.PushbackInputStream. stdout)
pid (.pid p)
socket-port (when socket (read-port pid))
[stdin stdout]
(if socket
(let [^Socket socket
(loop []
(if-let [sock (try (create-socket "localhost" socket-port)
(catch java.net.ConnectException _
nil))]
sock
(recur)))]
[(.getOutputStream socket)
(PushbackInputStream. (.getInputStream socket))])
[(.getOutputStream p) (java.io.PushbackInputStream. (.getInputStream p))])
_ (write stdin {"op" "describe"
"id" (next-id)})
reply (read stdout)
@ -246,14 +258,6 @@
ops (some->> (get reply "ops") keys (map keyword) set)
readers (when (identical? :edn format)
(read-readers reply resolve))
socket-port (get reply "port")
[stdin stdout _gobbler]
(if socket-port
(let [socket (create-socket "localhost" socket-port)]
[(.getOutputStream socket)
(PushbackInputStream. (.getInputStream socket))
(gobbler stdout)])
[stdin stdout])
pod {:process p
:pod-spec pod-spec
:stdin stdin

15
src/babashka/pods/jvm.clj
View file

@ -33,15 +33,16 @@
(defn load-pod
([pod-spec] (load-pod pod-spec nil))
([pod-spec _opts]
([pod-spec opts]
(let [pod (impl/load-pod
pod-spec
{:remove-ns remove-ns
:resolve (fn [sym]
(or (resolve sym)
(intern
(create-ns (symbol (namespace sym)))
(symbol (name sym)))))})
(merge {:remove-ns remove-ns
:resolve (fn [sym]
(or (resolve sym)
(intern
(create-ns (symbol (namespace sym)))
(symbol (name sym)))))}
opts))
namespaces (:namespaces pod)]
(swap! namespaces-to-load
merge

32
src/babashka/pods/sci.clj
View file

@ -23,26 +23,28 @@
(with-meta
(fn
([ctx pod-spec] (load-pod ctx pod-spec nil))
([ctx pod-spec _opts]
([ctx pod-spec opts]
(let [env (:env ctx)
pod (binding [*out* @sci/out
*err* @sci/err]
(impl/load-pod
pod-spec
{:remove-ns
(fn [sym]
(swap! env update :namespaces dissoc sym))
:resolve
(fn [sym]
(let [sym-ns (or (some-> (namespace sym)
symbol)
'clojure.core)
sym-name (symbol (name sym))]
(or (get-in @env [:namespaces sym-ns sym-name])
(let [v (sci/new-var sym {:predefined true})]
(swap! env assoc-in [:namespaces sym-ns sym-name]
v)
v))))}))
(merge
{:remove-ns
(fn [sym]
(swap! env update :namespaces dissoc sym))
:resolve
(fn [sym]
(let [sym-ns (or (some-> (namespace sym)
symbol)
'clojure.core)
sym-name (symbol (name sym))]
(or (get-in @env [:namespaces sym-ns sym-name])
(let [v (sci/new-var sym {:predefined true})]
(swap! env assoc-in [:namespaces sym-ns sym-name]
v)
v))))}
opts)))
namespaces (:namespaces pod)
namespaces-to-load (set (keep (fn [[ns-name _ defer?]]
(when defer?