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

Don't keep reading on EOF

Browse source
This commit is contained in:
Michiel Borkent 2020-05-12 11:42:42 +02:00
parent 630161441b
commit ed996f6f41
1 changed files with 38 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -38,41 +38,44 @@
:json #(cheshire/parse-string-strict % true))]
(try
(loop []
(let [reply (read stdout)
id (get reply "id")
id (bytes->string id)
value* (find reply "value")
value (some-> value*
second
bytes->string
read-fn)
status (get reply "status")
status (set (map (comp keyword bytes->string) status))
done? (contains? status :done)
error? (contains? status :error)
value (if error?
(let [message (or (some-> (get reply "ex-message")
bytes->string)
"")
data (or (some-> (get reply "ex-data")
bytes->string
read-fn)
{})]
(ex-info message data))
value)
chan (get @chans id)
out (some-> (get reply "out")
bytes->string)
err (some-> (get reply "err")
bytes->string)]
(when (or value* error?) (async/put! chan value))
(when (or done? error?) (async/close! chan))
(when out
(binding [*out* out-stream]
(println out)))
(when err (binding [*out* err-stream]
(println err))))
(recur))
(let [reply (try (read stdout)
(catch java.io.EOFException _
::EOF))]
(when-not (identical? ::EOF reply)
(let [id (get reply "id")
id (bytes->string id)
value* (find reply "value")
value (some-> value*
second
bytes->string
read-fn)
status (get reply "status")
status (set (map (comp keyword bytes->string) status))
done? (contains? status :done)
error? (contains? status :error)
value (if error?
(let [message (or (some-> (get reply "ex-message")
bytes->string)
"")
data (or (some-> (get reply "ex-data")
bytes->string
read-fn)
{})]
(ex-info message data))
value)
chan (get @chans id)
out (some-> (get reply "out")
bytes->string)
err (some-> (get reply "err")
bytes->string)]
(when (or value* error?) (async/put! chan value))
(when (or done? error?) (async/close! chan))
(when out
(binding [*out* out-stream]
(println out)))
(when err (binding [*out* err-stream]
(println err))))
(recur))))
(catch Exception e
(binding [*out* err-stream]
(prn e))))))