std.concurrent.bus

request and response messaging between registered threads

A bus owns per-thread queues, dispatches messages to handlers, and completes futures when responses return.

Concurrency overview

1    Overview

A bus combines thread registration, blocking queues, handler loops, and future-backed responses. Use it when work must be addressed to a specific long-running thread rather than submitted to an anonymous executor task.

2    Walkthrough

2.1    Open a handler

bus:with-temp starts a temporary bus and registers the current thread. bus:open then starts a named handler loop and returns its registration details.

(require '[std.concurrent.bus :as bus])

(bus/bus:with-temp message-bus
  (let [{:keys [id stopped]}
        @(bus/bus:open message-bus
                       (fn [{:keys [value]}]
                         {:value (inc value)}))]
    {:reply @(bus/bus:send message-bus id {:value 10})
     :info  (bus/info-bus message-bus)
     :stop  (bus/bus:close message-bus id)
     :stopped stopped}))

2.2    Inspect and control workers

Use the registry functions to inspect IDs and queues. Prefer bus:close for cooperative shutdown; use bus:kill only when a handler thread must be interrupted.

(bus/bus:all-ids message-bus)
(bus/bus:get-count message-bus)
(bus/bus:send-all message-bus {:op :refresh})
(bus/bus:close-all message-bus)

3    Lifecycle

bus creates and starts a component. bus:create only constructs it. Component-aware code can use start-bus, stop-bus, started?-bus, and info-bus directly.

4    API

+keys+ ^

NONE
(def +keys+)
link

+resource+ ^

NONE
(def +resource+
  (res/res:spec-add
   {:type :hara/concurrent.bus
    :instance {:create bus:create}}))
link

->Bus ^

[state results output counters]

NONE
(impl/defimpl Bus [state results output counters]
  :suffix "-bus"
  :string string-bus
  :protocols [std.protocol.component/IComponent
              :include [-start -stop -started? -info]])
link

Bus ^

NONE
(impl/defimpl Bus [state results output counters]
  :suffix "-bus"
  :string string-bus
  :protocols [std.protocol.component/IComponent
              :include [-start -stop -started? -info]])
link

bus ^

[] [m]
Added 3.0

creates and starts a bus

v 3.0
(defn bus
  ([] (bus nil))
  ([m]
   (-> (bus:create m)
       (component/start))))
link
(bus)
=> bus?

bus:all-ids ^

[{:keys [state], :as bus}]
Added 3.0

returns all registered ids

v 3.0
(defn bus:all-ids
  ([{:keys [state] :as bus}]
   (let [{:keys [threads]} @state]
     (set (keys threads)))))
link
(bus:with-temp bus
               (bus:all-ids bus))
=> (contains #{string?})

bus:all-threads ^

[{:keys [state], :as bus}]
Added 3.0

returns all registered threads

v 3.0
(defn bus:all-threads
  ([{:keys [state] :as bus}]
   (let [{:keys [threads]} @state]
     (collection/map-vals :thread threads))))
link
(bus:with-temp bus
               (= (first (vals (bus:all-threads bus)))
                  (Thread/currentThread)))
=> true

bus:close ^

[bus id]
Added 3.0

bus:closes all bus:opened loops

v 3.0
(defn bus:close
  ([bus id]
   (bus:send bus id {:op :exit})))
link
(bus:with-temp bus
               (let [{:keys [stopped id]} @(bus:open bus (fn [m]
                                                           (update m :value inc)))]
                 (Thread/sleep 10)
                 (bus:close bus id)
                 [@stopped (bus:get-count bus)]))
=> (contains-in [{:exit :normal,
                  :id string?
                  :unprocessed empty?
                  :start number?
                  :end number?}
                 1])

bus:close-all ^

[bus]
Added 3.0

stops all thread loops

v 3.0
(defn bus:close-all
  ([bus]
   (collection/map-entries (fn [id]
                    [id (bus:close bus id)])
                  (bus:all-ids bus))))
link
(bus:with-temp bus
               (let [_ @(bus:open bus (fn [m]
                                        (update m :value inc)))
                     _ @(bus:open bus (fn [m]
                                        (update m :value inc)))
                     _ @(bus:open bus (fn [m]
                                        (update m :value inc)))
                     _ @(bus:open bus (fn [m]
                                        (update m :value inc)))]
                 (Thread/sleep 10)
                 (bus:close-all bus)
                 (Thread/sleep 100)
                 (bus:get-count bus)))
=> 1

bus:create ^

[] [m]
Added 3.0

creates a bus

v 3.0
(defn bus:create
  ([] (bus:create nil))
  ([m]
   (map->Bus (merge {:state   (atom {})
                     :results (atom {})
                     :output  (queue/queue)
                     :counters {:sent (atom 0)
                                :received (atom 0)}}
                    m))))
link
(bus:create)
=> bus?

bus:deregister ^

[bus] [{:keys [state], :as bus} id]
Added 3.0

deregisters from the bus

v 3.0
(defn bus:deregister
  ([bus]
   (bus:deregister bus (bus:get-id bus)))
  ([{:keys [state] :as bus} id]
   (swap! state (fn [m]
                  (let [{:keys [thread]} (get-in m [:threads id])]
                    (-> m
                        (update :threads dissoc id)
                        (update :lookup dissoc thread)))))))
link
(bus:with-temp bus
               (bus:register bus "foo" (Thread.))
               (bus:deregister bus "foo")
               (bus:has-id? bus "foo"))
=> false

bus:get-count ^

[{:keys [state], :as bus}]
Added 3.0

returns the number of threads registered

v 3.0
(defn bus:get-count
  ([{:keys [state] :as bus}]
   (-> state deref :threads count)))
link
(bus:with-temp bus
               (bus:get-count bus))
=> 1

bus:get-id ^

[bus] [{:keys [state]} thread]
Added 3.0

gets registered id given thread

v 3.0
(defn bus:get-id
  ([bus]
   (bus:get-id bus (thread/thread:current)))
  ([{:keys [state]} thread]
   (let [{:keys [lookup]} @state]
     (get lookup thread))))
link
(bus:with-temp bus
               (bus:get-id bus))
=> string?

bus:get-queue ^

[bus] [{:keys [state], :as bus} obj]
Added 3.0

gets the message queue associated with the thread

v 3.0
(defn bus:get-queue
  ([bus]
   (bus:get-queue bus (thread/thread:current)))
  ([{:keys [state] :as bus} obj]
   (let [{:keys [threads lookup]} @state]
     (:queue (if (instance? Thread obj)
               (get threads (get lookup obj))
               (get threads obj))))))
link
(bus:with-temp bus
               (vec (bus:get-queue bus)))
=> []

bus:get-thread ^

[{:keys [state]} id]
Added 3.0

gets thread given an id

v 3.0
(defn bus:get-thread
  ([{:keys [state]} id]
   (let [{:keys [threads]} @state]
     (:thread (get threads id)))))
link
(bus:with-temp bus
               (->> (bus:get-id bus)
                    (bus:get-thread bus)))
=> (cc/thread:current)

bus:has-id? ^

[{:keys [state]} id]
Added 3.0

checks that the bus has a given id

v 3.0
(defn bus:has-id?
  ([{:keys [state]} id]
   (let [{:keys [threads]} @state]
     (boolean (get threads id)))))
link
(bus:with-temp bus
               (bus:has-id? bus (bus:get-id bus)))
=> true

bus:kill ^

[bus id]
Added 3.0

bus:closes all bus:opened loops

v 3.0
(defn bus:kill
  ([bus id]
   (let [thread (bus:get-thread bus id)]
     (if (and thread (not= thread (thread/thread:current)))
       (doto thread
         (thread/thread:interrupt))))))
link
(bus:with-temp bus
               (let [{:keys [id stopped]} @(bus:open bus (fn [m]
                                                           (update m :value inc)))]
                 (Thread/sleep 10)
                 (bus:kill bus id)
                 [(deref stopped 1000 :timeout) (bus:get-count bus)]))
=> (contains-in [{:exit :interrupted
                  :id string?
                  :unprocessed empty?
                  :start number?
                  :end number?}
                 1])

bus:kill-all ^

[bus]
Added 3.0

stops all thread loops

v 3.0
(defn bus:kill-all
  ([bus]
   (collection/map-entries (fn [id]
                    [id (bus:kill bus id)])
                  (bus:all-ids bus))))
link
(bus:with-temp bus
               (let [_ @(bus:open bus (fn [m]
                                        (update m :value inc)))
                     _ @(bus:open bus (fn [m]
                                        (update m :value inc)))
                     _ @(bus:open bus (fn [m]
                                        (update m :value inc)))
                     _ @(bus:open bus (fn [m]
                                        (update m :value inc)))]
                 (Thread/sleep 10)
                 (bus:kill-all bus)
                 (Thread/sleep 10)
                 (bus:get-count bus)))
=> 1

bus:open ^

[bus handler] [bus handler {:keys [id], :as opts, :or {id (std.lib.foundation/sid)}}]
Added 3.0

bus:opens a new handler loop given function

v 3.0
(defn bus:open
  ([bus handler]
   (bus:open bus handler {}))
  ([bus handler {:keys [id] :as opts
                 :or {id (std.lib.foundation/sid)}}]
   (if-not (component/started? bus) (std.lib.foundation/error "Bus not started" {:bus bus}))
   
   (let [started  (f/incomplete)
         opts     (assoc opts :id id :started started :stopped (f/incomplete))
         _        (f/future
                    (run-handler bus handler opts))]
     started)))
link
(bus:with-temp bus
               (let [{:keys [id]} (deref (bus:open bus (fn [m]
                                                         (update m :value inc)))
                                         1000 :timeout)]
                 (Thread/sleep 100)
                 (deref (bus:send bus id {:value 1}) 1000 :timeout)))
=> (contains {:value 2, :id string?})

(bus:with-temp bus
               (let [{:keys [id stopped]} (deref (bus:open bus (fn [m]
                                                                 (update m :value inc))
                                                           {:timeout 400})
                                                 1000 :timeout)]
                 [(deref stopped 1000 :timeout) (bus:get-count bus)]))
=> (contains-in [{:exit :timeout
                  :id string?
                  :start number?
                  :end number?}
                 1])

bus:register ^

[bus] [bus id] [{:keys [state], :as bus} id thread]
Added 3.0

registers a thread to the bus

v 3.0
(defn bus:register
  ([bus]
   (bus:register bus (std.lib.foundation/sid)))
  ([bus id]
   (bus:register bus id (thread/thread:current)))
  ([{:keys [state] :as bus} id thread]
   (let [queue (queue/queue)]
     (swap! state (fn [m]
                    (-> m
                        (assoc-in [:threads id]
                                  {:id id :thread thread :queue queue})
                        (assoc-in [:lookup thread] id))))
     queue)))
link
(bus:with-temp bus
               (bus:register bus "foo" (Thread.))
               (bus:has-id? bus "foo"))
=> true

bus:reset-counters ^

[bus]
Added 4.0

resets the counters for a bus

v 4.0
(defn bus:reset-counters
  [bus]
  (collection/map-vals #(atom/swap-return! % (fn [v] [v 0]))
              (:counters bus)))
link
(bus:with-temp bus
               (bus:send bus (bus:get-id bus) {:op :test})
               (bus:reset-counters bus)
               @(:received (:counters bus)))
=> 0

bus:send ^

[{:keys [results counters], :as bus} id msg]
Added 3.0

sends a message to the given thread

v 3.0
(defn bus:send
  ([{:keys [results counters] :as bus} id msg]
   (if-let [queue  (bus:get-queue bus id)]
     (let [msg-id (or (:id msg) (std.lib.foundation/sid))
           return (f/incomplete)]
       (swap! results assoc msg-id return)
       (queue/put queue (assoc msg :id msg-id))
       (swap! (:sent counters) inc)
       return))))
link
(bus:with-temp bus
               (bus:send bus (bus:get-id bus)
                         {:op :hello :message "world"})
               (cc/take (bus:get-queue bus) 1000 :ms))
=> (contains {:op :hello, :message "world", :id string?})

bus:send-all ^

[bus msg]
Added 3.0

bus:sends message to all thread queues

v 3.0
(defn bus:send-all
  ([bus msg]
   (collection/map-entries (fn [id] [id (bus:send bus id msg)]) (bus:all-ids bus))))
link
(bus:with-temp bus
               (bus:send-all bus {:op :all})
               (bus:wait bus {:timeout 1000}))
=> (contains {:op :all})

bus:wait ^

[bus] [bus {:keys [timeout timeunit], :or {timeunit :ms}}]
Added 3.0

bus:waits on the message queue for message

v 3.0
(defn bus:wait
  ([bus]
   (bus:wait bus {}))
  ([bus {:keys [timeout timeunit]
         :or {timeunit :ms}}]
   (queue/take (bus:get-queue bus) timeout timeunit)))
link
(bus:with-temp bus
               (bus:send bus (bus:get-id bus)
                         {:op :hello :message "world"})
               (bus:wait bus {:timeout 1000}))
=> (contains {:op :hello, :message "world", :id string?})

(bus:with-temp bus
               (bus:wait bus {:timeout 100}))

bus:with-temp ^

[var & body]
Added 3.0

checks if object is instance of Bus

v 3.0
(defmacro bus:with-temp
  ([var & body]
   `(let [~var   (bus)
          ~'_    (bus:register ~var)]
      (try ~@body
           (finally (bus:deregister ~var)
                    (stop-bus ~var))))))
link
(bus:with-temp bus
               (bus? bus))
=> true

bus? ^

[obj]
Added 3.0

checks if object is instance of Bus

v 3.0
(defn bus?
  ([obj]
   (instance? Bus obj)))
link
(bus? (bus))
=> true

handler-thunk ^

[{:keys [output], :as bus} handler {:keys [stopped on-timeout], session-id :id, :as opts}]
Added 3.0

creates a thread loop for given message handler

v 3.0
(defn handler-thunk
  ([{:keys [output] :as bus} handler {:keys [stopped on-timeout] session-id :id :as opts}]
   (fn []
     (try
       (loop []
         (if-let [{:keys [id op] :as msg} (bus:wait bus opts)]
           (cond (not= op :exit)
                 (let [[status data exception]
                       (try
                         [:success (handler msg) nil]
                         (catch InterruptedException e (throw e))
                         (catch Throwable t [:error nil t]))]
                   (queue/put output {:id id :status status :data data :exception exception})
                   (recur))

                 :else
                 (do (queue/put output {:id id :status :success :data {:exit true :stopped stopped}})
                     {:exit :normal :id session-id :unprocessed (bus:get-queue bus session-id)}))
           (let [return (and on-timeout (on-timeout))]
             (if-not (true? return)
               {:exit :timeout :id session-id}
               (recur)))))
       (catch InterruptedException e
         {:exit :interrupted :id session-id :unprocessed (bus:get-queue bus session-id)})))))
link
(bus:with-temp bus
               (let [output (q/queue)
                     bus (assoc bus :output output)
                     handler (fn [msg] (assoc msg :handled true))
                     thunk (handler-thunk bus handler {:stopped (f/incomplete)})
                     id (bus:get-id bus)]
                 (f/future
                   (bus:register bus id (Thread/currentThread))
                   (thunk))
                 (Thread/sleep 10)
                 (bus:send bus id {:op :test})
                 (q/take output 1000 :ms)))
=> (contains {:status :success
              :data (contains {:handled true})})

info-bus ^

[{:keys [state results output counters], :as bus}]
Added 3.0

returns info about the bus

v 3.0
(defn info-bus
  ([{:keys [state results output counters] :as bus}]
   {:running  (started?-bus bus)
    :threads  (count (bus:all-ids bus))
    :waiting  (count @results)
    :queued   (collection/map-vals (comp count :queue) (:threads @state))
    :sent     @(:sent counters)
    :received @(:received counters)}))
link
(bus:with-temp bus
               (loop [i 0]
                 (if (started?-bus bus)
                   (info-bus bus)
                   (if (< i 10)
                     (do (Thread/sleep 20) (recur (inc i)))
                     (info-bus bus)))))
=> (contains {:running true})

main-loop ^

[{:keys [state], :as bus}]
Added 3.0

creates a new message return loop

v 3.0
(defn main-loop
  ([{:keys [state] :as bus}]
   (let [thunk (main-thunk bus)]
     (swap! state assoc :main (thread/thread:current))
     (thunk))))
link
(let [state (atom {})]
  (with-redefs [main-thunk (fn [_]
                             (fn []
                               :done))]
    [(main-loop {:state state})
     (= (Thread/currentThread) (:main @state))]))
=> [:done true]

main-thunk ^

[{:keys [results output counters], :as bus}]
Added 3.0

creates main message return handler

v 3.0
(defn main-thunk
  ([{:keys [results output counters] :as bus}]
   (fn []
     (try
       (loop [{:keys [id status data exception] :as result} (queue/take output)]
         (let [return (get @results id)
               _ (swap! results dissoc id)
               _ (if (= :success status)
                   (f/future:force return data)
                   (f/future:force return :exception exception))
               _ (swap! (:received counters) inc)]
           (recur (queue/take output))))
       (catch InterruptedException e)))))
link
(bus:with-temp bus
               (let [thunk (main-thunk bus)
                     output (:output bus)
                     results (:results bus)
                     id "test-id"
                     ret (f/incomplete)
                     t (Thread. thunk)]
                 (swap! results assoc id ret)
                 (q/put output {:id id :status :success :data "data"})
                 (.start t)
                 (try
                   (deref ret 1000 :timeout)
                   (finally
                     (.interrupt t)
                     (.join t 100)))))
=> "data"

map->Bus ^

[m__7997__auto__]

NONE
(impl/defimpl Bus [state results output counters]
  :suffix "-bus"
  :string string-bus
  :protocols [std.protocol.component/IComponent
              :include [-start -stop -started? -info]])
link

run-handler ^

[bus handler] [bus handler {:keys [id started stopped on-start on-stop], :as opts}]
Added 3.0

runs the handler in a thread loop

v 3.0
(defn run-handler
  ([bus handler]
   (run-handler bus handler {}))
  ([bus handler {:keys [id started stopped on-start on-stop] :as opts}]
   (try
     (let [thunk (handler-thunk bus handler opts)
           id (or id (std.lib.foundation/sid))
           start (time/time-ns)
           thread (thread/thread:current)
           _ (bus:register bus id thread)
           _ (f/future:force started {:id id :thread thread :stopped stopped})
           _ (if on-start (on-start))
           output (thunk)
           _ (if on-stop (on-stop))
           _ (bus:deregister bus id)
           end (time/time-ns)
           output (assoc output :start start :end end)
           _ (f/future:force stopped output)]
       output)
     (catch Throwable t
       (.printStackTrace t)))))
link
(bus:with-temp bus
               (let [started (f/incomplete)
                     stopped (f/incomplete)
                     _ (f/future (run-handler bus (fn [m] (assoc m :done true))
                                               {:started started :stopped stopped}))
                     {id :id} (deref started 1000 :timeout)]
                 (bus:send bus id {:op :test})
                 (bus:send bus id {:op :exit})
                 (deref stopped 1000 :timeout)))
=> (contains {:exit :normal})

start-bus ^

[bus]
Added 3.0

starts the bus

v 3.0
(defn start-bus
  ([bus]
   (if-not (started?-bus bus)
     (thread/thread {:handler #(main-loop bus)
                     :daemon true
                     :start true}))
   bus))
link
(let [bus (start-bus (bus:create))]
  (loop [i 0]
    (if (started?-bus bus)
      true
      (if (< i 10)
        (do (Thread/sleep 20) (recur (inc i)))
        false))))
=> true

started?-bus ^

[{:keys [state], :as bus}]
Added 3.0

checks if bus is running

v 3.0
(defn started?-bus
  ([{:keys [state] :as bus}]
   (boolean (when-let [main ^Thread (:main @state)]
              (.isAlive main)))))
link
(bus:with-temp bus
               (Thread/sleep 10)
               (started?-bus bus))
=> true

stop-bus ^

[{:keys [state], :as bus}]
Added 3.0

stops the bus

v 3.0
(defn stop-bus
  ([{:keys [state] :as bus}]
   (if (started?-bus bus)
     (let [main ^Thread (:main @state)
           _ (swap! state dissoc :main)]
       (thread/thread:interrupt main)
       (bus:kill-all bus)))
   bus))
link
(let [bus (doto (bus:create)
            (start-bus))]
  (loop [i 0]
    (if (started?-bus bus)
      (do (stop-bus bus)
          (not (started?-bus bus)))
      (if (< i 10)
        (do (Thread/sleep 20) (recur (inc i)))
        false))))
=> true

string-bus ^

[bus]

NONE
(defn- string-bus
  ([bus]
   (str "#bus" (info-bus bus))))
link

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