std.concurrent.relay

interactive process and socket streams over a shared message bus

Wrap a process or socket as a lifecycle-aware relay with structured read, write, flush, exit, and custom stream operations.

Concurrency overview

1    Overview

A relay pairs input and output streams with a shared bus. It can attach to a running Process or Socket, or create one from process and network options. The transport namespace implements the low-level stream commands.

2    Walkthrough

2.1    Start a process relay

(require '[std.concurrent.relay :as relay])
(require '[std.lib.component :as component])

(def shell
  (relay/relay
   {:type :process
    :args ["cat"]
    :options {:receive {:format identity}
              :send {:format identity}}}))

(relay/send shell {:op :partial :line "hello"})
(relay/send shell {:op :read-line})
(relay/send shell {:op :flush})
(component/stop shell)

2.2    Attach to an existing endpoint

Supply :attached when another part of the application already owns the Process or Socket. relay:create constructs without starting, while relay constructs and starts.

(def attached-relay
  (relay/relay:create
   {:type :socket
    :attached existing-socket
    :options {}}))

(component/start attached-relay)
(relay/send attached-relay {:op :read-some :length 1024})
(component/stop attached-relay)

3    API

*bus* ^

NONE
(defonce ^:dynamic *bus* nil)
link

+control-ops+ ^

NONE
(def +control-ops+
  #{:flush
    :raw
    :partial
    :exit
    :kill})
link

+init+ ^

NONE
(def +init+
  (resource/res:variant-add
   :hara/concurrent.bus
   {:id    :relay
    :alias :hara/relay
    :mode {:allow #{:global} :default :global}
    :instance {:setup    (fn [bus] (component/start bus) (f/set! *bus* bus) bus)
               :teardown (fn [bus] (f/set! *bus* nil) (component/stop bus) bus)}}))
link

+read-ops+ ^

NONE
(def +read-ops+
  #{:count        
    :clean        
    :clean-some   
    :read-all     
    :read-all-bytes
    :read-some    
    :read-some-bytes
    :read-limit   
    :read-line    
    :custom-line  
    :custom})
link

->Relay ^

[type options instance]

NONE
(impl/defimpl Relay [type options instance]           
  :string relay-string
  :prefix "relay-"                               
  :protocols [protocol.component/IComponent
              :exclude [-kill]
              protocol.component/IComponentQuery 
              :include [-started? -stopped?]])
link

->RelayStream ^

[type id raw bus history format]

NONE
(impl/defimpl RelayStream [type id raw bus history format]
  :string relay-stream-string)
link

Relay ^

NONE
(impl/defimpl Relay [type options instance]           
  :string relay-string
  :prefix "relay-"                               
  :protocols [protocol.component/IComponent
              :exclude [-kill]
              protocol.component/IComponentQuery 
              :include [-started? -stopped?]])
link

RelayStream ^

NONE
(impl/defimpl RelayStream [type id raw bus history format]
  :string relay-stream-string)
link

attach-interactive ^

[istream] [{:keys [bus], :as istream} id]
Added 4.0

attaches a bus process to an input stream

v 4.0
(defn attach-interactive
  ([istream]
   (attach-interactive istream nil))
  ([{:keys [bus] :as istream} id]
   (let [thread-handler (fn [{:keys [op] :as message}]
                          (transport/process-op istream op message))
         {:keys [thread]} @(bus/bus:open bus thread-handler {:id id})]
     thread)))
link
(bus/bus:with-temp bus
  (attach-interactive {:raw (java.io.ByteArrayInputStream. (.getBytes "hello"))
                       :bus bus}))
=> Thread

attach-read-passive ^

[{:keys [raw return result final], :as istream} {:keys [op handler], :as message}]
Added 4.0

attaches a passive process to an input stream

v 4.0
(defn attach-read-passive
  ([{:keys [raw return result final] :as istream} {:keys [op handler] :as message}]
   (future/future:run
    (bound-fn []
      (let [return    (env/explode (transport/process-op istream op message))
            _         (if final (future/future:force final return))]
        return)))))
link
@(attach-read-passive
  {:raw (java.io.ByteArrayInputStream. (.getBytes "hello"))}
  {:op :read-all})
=> (contains {:output "hello"})

get-bus ^

[]
Added 3.0

gets the common stream bus

v 3.0
(defn get-bus
  []
  (or *bus*
      (resource/res :hara/relay)))
link
(get-bus)
=> map?

instance-map ^

[id [iraw oraw] options & [m]]

NONE
(defn- instance-map
  [id [iraw oraw] options & [m]]
  (let [istream  (relay-stream id :input  iraw (:receive options))
        ostream  (relay-stream id :output oraw (:send options))]
    (merge {:id  id
            :in  istream
            :out ostream
            :started (time/time-ns)}
           m)))
link

make-instance ^

[obj] [obj id options]
Added 4.0

creates an instance

v 4.0
(defn make-instance
  ([obj]
   (make-instance obj (f/sid) {}))
  ([obj id options]
   (cond (instance? java.net.Socket obj)
         (make-socket-instance obj id options)

         (instance? Process obj)
         (make-process-instance obj id options)

         :else (f/error "Unsupported type: " 
                        {:require [java.net.Socket Process]}))))
link
(make-instance (os/sh "ls" {:wait false}))
=> map?

make-process-instance ^

[process id options]
Added 4.0

creates a process instance

v 4.0
(defn make-process-instance
  ([^Process process id options]
   (let [estream    (relay-stream id :error
                                  (.getErrorStream process)
                                  (merge {:final (future/incomplete)}
                                         (:error options)))
         thread      (attach-read-passive estream (or (-> options :error :custom)
                                                      {:op :custom-line
                                                       :handler (fn [line estream]
                                                                  ;; look at saving to history, displaying, etc.
                                                                  (env/prn line))}))]
     (instance-map id [(.getInputStream process)
                       (.getOutputStream process)]
                   options
                   {:type :process
                    :process process
                    :err (assoc estream :thread thread)}))))
link
(def +instance+ (make-process-instance +process+ "hello"  {}))

+instance+
=> map?

@(:thread (:err +instance+))
=> map?

make-socket-instance ^

[socket id options]
Added 4.0

creates a socket instance

v 4.0
(defn make-socket-instance
  ([^java.net.Socket socket id options]
   (instance-map id [(.getInputStream socket)
                     (.getOutputStream socket)]
                 options
                 {:type :socket
                  :socket socket})))
link
(def +instance+ (make-socket-instance +socket+
                                      "hello"
                                      {}))
(set (keys +instance+))
=> #{:started :socket :type :out :id :in}

(do (.close ^ServerSocket +server+)
    (.close ^Socket +socket+))

map->Relay ^

[m__7997__auto__]

NONE
(impl/defimpl Relay [type options instance]           
  :string relay-string
  :prefix "relay-"                               
  :protocols [protocol.component/IComponent
              :exclude [-kill]
              protocol.component/IComponentQuery 
              :include [-started? -stopped?]])
link

map->RelayStream ^

[m__7997__auto__]

NONE
(impl/defimpl RelayStream [type id raw bus history format]
  :string relay-stream-string)
link

relay ^

[{:keys [type], :as m}]
Added 3.0

creates and starts a relay

v 3.0
(defn relay
  ([{:keys [type] :as m}]
   (-> (relay:create m)
       (component/start))))
link
(relay {:type :process :args ["bash"]})
=> map?

relay-start ^

[{:keys [id type instance options hooks attached], :as relay}]
Added 4.0

starts the relay

v 4.0
(defn relay-start
  ([{:keys [id type instance options hooks attached] :as relay}]
   (if (relay-started? relay)
     relay
     (let [obj (or attached
                   (case type
                     :socket  (network/socket (:host relay) (:port relay))
                     :process (os/sh (merge (select-keys relay [:root :env :args])
                                           {:wait false
                                            :inherit false
                                            :wrap false
                                            :output false}))))
           instance   (make-instance obj id options)
           _          (if-let [custom (get-in options [:receive :custom])]
                        (attach-read-passive (:in instance) (assoc custom
                                                                   :instance instance))
                        (attach-interactive (:in instance)
                                            (-> instance :in :id)))
           _     (reset! (:instance relay) instance)
           {:keys [setup]} hooks
           _   (if setup (setup relay))]
       relay))))
link
(def +relay+ (doto (relay:create {:type :process
                                  :args ["bash"]})
               (relay-start)))

(cc/bus:has-id? (get-bus) (:id +relay+))
=> true

(do (def +process+ (:process @(:instance +relay+)))
    (.isAlive ^Process +process+))
=> true

(:output @(send +relay+ "echo hello"))
=> "hellon"

(do (send +relay+ {:op :partial
                    :line "echo hello"})
    (send +relay+ {:op :partial
                   :line "echo hello"})

    (:count   @(send +relay+ {:op :count})))
=> 12

(:output  @(send +relay+ {:op :read-limit
                          :limit 8}))
=> "hellonhe"

(:dropped @(send +relay+ {:op :clean}))
=> 4

(component/stop +relay+)
=> map?

relay-started? ^

[{:keys [instance]}]

NONE
(defn- relay-started?
  ([{:keys [instance]}]
   (let [{:keys [type in ^Process process]} @instance
         {:keys [bus id]} in]
     (and bus
          (or (bus/bus:has-id? bus id)
              (and (= type :process)
                   (.isAlive process)))))))
link

relay-stop ^

[{:keys [type hooks instance], :as relay}]
Added 4.0

stops the relay

v 4.0
(defn relay-stop
  [{:keys [type hooks instance] :as relay}]
  (if (relay-stopped? relay)
    relay
    (let [{:keys [in type]} @instance
          {:keys [bus id]} in
          _  (and bus (env/explode (bus/bus:kill bus id)))
          _  (component/stop (get @instance type))
          _  (reset! instance nil)
          {:keys [teardown]} hooks
           _   (if teardown (teardown relay))]
      relay)))
link
(let [relay (doto (relay:create {:type :process
                                 :args ["bash"]})
              (relay-start))]
  (relay-stop relay)
  (component/stopped? relay))
=> true

relay-stopped? ^

[relay]

NONE
(defn- relay-stopped?
  ([relay]
   (not (relay-started? relay))))
link

relay-stream ^

[id type raw options]
Added 4.0

creates a relay stream

v 4.0
(defn relay-stream
  [id type raw options]
  (let [bus      (get-bus)]
    (map->RelayStream
     (merge {:type type
             :id id
             :raw raw
             :bus bus
             :history (q/queue)
             :format  identity}
            options))))
link
(relay-stream "hello"
              :input
              nil
              {})
=> relay-stream?

relay-stream-string ^

[{:keys [type id raw bus history format]}]

NONE
(defn- relay-stream-string
  [{:keys [type id raw bus history format]}]
  (str "#relay.stream" [type id (count history)]))
link

relay-stream? ^

[obj]
Added 4.0

checks if object is a relay stream

v 4.0
(defn relay-stream?
  [obj]
  (instance? RelayStream obj))
link
(relay-stream? (relay-stream "hello" :input nil {}))
=> true

relay-string ^

[{:keys [type options instance]}]

NONE
(defn- relay-string [{:keys [type options instance]}]
  (str "#relay" [type (or options {}) @instance]))
link

relay:create ^

[{:keys [id type], :as m}]
Added 4.0

creates a relay

v 4.0
(defn relay:create
  ([{:keys [id type] :as m}]
   (map->Relay (assoc m
                      :id (or id (f/sid))
                      :instance (atom nil)))))
link
(relay:create {:type :process :args ["bash"]})
=> map?

send ^

[{:keys [instance], :as relay} msg]
Added 3.0

sends command to relay

v 3.0
(defn send
  ([{:keys [instance] :as relay} msg]
   (let [{:keys [in out]}  (or @instance
                               (f/error "Relay not started"
                                        {:relay relay :msg msg}))
         {:keys [op line]
          :as msg}    (if (string? msg)
                        {:op :read-some
                         :line msg}
                        msg)
         op  (or (transport/+read-alias+ op)
                 (+read-ops+ op)
                 (+control-ops+ op)
                 (f/error "Op not valid." {:input op
                                           :available (set (concat +control-ops+
                                                                   +read-ops+))
                                           :alias transport/+read-alias+}))]
     (case op
       :kill     (relay-stop relay)
       :exit     (do (.close ^OutputStream (:raw out))
                     (bus/bus:close (:bus in) (:id in)))
       :flush    (transport/send-write-flush out)
       :raw      (transport/send-write-raw out line)
       :partial  (transport/send-write-line out line)
       (transport/send-command in out op line msg)))))
link
(let [r (relay {:type :process :args ["bash"]})]
  (try
    (-> @(send r "echo hello") :output)
    => "hellon"
    (finally
      (relay-stop r))))

with:bus ^

[[bus] & body]
Added 4.0

sets the default relay bus

v 4.0
(defmacro with:bus
  [[bus] & body]
  `(binding [*bus* bus]
     ~@body))
link
(macroexpand-1 '(with:bus [bus]
                  *bus*))
=> '(clojure.core/binding [std.concurrent.relay/*bus* std.concurrent.relay/bus]
      *bus*)

+read-alias+ ^

NONE
(def +read-alias+)
link

bytes-output ^

[]
Added 4.0

creates a byte output stream

v 4.0
(defn bytes-output
  (^ByteArrayOutputStream []
   (ByteArrayOutputStream.)))
link
(t/bytes-output)
=> java.io.ByteArrayOutputStream

mock-input-stream ^

[out]
Added 4.0

creates a mock input stream

v 4.0
(defn mock-input-stream
  [out]
  (proxy [java.io.InputStream] []
    (available [] (count (first @out)))
    (readAllBytes []
      (atom/swap-return! out
        (fn [arr]
          [(.getBytes (clojure.string/join arr)) []])))
    (readNBytes [n]
      (atom/swap-return! out
        (fn [arr]
          (if (pos? (- (count (first arr)) n))
            [(.getBytes ^String (subs (first arr) 0 n))
             (vec (cons (subs (first arr) n)
                        (rest arr)))]
            [(.getBytes ^String (first arr)) (vec (rest arr))]))))))
link
(let [stream (t/mock-input-stream (atom ["abc" "de"]))]
  [(String. (.readNBytes stream 2))
   (String. (.readAllBytes stream))])
=> ["ab" "cde"]

op-clean ^

[{:keys [raw], :as istream}]
Added 4.0

cleans the current input stream

v 4.0
(defn op-clean
  ([{:keys [^InputStream raw] :as istream}]
   (let [n (.available raw)
         _ (.readNBytes raw n)
         t (time/time-ns)]
     {:start t
      :end t
      :dropped n})))
link
(do (def +input+ (atom ["OK " "Hello "]))

    (-> (t/op-clean {:raw (t/mock-input-stream
                           +input+)})
        :dropped))
=> 3

@+input+
=> ["Hello "]

op-clean-some ^

[istream timeout interval]
Added 4.0

cleans until timeout

v 4.0
(defn op-clean-some
  ([istream timeout interval]
   (let [counter (volatile! 0)
         count-fn (fn [^bytes bytes]
                    (vswap! counter + (count bytes)))
         stats (read-bytes-some istream count-fn timeout interval)]
     (assoc stats :dropped @counter))))
link
(do (def +input+ (atom ["OK " "Hello "]))

    (-> (t/op-clean-some {:raw (t/mock-input-stream
                                +input+)}
                         10
                         01)
        :dropped))
=> number?

@+input+
=> []

op-count ^

[{:keys [raw], :as istream}]
Added 4.0

outputs the count in the stream

v 4.0
(defn op-count
  ([{:keys [^InputStream raw] :as istream}]
   (let [t (time/time-ns)]
     {:start t
      :end t
      :count (.available raw)})))
link
(-> (t/op-count {:raw (t/mock-input-stream
                       (atom ["OK "
                              "OK "]))})
    :count)
=> 3

op-read-all ^

[istream]
Added 4.0

reads all bytes from stream as string

v 4.0
(defn op-read-all
  ([istream]
   (update (op-read-all-bytes istream)
           :output f/string)))
link
(-> (t/op-read-all {:raw (t/mock-input-stream (atom ["OK " "Hello " "World "])
                                              )})
    :output)
=> "OK Hello World "

op-read-all-bytes ^

[{:keys [raw], :as istream}]
Added 4.0

reads all bytes from the stream

v 4.0
(defn op-read-all-bytes
  ([{:keys [^InputStream raw] :as istream}]
   (let [start (time/time-ns)
         bytes (.readAllBytes raw)
         end   (time/time-ns)]
     {:output bytes :start start :end end})))
link
(-> (t/op-read-all-bytes {:raw (t/mock-input-stream (atom ["OK " "Hello " "World "])
                                                    )})
    :output
    f/string)
=> "OK Hello World "

op-read-limit ^

[istream limit timeout interval]
Added 4.0

reads an limited amount of characters from stream or until timeout

v 4.0
(defn op-read-limit
  ([istream limit timeout interval]
   (let [out  (volatile! nil)
         save-fn (fn [s] (vreset! out s))
         stats (read-bytes-limit istream save-fn save-fn limit timeout interval)]
     (assoc stats :output @out))))
link
(do (def +state+  (volatile! []))
    (def +input+ (atom ["OK " "Hello "]))
    (def +p+ (future/future (Thread/sleep 50)
                       (swap! +input+ conj "Worldn")
                       (Thread/sleep 50)
                       (swap! +input+ conj "Again ")))
    (-> (t/op-read-limit {:raw (t/mock-input-stream +input+)}
                         18
                         200
                         20)
        :output))
=> "OK Hello WorldnAga"
@+p+

op-read-line ^

[istream timeout interval]
Added 4.0

read line from stream or until timeout

v 4.0
(defn op-read-line
  ([istream timeout interval]
   (let [out  (volatile! nil)
         save-fn (fn [s] (vreset! out s))
         stats (read-bytes-line istream save-fn save-fn timeout interval)]
     (assoc stats :output @out))))
link
(do (def +state+  (volatile! []))
    (def +input+ (atom ["OK " "Hello "]))
    (def +p+ (future/future (Thread/sleep 50)
                       (swap! +input+ conj "Worldn")
                       (Thread/sleep 50)
                      (swap! +input+ conj "Again ")))
    (-> (t/op-read-line {:raw (t/mock-input-stream +input+)}
                        80
                        20)
        :output))
=> "OK Hello Worldn"
@+p+

op-read-some ^

[istream timeout interval]
Added 4.0

read bytes from stream until timeout as string

v 4.0
(defn op-read-some
  ([istream timeout interval]
   (update (op-read-some-bytes istream timeout interval)
           :output f/string)))
link
(do (def +state+  (volatile! []))
    (def +input+ (atom ["OK " "Hello "]))
    (def +p+ (future/future (Thread/sleep 50)
                       (swap! +input+ conj "World ")
                       (Thread/sleep 50)
                       (swap! +input+ conj "Again ")))
    (-> (t/op-read-some {:raw (t/mock-input-stream +input+)}
                        80
                        20)
        :output))
=> "OK Hello World Again "
@+p+

op-read-some-bytes ^

[istream timeout interval]
Added 4.0

read bytes from stream until timeout

v 4.0
(defn op-read-some-bytes
  ([istream timeout interval]
   (let [baos (ByteArrayOutputStream.)
         save-fn (fn [^bytes bytes] (.writeBytes baos bytes))
         stats (read-bytes-some istream save-fn timeout interval)]
     (assoc stats :output (.toByteArray baos)))))
link
(do (def +state+  (volatile! []))
    (def +input+ (atom ["OK " "Hello "]))
    (def +p+ (future/future (Thread/sleep 50)
                       (swap! +input+ conj "World ")
                       (Thread/sleep 50)
                      (swap! +input+ conj "Again ")))
    (-> (t/op-read-some-bytes {:raw (t/mock-input-stream +input+)}
                              80
                              20)
        :output
        f/string))
=> "OK Hello World Again "
@+p+

process-by-handler ^

[{:keys [raw], :as istream} handler msg]
Added 4.0

process the input with a function

v 4.0
(defn process-by-handler
  ([{:keys [raw] :as istream} handler msg]
   (let [start (time/time-ns)]
     (try
       (handler raw msg)
       (catch java.io.IOException e))
     {:start start :end (time/time-ns)})))
link
(def +state+ (volatile! []))
(t/process-by-handler {:raw (java.io.ByteArrayInputStream.
                             (.getBytes "Hellon Worldn"))}
                      (fn [^java.io.InputStream is _]
                        (vreset! +state+ (f/string (.readAllBytes is))))
                      {})

@+state+
=> "Hellon Worldn"

process-by-line ^

[{:keys [raw], :as istream} line-fn msg]
Added 4.0

process each line using a function

v 4.0
(defn process-by-line
  ([{:keys [raw] :as istream} line-fn msg]
   (let [start (time/time-ns)]
     (with-open [reader (BufferedReader. (InputStreamReader. raw))]
       (try
         (loop [line (.readLine reader)]
           (when line
             (line-fn line msg)
             (recur (.readLine reader))))
         (catch java.io.IOException e)))
     {:start start :end (time/time-ns)})))
link
(def +state+ (volatile! []))
(t/process-by-line {:raw (java.io.ByteArrayInputStream.
                          (.getBytes "Hellon Worldn"))}
                   (fn [line _]
                     (vswap! +state+ conj line))
                   {})
@+state+
=> ["Hello" " World"]

process-op ^

[istream op {:keys [timeout interval limit handler], :as msg}]
Added 4.0

processes an op given a command

v 4.0
(defn process-op
  ([istream op {:keys [timeout interval limit handler] :as msg}]
   (let [timeout  (or timeout  (:timeout  istream) 120)
         interval (or interval (:interval istream) 15)
         op    (or (+read-alias+ op)
                   op)
         status   (case op
                    :count        (op-count istream)
                    :clean        (op-clean istream)
                    :clean-some   (op-clean-some timeout interval)
                    :read-all     (op-read-all istream)
                    :read-all-bytes  (op-read-all-bytes istream)
                    :read-some    (op-read-some istream timeout interval)
                    :read-some-bytes (op-read-some-bytes istream timeout interval)
                    :read-limit   (op-read-limit istream (or limit 0) timeout interval)
                    :read-line    (op-read-line  istream timeout interval)
                    :custom-line  (process-by-line istream handler msg)
                    :custom       (process-by-handler istream handler msg))]
     status)))
link
(-> (t/process-op {:raw (java.io.ByteArrayInputStream.
                         (.getBytes "Hellon Worldn"))}
                  :count
                  {})
    :count)
=> 13

(-> (t/process-op {:raw (java.io.ByteArrayInputStream.
                         (.getBytes "Hellon Worldn"))}
                  :clean
                  {})
    :dropped)
=> 13

read-bytes-limit ^

[{:keys [raw]} f f-timeout limit timeout interval]
Added 4.0

reads a limited number of bytes from stream

v 4.0
(defn read-bytes-limit
  ([{:keys [^InputStream raw]} f f-timeout limit timeout interval]
   (let [start   (time/time-ms)
         output  (loop [line  ""
                        limit limit]
                   (cond (zero? limit)
                         (f line)
                         
                         (< (- (time/time-ms) start)
                            timeout)
                         (let [n (.available raw)]
                           (if (zero? n)
                             (do (Thread/sleep ^long interval)
                                 (recur line limit))
                             (let [needed  (min n limit)
                                   s (String. (.readNBytes raw needed))]
                               (recur (str line s) (- limit needed)))))

                         :else
                         (f-timeout line)))]
     {:start start
      :end (time/time-ns)})))
link
(dotimes [i 2]
  (t/read-bytes-limit {:raw +stream+}
                     (fn [v]
                       (vswap! +state+ conj (f/string v)))
                     identity
                     2
                     10
                     10))
@+state+
=> ["OK" " H"]

read-bytes-line ^

[{:keys [raw]} f f-timeout timeout interval]
Added 4.0

reads individual lines from stream

v 4.0
(defn read-bytes-line
  ([{:keys [^InputStream raw]} f f-timeout timeout interval]
   (let [start   (time/time-ms)
         output  (loop [line ""]
                   (cond (< (- (time/time-ms) start)
                            timeout)
                         (let [n (.available raw)]
                           (if (zero? n)
                             (do (Thread/sleep ^long interval)
                                 (recur line))
                             (let [s (String. (.readNBytes raw n))]
                               (if (clojure.string/ends-with? s "n")
                                 (f (str line s))
                                 (recur (str line s))))))

                         :else
                         (f-timeout line)))]
     {:start start
      :end (time/time-ns)})))
link
(dotimes [i 2]
  (t/read-bytes-line {:raw +stream+}
                     (fn [v]
                       (vswap! +state+ conj (f/string v)))
                     identity
                     10
                     10))

@+state+
=> ["OK Hellon" "Worldn"]

read-bytes-some ^

[{:keys [raw]} f timeout interval]
Added 4.0

reads until timeout

v 4.0
(defn read-bytes-some
  ([{:keys [^InputStream raw]} f timeout interval]
   (let [start   (time/time-ns)
         counter (f/counter 0)
         max     (inc (quot timeout interval))]
     (loop []
       (if (< @counter max)
         (let [n (.available raw)]
           (if (zero? n)
             (do (f/inc! counter)
                 (Thread/sleep ^long interval)
                 (recur))
             (do (f/reset! counter 0)
                 (f (.readNBytes raw n))
                 (recur))))))
     {:checks @counter
      :start start
      :end (time/time-ns)})))
link
(t/read-bytes-some {:raw +stream+}
                   (fn [v]
                     (vswap! +state+ conj (f/string v)))
                   10
                   10)
@+state+
=> ["OK "
    "Hellon"
    "Worldn"]

(do (def +state+  (volatile! []))
    (def +input+ (atom ["OK " "Hello "]))
    (def +p+ (future/future (Thread/sleep 500)
                       (swap! +input+ conj "World ")))
    (t/read-bytes-some {:raw (t/mock-input-stream +input+)}
                       (fn [v]
                         (vswap! +state+ conj (f/string v)))
                       10
                       10)
    (Thread/sleep 200)
    @+state+)
=> ["OK " "Hello "]
@+p+

(do (def +state+  (volatile! []))
    (def +input+ (atom ["OK " "Hello "]))
    (def +p+ (future/future (Thread/sleep 50)
                       (swap! +input+ conj "World ")
                       (Thread/sleep 50)
                       (swap! +input+ conj "Again ")))
    (t/read-bytes-some {:raw (t/mock-input-stream +input+)}
                       (fn [v]
                         (vswap! +state+ conj (f/string v)))
                       80
                       20)
    @+state+)
=> ["OK " "Hello " "World " "Again "]
@+p+

send-command ^

[istream ostream op line {:keys [direct], :as msg}]
Added 4.0

sends a command to output and then waits on the input

v 4.0
(defn send-command
  ([istream
    ostream
    op
    ^String line
    {:keys [direct] :as msg}]
   (do (if line
         (send-write-line ostream line))
       (if (and (:bus istream)
                (not (false? direct)))
         (bus/bus:send (:bus istream) (:id istream) (dissoc msg :line))
         (process-op istream op (dissoc msg :line))))))
link
(let [in  {:raw (java.io.ByteArrayInputStream. (.getBytes "hello"))}
      out {:raw (java.io.ByteArrayOutputStream.)}]
  [(t/send-command in out :read-all "PING" {:direct false})
   (.toString ^java.io.ByteArrayOutputStream (:raw out))])
=> (contains [(contains {:output "hello"}) "PINGn"])

send-write-flush ^

[{:keys [raw], :as ostream}]
Added 4.0

sends a raw flush command

v 4.0
(defn send-write-flush
  ([{:keys [raw] :as ostream}]
   (doto ^OutputStream raw
     (.flush))))
link
(let [out (java.io.ByteArrayOutputStream.)]
  (t/send-write-flush {:raw out}))
=> java.io.ByteArrayOutputStream

send-write-line ^

[{:keys [raw], :as ostream} line]
Added 4.0

sends a command with a newline and flush

v 4.0
(defn send-write-line
  ([{:keys [raw] :as ostream} ^String line]
   (doto ^OutputStream raw
     (.write (.getBytes line))
     (.write (.getBytes "n"))
     (.flush))))
link
(let [out (java.io.ByteArrayOutputStream.)]
  (t/send-write-line {:raw out} "hello")
  (.toString out))
=> "hellon"

send-write-raw ^

[{:keys [raw], :as ostream} bytes]
Added 4.0

sends a raw write command

v 4.0
(defn send-write-raw
  ([{:keys [raw] :as ostream} ^bytes bytes]
   (doto ^OutputStream raw
     (.write bytes))))
link
(let [out (java.io.ByteArrayOutputStream.)]
  (t/send-write-raw {:raw out} (.getBytes "hello"))
  (.toString out))
=> "hello"

Related

jvm.artifactdocs

artifact coordinates and paths

jvm.classloaderdocs

classpath and classloaders

jvm.depsdocs

JVM dependency resolution

jvm.monitordocs

JVM metrics

jvm.namespacedocs

namespace management

jvm.protocoldocs

JVM extension protocols

jvm.reflectdocs

Java reflection

jvm.requiredocs

dependency-aware requiring