std.concurrent.request

single, bulk, asynchronous, and transactional request orchestration

Define one client request protocol and reuse it across synchronous calls, futures, batching, transformations, and transactions.

Concurrency overview

1    Overview

Request clients implement std.protocol.request/IRequest. The orchestration layer prepares options, invokes single or bulk transport functions, processes outputs, and applies pre, post, chain, timing, and error handlers. Plain functions also implement the request protocol.

2    Walkthrough

2.1    Make a single request

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

;; Functions are valid clients; each command is an argument vector.
(request/req + [1 2 3])

(request/req + [1 2 3]
             {:pre [(fn [args] (conj args 4))]
              :post [inc]
              :chain [str]})

2.2    Run asynchronously

Set :async to receive a future. :measure can be a callback, atom, or map of start, stop, and timer handlers.

(def pending
  (request/req + [1 2 3]
               {:async true
                :measure println}))
@pending

2.3    Batch and transact

(request/req:bulk [+ {:async false}]
  (request/req + [1 2])
  (request/req + [3 4]))

(request/req:transact [+ {}]
  (request/req + [1 2])
  (request/req + [3 4]))

(request/bulk:map + request/req-fn [[1 2] [3 4]])

3    API

*bulk* ^

NONE
(def ^:dynamic *bulk*)
link

*current* ^

NONE
(def ^:dynamic *current*)
link

*debug* ^

NONE
(def ^:dynamic *debug*    nil)
link

*inputs* ^

NONE
(def ^:dynamic *inputs*   nil)
link

*transact* ^

NONE
(def ^:dynamic *transact*)
link

bulk ^

[client thunk] [client thunk opts]
Added 3.0

allows query inputs to be combined

v 3.0
(defn bulk
  ([client thunk]
   (bulk client thunk {}))
  ([client thunk opts]
   (let [{:keys [async catch chain post] :as opts} (req:opts-init opts)
         [bulk current] (bulk-collect client thunk chain)
         received (f/on:all current
                            (fn [& arr]
                              (reduce h/call (apply vector arr) post)))
         received (cond-> received
                    catch (f/on:exception catch))
         final    (f/future:chain received chain)
         _ (cond bulk
                 (bulk-process client bulk opts)

                 :else
                 (swap! (get *current* client) conj final))]
     (req:return final (boolean (or (not bulk)
                                    (get *transact* client)
                                    async))))))
link
(bulk |client|
      (fn []
        (req-fn |client| {:type :eval, :form '(+ 1 2 3)})
        (req-fn |client| {:type :eval, :form '(+ 4 5 6)})))
=> [6 15]

(bulk |client|
      (fn []
        (req-fn |client| {:type :eval, :form '(+ 1 2 3)})
        (req-fn |client| {:type :eval, :form '(+ 4 5 6)}))
      {:chain [#(apply + %)]})
=> 21

(bulk |client|
      (fn []
        (req-fn |client| {:type :eval, :form 1} {:chain [inc]})
        (req-fn |client| {:type :eval, :form 2} {:chain [inc]})))
=> [2 3]

(bulk |client|
      (fn []
        (bulk |client|
              (fn []
                (req-fn |client| {:type :eval, :form 1} {:chain [inc]})
                (req-fn |client| {:type :eval, :form 2} {:chain [inc]})))
        (req-fn |client| {:type :eval, :form 3} {:chain [inc]})
        (req-fn |client| {:type :eval, :form 4} {:chain [inc]})))
=> [[2 3] 4 5]

(bulk |client|
      (fn []
        (bulk |client|
              (fn []
                (req-fn |client| {:type :eval, :form 1} {:chain [inc]})
                (req-fn |client| {:type :eval, :form 2} {:chain [inc]}))
              {:chain [#(apply + %)]})
        (req-fn |client| {:type :eval, :form 3} {:chain [inc]})
        (req-fn |client| {:type :eval, :form 4} {:chain [inc]}))
      {:chain [#(apply + %)]})
=> 14

bulk-collect ^

[client thunk chain]
Added 3.0

collects bulk inputs

v 3.0
(defn bulk-collect
  ([client thunk chain]
   (let [top     (get *bulk* client)
         context (or top (bulk-context))
         current (atom [])]
     (binding [*bulk*    (assoc *bulk* client context)
               *current* (assoc *current* client current)]
       (thunk)
       [(if (nil? top)
          (-> (dissoc context :cache)
              (merge @(:cache context))))
        @current]))))
link
(-> (bulk-collect |client|
                  (fn []
                    (req:unit |client| {:type :eval :form '(+ 1 2 3)})
                    (req:unit |client| {:type :eval :form '(+ 4 5 6)}))
                  [])
    first :inputs)
=> [{:type :eval, :form '(+ 1 2 3)}
    {:type :eval, :form '(+ 4 5 6)}]

bulk-context ^

[]
Added 3.0

creates a bulk context

v 3.0
(defn bulk-context
  ([]
   {:received (f/incomplete)
    :cache (atom {:inputs  []
                  :outputs []})}))
link
(bulk-context)
=> map?

bulk-process ^

[client {:keys [inputs received]} {:keys [pre], :as opts}]
Added 3.0

processes the client given bulk inputs

v 3.0
(defn bulk-process
  ([client {:keys [inputs received]} {:keys [pre] :as opts}]
   (let [result  (h/->> (reduce h/call inputs pre)
                        (request-bulk client % opts)
                        (process-bulk client inputs % opts))
         _  (f/future:force received result)]
     received)))
link
@(bulk-process |client|
               {:inputs [{:type :eval, :form '(+ 1 2 3)}
                         {:type :eval, :form '(+ 4 5 6)}]
                :received (future/incomplete)}
               {})
=> [6 15]

bulk:inputs ^

[client thunk]
Added 3.0

capture all inputs on the client

v 3.0
(defn bulk:inputs
  ([client thunk]
   (let [context (bulk-context)]
     (binding [*bulk*    (assoc *bulk* client context)
               *current* (assoc *current* client (atom []))]
       (thunk)
       (:inputs @(:cache context))))))
link
(bulk:inputs |client|
             (fn []
               (req:unit |client| {:type :eval :form '(+ 1 2 3)})
               (req:unit |client| {:type :eval :form '(+ 4 5 6)})))
=> [{:type :eval, :form '(+ 1 2 3)}
    {:type :eval, :form '(+ 4 5 6)}]

bulk:map ^

[client f inputs] [client f inputs opts]
Added 3.0

map function across a client

v 3.0
(defn bulk:map
  ([client f inputs]
   (bulk:map client f inputs {}))
  ([client f inputs opts]
   (bulk client
         (fn []
           (mapv (partial f client) inputs))
         opts)))
link
(bulk:map |client|
          req-fn [{:type :eval :form 1}
                  {:type :eval :form 2}])
=> [1 2]

bulk:transact ^

[client thunk] [client thunk opts]
Added 3.0

creates a transaction within a bulk context

v 3.0
(defn bulk:transact
  ([client thunk]
   (bulk:transact client thunk {}))
  ([client thunk opts]
   (let [{:keys [async chain pre post] :as opts} (req:opts-init opts)
         state (volatile! nil)
         _     (reduce h/call nil pre)]
     (bulk client
           (fn []
             (let [res (transact client thunk)]
               (vreset! state res)))
           (req:opts-clean opts))
     (-> @state
         (f/on:all (fn [& arr]
                     (reduce h/call (apply vector arr) post)))
         (f/future:chain chain)
         (req:return (boolean (or (get *bulk* client)
                                  (get *transact* client)
                                  async)))))))
link
(bulk:transact |client|
               (fn []
                 (req |client| {:type :eval :form 1}
                      {:chain [inc]})
                 (req |client| {:type :eval :form 1}
                      {:chain [#(* 2 %)]})))
=> [2 2]

fn-process-bulk ^

[_ _ data _]

NONE
(defn- fn-process-bulk
  ([_ _ data _] data))
link

fn-process-single ^

[_ data _]

NONE
(defn- fn-process-single
  ([_ data _] data))
link

fn-request-bulk ^

[f arr _]

NONE
(defn- fn-request-bulk
  ([f arr _]
   (mapv (partial apply f) arr)))
link

fn-request-single ^

[f args _]

NONE
(defn- fn-request-single
  ([f args _]
   (apply f args)))
link

measure:debug ^

[{:keys [context]}]
Added 3.0

creates a debug measure

v 3.0
(defn measure:debug
  ([{:keys [context]}]
   {:start (fn [input]
             (env/local :println :START
                        context input))
    :stop  (fn [output]
             (env/local :println :STOP
                        context output))}))
link
(measure:debug {})
=> (contains {:start fn? :stop fn?})

opts:timer ^

[timer]
Added 3.0

creates the timer

v 3.0
(defn opts:timer
  ([timer]
   (let [timer (cond (h/atom? timer)
                     (partial reset! timer)

                     :else timer)
         timer  (if (fn? timer)
                  (doto (f/incomplete)
                    (f/on:success timer))
                  timer)]
     timer)))
link
(opts:timer prn)
=> future/future?

opts:wrap-measure ^

[opts measure]
Added 3.0

creates the measure opts

v 3.0
(defn opts:wrap-measure
  ([opts measure]
   (let [{:keys [start stop timer]
          :or {start h/NIL
               stop  h/NIL}} (cond (map? measure)
                                   measure

                                   :else {:timer measure})
         timer    (opts:timer timer)
         holder   (atom nil)
         start-fn (fn [input]
                    (start input)
                    (reset! holder (t/time-ns))
                    input)
         stop-fn  (fn [output]
                    (if timer
                      (f/future:force timer
                                      {:output output
                                       :start @holder
                                       :end (t/time-ns)}))
                    (stop output)
                    output)]
     (-> opts
         (update :pre   (comp vec (partial cons start-fn)))
         (update :chain (fnil #(conj % stop-fn) []))))))
link
(opts:wrap-measure {} prn)
=> (contains-in {:pre [fn?], :chain [fn?]})

req ^

[client command] [client command opts]
Added 3.0

execute command on single, bulk and transact calls

v 3.0
(defmacro req
  ([client command]
   (let [context (-> (meta &form)
                     (assoc :ns (env/ns-sym)))]
     `(req-fn ~client ~command {:context (quote ~context)})))
  ([client command opts]
   (let [context (-> (meta &form)
                     (assoc :ns (env/ns-sym)))]
     `(req-fn ~client ~command (assoc ~opts :context (quote ~context))))))
link
(req |client| {:type :eval :form 1})
=> 1

(req |client| {:type :transact/start})
=> :transact/start

(req |client| {:type :eval :form 1})
=> :transact/queued

(req |client| {:type :eval :form 1})
=> :transact/queued

(req |client| {:type :transact/end})
=> [1 1]

req-fn ^

[client command] [client command opts]
Added 3.0

function for req

v 3.0
(defn req-fn
  ([client command]
   (req-fn client command {}))
  ([client command opts]
   (let [bulk     (get *bulk* client)
         transact (get *transact* client)
         opts     (req:opts-init opts)
         opts     (if transact
                    (transact-prep client transact command opts)
                    opts)]
     (cond *inputs*
           (swap! *inputs* conj command)

           bulk
           (req:unit client command opts)

           :else
           (req:single client command opts)))))
link
(binding [*inputs* (atom [])]
  (req-fn :client {:type :eval :form 1})
  @*inputs*)
=> [{:type :eval :form 1}]

req.fn ^

Added 3.0

functions

v 3.0
link
(req + [1 2 3 4])
=> 10

@(req + [1 2 3 4] {:async true})
=> 10

req.more ^

Added 3.0

execute command on single, bulk and transact calls

v 3.0
link
(req-fn |client| {:type :eval :form 1}
        {:post [inc inc]})
=> 3

(bulk |client|
      (fn []
        (req-fn |client| {:type :eval :form 1}
                {:post [inc inc]})))
=> [3]

(bulk |client|
      (fn []
        (req-fn |client| {:type :eval :form 1}
                {:pre [#(update % :form inc)]
                 :post [inc]})))
=> [3]

(bulk |client|
      (fn []
        (req-fn |client| {:type :eval :form 1}
                {:pre [#(update % :form inc)]
                 :post  [inc]
                 :chain [inc]})))
=> [4]

(->> (transact |client|
               (fn []
                 (req-fn |client| {:type :eval :form 1}
                         {:pre   [#(update % :form inc)]
                          :post  [inc]
                          :chain [inc]})))
     (map deref))
=> [4]

req.time ^

Added 3.0

execute command on single, bulk and transact calls

v 3.0
link
;;
;; Manual
;;
(let [state (atom {})
      _ (req-fn |client| {:type :eval :form 1}
                {:pre  [(fn [x]
                          (swap! state assoc :start (time/time-ns))
                          x)]
                 :post [(fn [x]
                          (swap! state assoc :end (time/time-ns))
                          x)]})
      {:keys [start end]} @state]
  (time/format-ns (- end start)))
=> string?

;;
;; With measure key
;;
(def -p- (promise))
(req-fn |client| {:type :eval :form 1}
        {:measure #(deliver -p- %)})

@-p-
;; {:output 1, :start 1600748490664158898, :end 1600748490664239782}
=> (contains {:start number?
              :end number?
              :output 1})

req:bulk ^

[[client opts] & body]
Added 3.0

creates a bulk request

v 3.0
(defmacro req:bulk
  ([[client opts] & body]
   (let [context (-> (meta &form)
                     (assoc :ns (env/ns-sym)
                            :type :bulk))]
     `(bulk ~client
            (fn []
              ~@body)
            (assoc ~opts :context (quote ~context))))))
link
(req:bulk [+]
          (req + [1 2 3])
          (req + [1 2 3])
          (req - [4 5 6]))
=> [6 6]

req:bulk.debug ^

Added 3.0

creates a bulk request

v 3.0
link
(-> (req:bulk [+ {:debug true}]
              (req-fn + [1 2 3] {:debug true})
              (req-fn + [1 2 3] {:debug true})
              (req-fn - [4 5 6] {:debug true}))
    (env/with-out-str))
=> string?

req:in ^

[& body]
Added 3.0

captures the command input

v 3.0
(defmacro req:in
  ([& body]
   `(binding [*inputs* (atom [])]
      ((fn [] ~@body))
      @*inputs*)))
link
(req:in
 (req nil [1])
 (req nil [2])
 (req nil [3]))
=> [[1] [2] [3]]

req:opts ^

[opts] [opts {:keys [pre post chain], :as m}]
Added 3.0

clean opts for processing inputs

v 3.0
(defn req:opts
  ([opts]
   (req:opts opts nil))
  ([opts {:keys [pre post chain] :as m}]
   (cond-> opts
     m     (merge (dissoc m :chain :post :pre))
     pre   (update :pre   #(vec (into % pre)))
     post  (update :post  (comp vec (partial concat post)))
     chain (update :chain (comp vec (partial concat chain))))))
link
(req:opts {} {})
=> {}

req:opts-clean ^

[opts]
Added 3.0

clean opts for processing inputs

v 3.0
(defn req:opts-clean
  ([opts]
   (dissoc opts :measure :time :debug :pre :post
           :chain :async :final :received :transact)))
link
(req:opts-clean {})
=> {}

req:opts-init ^

[{:keys [measure debug context], :as opts}]
Added 3.0

initialise request opts

v 3.0
(defn req:opts-init
  ([{:keys [measure debug context] :as opts}]
   (cond-> opts
     debug   (opts:wrap-measure (measure:debug opts))
     measure (opts:wrap-measure measure))))
link
(let [opts (req:opts-init {:debug true
                           :context {:line 1}})]
  [(= 1 (count (:pre opts)))
   (fn? (first (:pre opts)))
   (= 1 (count (:chain opts)))
   (fn? (first (:chain opts)))])
=> [true true true true]

req:return ^

[result async]
Added 3.0

returns the output

v 3.0
(defn req:return
  ([result async]
   (cond async result

         (f/future? result)
         @result

         :else
         result)))
link
(req:return 1 false)
=> 1

(req:return (future/completed 1)
            false)
=> 1

(req:return (future/completed 1) true)
=> future/future?

req:single ^

[client command] [client command opts]
Added 3.0

creates a single request call

v 3.0
(defn req:single
  ([client command]
   (req:single client command {}))
  ([client command opts]
   (let [{:keys [pre post final async] :as opts} (req:single-prep opts)
         result (h/-> command
                      (reduce h/call % pre)
                      (request-single client % opts)
                      (process-single client % opts))
         final  (req:single-complete opts result)]
     (req:return final (if (get *transact* client)
                         true
                         async)))))
link
(req:single |client| {:type :eval :form '(+ 1 2 3 4)})
=> 10

@(req:single |client| {:type :eval :form '(+ 1 2 3 4)}
             {:async true})
=> 10

(req:single |client| {:type :eval :form '(+ 1 2 3 4)}
            {:chain [inc inc]})
=> 12

@(req:single |client| {:type :eval :form '(+ 1 2 3 4)}
             {:async true
              :chain [inc inc]})
=> 12

req:single-complete ^

[{:keys [async transacted received final post chain], :as opts} result]
Added 3.0

completes the `req:single` call

v 3.0
(defn req:single-complete
  ([{:keys [async transacted received final post chain] :as opts} result]
   (cond transacted final

         async
         (do (f/future:force received
                             (reduce h/call result post))
             final)

         :else (h/-> result
                     (reduce h/call % post)
                     (reduce h/call % chain)))))
link
(req:single-complete (req:single-prep {})
                     1)
=> 1

@(req:single-complete (req:single-prep {:async true})
                      1)
=> 1

req:single-prep ^

[{:keys [async received transacted chain], :as opts}]
Added 3.0

prepares `req:single` options

v 3.0
(defn req:single-prep
  ([{:keys [async received transacted chain] :as opts}]
   (cond async
         (let [received (or received (f/incomplete))
               final    (if transacted
                          transacted
                          (f/future:chain received chain))]
           (assoc opts :received received :final final))

         :else opts)))
link
(req:single-prep {:async true})
=> map?

(req:single-prep {:async true :transacted (future/incomplete)})
=> map?

req:transact ^

[[client opts] & body]
Added 3.0

creates a bulk transaction request

v 3.0
(defmacro req:transact
  ([[client opts] & body]
   (let [context (-> (meta &form)
                     (assoc :ns (env/ns-sym)
                            :type :transact))]
     `(bulk:transact ~client
                     (fn []
                       ~@body)
                     (assoc ~opts :context (quote ~context))))))
link
(-> (req:transact [|client| {:debug true}]
                  (req |client| {:type :eval :form 1} {:debug true})
                  (req |client| {:type :eval :form 2} {:debug true}))
    (env/with-out-str))
=> string?

req:unit ^

[client command] [client command {:keys [pre post transacted catch chain], :as opts}]
Added 3.0

creates a single request call with `*bulk*` context

v 3.0
(defn req:unit
  ([client command]
   (req:unit client command {}))
  ([client command {:keys [pre post transacted catch chain] :as opts}]
   (let [{:keys [received cache]} (get *bulk* client)
         output (at/swap-return! cache
                                 (fn [{:keys [inputs outputs] :as m}]
                                   (let [i        (count inputs)
                                         output   (f/on:success received
                                                                (fn [arr]
                                                                  (reduce h/call (nth arr i) post)))
                                         output   (cond-> output
                                                    catch (f/on:exception catch))
                                         final    (if transacted
                                                    (f/future:chain transacted [])
                                                    (f/future:chain output chain))
                                         command  (if (h/iobj? command)
                                                    (vary-meta command merge (req:opts-clean opts))
                                                    command)]
                                     [final  (-> m
                                                 (update :inputs  conj (reduce h/call command pre))
                                                 (update :outputs conj final))])))
         _  (swap! (get *current* client) conj output)]
     output)))
link
(binding [*bulk*    {|client| (bulk-context)}
          *current* {|client| (atom [])}]
  (req:unit |client| {:type :eval :form '(+ 1 2 3)})
  (-> (get *bulk* |client|)
      :cache deref :inputs))
=> {:type :eval :form '(+ 1 2 3)}

transact ^

[client thunk]
Added 3.0

enable transactions on client

v 3.0
(defn transact
  ([client thunk]
   (let [top       (get *transact* client)
         context   (or top (transact-context))
         start-cmd (transact-start client)
         end-cmd   (transact-end client)
         _         (if (nil? top)
                     (req-fn client start-cmd {:async true}))
         results   (binding [*transact* (assoc *transact* client context)]
                     (thunk)
                     (:outputs @context))
         final     (-> @context :inputs :final)
         output    (if (nil? top)
                     (req-fn client end-cmd {:async true}))
         _   (if output
               (f/on:complete output
                              (fn [val err]
                                (if err
                                  (f/future:force final :exception err)
                                  (f/future:force final :value val)))))]
     results)))
link
(->> (transact |client|
               (fn []
                 (req |client| {:type :eval :form 1}
                      {:chain [#(* 8 %)]})))
     (map deref))
=> '(8)

;;
;; bulk in transact
;;

(->> (transact |client|
               (fn []
                 (bulk |client|
                       (fn []
                         (transact |client|
                                   (fn []
                                     (req |client|
                                          {:type :eval :form 1}
                                          {})))
                         (req |client|
                              {:type :eval :form 2}
                              {})))
                 (bulk |client|
                       (fn []
                         (req |client|
                              {:type :eval :form 3}
                              {})
                         (req |client|
                              {:type :eval :form 4}
                              {})))))
     (map deref))
=> '(1 2 3 4)

;;
;; Nesting bulk and transact
;;

(bulk |client|
      (fn []
        (req |client| {:type :eval :form 1}
             {:chain [#(* 8 %)]})
        (transact |client|
                  (fn []
                    (req |client| {:type :eval :form 1}
                         {:chain [#(* 2 %)]})
                    (req |client| {:type :eval :form 1}
                         {:chain [#(* 2 %)]})))
        (transact |client|
                  (fn []
                    (bulk |client|
                          (fn []
                            (req |client| {:type :eval :form 2}
                                 {:chain [#(* 2 %)]})))))
        (transact |client|
                  (fn []
                    (req |client| {:type :eval :form 3}
                         {:chain [#(* 2 %)]})))))
=> [8
    :transact/start 2 2 [1 1]
    :transact/start [4] [2]
    :transact/start 6 [3]]

transact-context ^

[]
Added 3.0

creates a transact context

v 3.0
(defn transact-context
  ([]
   (atom {:inputs {:final (f/incomplete)
                   :count 0}
          :outputs []})))
link
(transact-context)
=> f/atom?

transact-prep ^

[client context command {:keys [post chain], :as opts}]
Added 3.0

prepare req:opts given transaction context

v 3.0
(defn transact-prep
  ([client context command {:keys [post chain] :as opts}]
   (let [transacted (at/swap-return! context
                                     (fn [{:keys [inputs outputs] :as m}]
                                       (let [{:keys [final count]} inputs
                                             transacted  (-> (f/on:success
                                                              final
                                                              (fn [arr]
                                                                (reduce h/call (nth arr count) post)))
                                                             (f/future:chain chain))]
                                         [transacted (-> (update-in m [:inputs :count] inc)
                                                         (update :outputs conj transacted))])))]
     (assoc opts :transacted transacted))))
link
(transact-prep |client| (transact-context)
               {:type :eval, :form 1}
               {:chain [inc]})
=> map?

transact:map ^

[client f inputs] [client f inputs opts]
Added 3.0

performs a transaction across the client

v 3.0
(defn transact:map
  ([client f inputs]
   (transact:map client f inputs {}))
  ([client f inputs opts]
   (bulk:transact client
                  (fn []
                    (mapv (partial f client) inputs))
                  opts)))
link
(transact:map |client|
              req-fn [{:type :eval :form 1}
                      {:type :eval :form 2}])

=> [1 2]

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