Add ':binding' support

Per metosin#54 in metosin/seippari, this commit
could be considered a starting point for supporting
bindings support in clojure.

The initial implementation used the starting
point @robert-stuttaford posted in the referenced
issue.

A tradeoff here, is that in order to support sync/async
w/ the various runtimes, using (bound-fn*) seems to be necessary
given the different thread pool implemenations. Counsequently,
there is a a perf hit needing to push/pop thread local vars. It
doesn't seem to be too much in the general case, but with manifold,
it seems to a an order of magnitude slower, though still in the
microseconds.

Author: bnert

src/sieppari/async.cljc

@@ -5,6 +5,12 @@
                    java.util.concurrent.CompletionException
                    java.util.function.Function)))
 
+#?(:clj
+  (defn -forward-bindings [f]
+    (fn [ctx]
+      (with-bindings (or (:bindings ctx) {})
+        ((bound-fn* f) ctx)))))
+
 (defprotocol AsyncContext
   (async? [t])
   (continue [t f])
@@ -21,20 +27,33 @@
    (extend-protocol AsyncContext
      Object
      (async? [_] false)
+     ; Given the implementation of enter/leave,
+     ; `continue` won't be called, and therefore,
+     ; the function call does not need to be bound
      (continue [t f] (f t))
      (await [t] t)))
 
 #?(:cljs
    (extend-protocol AsyncContext
      default
      (async? [_] false)
+     ; Given the implementation of enter/leave,
+     ; `continue` won't be called, and therefore,
+     ; the function call does not need to be bound
      (continue [t f] (f t))))
 
 #?(:clj
    (extend-protocol AsyncContext
      clojure.lang.IDeref
      (async? [_] true)
-     (continue [c f] (future (f @c)))
+     (continue [c f]
+       (cond
+         ; Make sure there is 
+         (map? @c)
+           (with-bindings (or (:bindings @c) {})
+             (future ((bound-fn* f) @c)))
+         :else
+           (future (f @c))))
      (catch [c f] (future (let [c @c]
                             (if (exception? c) (f c) c))))
      (await [c] @c)))
@@ -44,6 +63,8 @@
      CompletionStage
      (async? [_] true)
      (continue [this f]
+       ; Given the "context" is a completion stage, there isn't
+       ; a means 
        (.thenApply ^CompletionStage this
                    ^Function (->FunctionWrapper f)))
 

src/sieppari/async/core_async.cljc

@@ -9,7 +9,11 @@
   #?(:clj  clojure.core.async.impl.protocols.Channel
      :cljs cljs.core.async.impl.channels/ManyToManyChannel)
   (async? [_] true)
-  (continue [c f] (go (f (cca/<! c))))
+  #?(:clj  (continue [c f]
+            (let [f' (sa/-forward-bindings f)]
+              (go 
+                (f' (cca/<! c)))))
+     :cljs (continue [c f] (go (f (cca/<! c)))))
   (catch [c f] (go (let [c (cca/<! c)]
                      (if (exception? c) (f c) c))))
   #?(:clj (await [c] (<!! c))))

src/sieppari/async/manifold.clj

@@ -2,15 +2,31 @@
   (:require [sieppari.async :as sa]
             [manifold.deferred :as d]))
 
+; chain'-, as is being used here, is chain'-/3
+; chain'-/3, at the time of writing, has an arglist of:
+;  [d x f]
+; where:
+;  - `d` is a non-realized manifold deferred value, or nil
+;    to signal a deferred should be returned/provided
+;  - `x` is either a deferred or a value. If it is a deferred,
+;    then the deferred is recurively realized until a non-deferred value
+;    is yeilded.
+;  - `f` is a function applied to the unwrapped value `x`, before being either realized
+;    into `d` or being returned as a sucess or error deferred, depending on the result
+;    of `(f x)`.
 (extend-protocol sa/AsyncContext
   manifold.deferred.Deferred
   (async? [_] true)
-  (continue [d f] (d/chain'- nil d f))
-  (catch [d f] (d/catch' d f))
+  (continue [d f]
+    (d/chain'- nil d (sa/-forward-bindings f)))
+  (catch [d f]
+    (d/catch' d (sa/-forward-bindings f)))
   (await [d] (deref d))
 
   manifold.deferred.ErrorDeferred
   (async? [_] true)
-  (continue [d f] (d/chain'- nil d f))
-  (catch [d f] (d/catch' d f))
+  (continue [d f]
+    (d/chain'- nil d (sa/-forward-bindings f)))
+  (catch [d f]
+    (d/catch' d (sa/-forward-bindings f)))
   (await [d] (deref d)))

src/sieppari/core.cljc

@@ -14,10 +14,17 @@
   c/Context
   (context? [_] true))
 
+
 (defn- -try [ctx f]
   (if f
     (try
-      (let [ctx* (f ctx)]
+      (let [ctx* #?(:clj (with-bindings (or (:bindings ctx) {})
+                           ; Given the various async
+                           ; executors may exec on different threads,
+                           ; the fn must be bound in order to preserve
+                           ; bindings
+                           ((bound-fn* f) ctx))
+                    :cljs (f ctx))]
         (if (a/async? ctx*)
           (a/catch ctx* (fn [e] (assoc ctx :error e)))
           ctx*))
@@ -79,7 +86,7 @@
       (callback result))))
 
 (defn- remove-context-keys [ctx]
-  (dissoc ctx :error :queue :stack))
+  (dissoc ctx :bindings :error :queue :stack))
 
 ;;
 ;; Public API:

test/clj/sieppari/core_async_test.clj

@@ -246,3 +246,36 @@
                [:error :c]
                [:error :b]
                [:leave :a]])))
+
+(def ^:dynamic *boundv* 41)
+
+(defn bindings-handler [_]
+  (is (= 43 *boundv*))
+  (go
+    *boundv*))
+
+(def bindings-chain
+  [{:enter (fn [ctx]
+             (go
+               (assoc ctx
+                      :bindings
+                      {#'*boundv* 42})))
+    :leave (fn [ctx]
+             (go
+               (is (= 42 *boundv*))
+               ctx))}
+   {:enter (fn [ctx]
+             (is (= 42 *boundv*)
+                  "In interceptor failed")
+             (go
+               (update-in ctx [:bindings #'*boundv*] inc)))
+    :leave (fn [ctx]
+             (go
+               (update-in ctx [:bindings #'*boundv*] dec)))}
+
+   bindings-handler])
+
+(deftest async-bindings-test
+  (fact "bindings are conveyed across interceptor chain"
+    (sc/execute bindings-chain {}) => 43))
+

test/clj/sieppari/core_execute_test.clj

@@ -6,8 +6,7 @@
             [clojure.string :as str]))
 
 ;;
-;; Following tests use a test-chain that has some interceptors
-;; that fail on each stage function (enter, leave, error). The
+;; Following tests use a test-chain that has some interceptors that fail on each stage function (enter, leave, error). The
 ;; idea is that the tests override the expected stage functions
 ;; with test specific function. This ensures that no unexpected
 ;; stage functions are called.
@@ -269,6 +268,26 @@
         [:leave :x]
         [:leave :a]]))
 
+
+(def ^:dynamic *boundv* 41)
+
+(defn bindings-handler [_]
+  (is (= 42 *boundv*))
+  *boundv*)
+
+(def bindings-chain
+  [{:enter (fn [ctx] (assoc ctx
+                            :bindings
+                            {#'*boundv* 42}))}
+   {:enter (fn [ctx]
+             (is (= 42 *boundv*))
+             ctx)}
+   bindings-handler])
+
+(deftest use-bindings-test
+  (fact "bindings are conveyed across interceptor chain"
+    (s/execute bindings-chain {}) => 42))
+
 ; TODO: figure out how enqueue should work? Should enqueue add interceptors just
 ; before the handler?
 #_(deftest enqueue-interceptor-test

test/clj/sieppari/manifold_test.clj

@@ -228,3 +228,45 @@
                [:error :c]
                [:error :b]
                [:leave :a]])))
+
+(def ^:dynamic *boundv* 41)
+
+(defn bindings-handler [_]
+  (is (= 43 *boundv*))
+  (d/chain
+    nil
+    (fn [_]
+      *boundv*)))
+
+(def bindings-chain
+  [{:enter (fn [ctx]
+             (d/future
+               (assoc ctx
+                      :bindings
+                      {#'*boundv* 42})))
+    :leave (fn [ctx]
+             (d/chain
+               ctx
+               (fn [ctx']
+                 (is (= 42 *boundv*))
+                 ctx')))}
+   {:enter (fn [ctx]
+             (is (= 42 *boundv*)
+                  "In interceptor failed")
+             (d/chain
+               ctx
+               #(update-in
+                  %
+                  [:bindings #'*boundv*] inc)))
+    :leave (fn [ctx]
+             (d/chain
+               ctx
+               #(update-in
+                  %
+                  [:bindings #'*boundv*] dec)))}
+   bindings-handler])
+
+(deftest async-bindings-test
+  (fact "bindings are conveyed across interceptor chain"
+    (sc/execute bindings-chain {}) => 43))
+

test/clj/sieppari/promesa_test.clj

@@ -228,3 +228,38 @@
                [:error :c]
                [:error :b]
                [:leave :a]])))
+
+(def ^:dynamic *boundv* 41)
+
+(defn bindings-handler [_]
+  (is (= 43 *boundv*))
+  (p/resolved
+    *boundv*))
+
+(def bindings-chain
+  [{:enter (fn [ctx]
+             (p/resolved
+               (assoc ctx
+                      :bindings
+                      {#'*boundv* 42})))
+    :leave (fn [ctx]
+             (is (= 42 *boundv*))
+             ctx)}
+   {:enter (fn [ctx]
+             (is (= 42 *boundv*))
+             (-> ctx
+                 (update-in [:bindings #'*boundv*]
+                            inc)
+                 (p/resolved)))
+    :leave (fn [ctx]
+             (is (= 43 *boundv*))
+             (-> ctx
+                 (update-in [:bindings #'*boundv*]
+                            dec)
+                 (p/resolved)))}
+   bindings-handler])
+
+(deftest async-bindings-test
+  (fact "bindings are conveyed across interceptor chain"
+    (sc/execute bindings-chain {}) => 43))
+