knitty.deferred — com.github.anjensan/knitty 0.7.0-alpha2

Provides asynchronous programming support for Knitty.

This namespace integrates Knitty's asynchronous operations with the Manifold deferred library. It supplies functions and macros to create, manage, and compose Knitty deferreds (KDeferred), wrap values and errors, construct custom executor pools, and implement asynchronous control flows (such as future, loop, while, bind, and zip operations). This integration enables efficient and composable asynchronous computations within the Knitty system.

Note: Core functions like future, future-call, run!, while, reduce, and loop are excluded from clojure.core in favor of custom implementations provided here.

*executor* clj

*sched-executor* clj

alt clj

(alt a)

(alt a b)

(alt a b c)

(alt a b c d)

(alt a b c d & vs)

Takes several values, some of which may be deferred, and returns a deferred that will yield the value that is realized first.

await! clj macro

(await! ls & ds)

await!* clj

(await!* ls ds)

Like await! but accept iterable collection of deferreds.

bind clj

(bind d val-fn)

(bind d val-fn err-fn)

Bind 1-arg callbacks fn to deferred. Returns new deferred with amended value. Fn val-fn takes single arugment with unwrapped value and may return a new value, another deferred or throw an exception. Optional err-fn accepts single argument with error.

bind-> clj macro

(bind-> expr & forms)

Combination of bind and threading macro ->. Similar to manifold.deferred/chain, but does not allow dynamic applying with seq of arguments. Use reduce if number of binded callbacks is known only at runtime.

bind-err clj

(bind-err mv f)

(bind-err mv exc f)

Bind 1-arg callback fn to deferred. Callback called when deferred is realized with an error. Use throw to re-throw error or return a new value. Optinal argument exc may specify which kind of errors we want to handle. It may be 1-arg predicate fn, subclass of java.lang.Throwable or map. In the last case only instances of IExceptionInfo are handled where their ex-data is a subset of the exc map.

(-> d
  (bind-err java.lang.IOException       #(handle-instances-of-ioexception %))
  (bind-err #(re-find #"text" (str %))  #(hande-exceptions-with-matched-text %))
  (bind-err {:type :my-error}           #(hande-exinfo-with-matched-type %))
  (bind-err                             #(handle-any-error %)))

bind-fnl clj

(bind-fnl d f0)

Bind 0-arg function to be executed when deferred is realized (either with value or error). Callback result is ignored, any thrown execiptions are re-wrapped.

bind-onto clj

(bind-onto d executor val-fn)

(bind-onto d executor val-fn err-fn)

Similar to bind, but callbacks on a provided executor.

bind-val clj

(bind-val d val-fn)

bind=> clj macro

(bind=> => expr & forms)

This macro allows you to chain deferred computations in a readable way, similar to bind->, but with the ability to apply a custom threading operator (e.g., =>).

(bind=> ->> (future [1 2 3])
  (map inc)
  (filter even?)
  (vec))

(bind=> (as-> x) (future 10)
  (+ x 1)
  (* 2 x))

build-fork-join-pool clj

(build-fork-join-pool {:keys [parallelism factory factory-prefix
                              exception-handler max-size min-size saturate
                              keep-alive-seconds min-runnable async-mode]})

Creates and returns a new Java ForkJoinPool instance with custom configuration.

Options (all keys are optional):

• :parallelism - Number of worker threads (default: available processors)
• :factory - Custom ForkJoinWorkerThreadFactory (default: internal factory with name prefix)
• :factory-prefix - String prefix for thread names (default: "knitty-fjp")
• :exception-handler - Function (fn [thread exception]) to handle uncaught exceptions in pool threads
• :max-size - Maximum pool size (default: 32768)
• :min-size - Minimum pool size (default: 0)
• :saturate - Predicate function (fn [pool]) to determine if pool is saturated
• :keep-alive-seconds - Idle thread keep-alive time in seconds (default: 60)
• :min-runnable - Minimum number of runnable threads (default: 1)
• :async-mode - If true, uses async mode for task scheduling (default: false)

call-after-all-arr' clj macro

(call-after-all-arr' ds-arr f)

Schedules an expression f to be called after all deferreds in the array ds-arr are realized. Returns a deferred that resolves to the result of f.

chain* clj

(chain* x fs)

Composes functions over the value x, returning a deferred containing the result.

claim! clj

(claim! d)

(claim! d token)

Attempts to claim the deferred for future updates.

connect clj

(connect d-from d-dest)

Conveys the realized value of d-from into d-dest.

create clj

(create)

(create token)

Create a new unrealized deferred.

deferred? clj

(deferred? x)

Returns true if the object is an instance of a deferred.

do-wrap clj macro

(do-wrap & body)

Run body and returns value as deferred, catch and wrap any exceptions.

error! clj macro

(error! d x)

(error! d x token)

Puts a deferred into a realized state with provided error x.

future clj macro

(future & body)

Equivalent to clojure.core/future, but returns a Knitty deferred.

future-call clj

(future-call f)

(future-call executor f)

Equivalent to clojure.core/future-call, but returns a Knitty deferred.

future-with clj macro

(future-with executor & body)

Equivalent to clo,jure.core/future, but returns a Knitty deferred.

impl-iterate-while* clj macro

(impl-iterate-while* init f p)

(impl-iterate-while* init
                     [fn1 [stepf-x] & stepf-body]
                     [fn2 [somef-x] & somef-body]
                     [fn3 [retrn-x] & retrn-body])

Internal macro for implementing iterative loops with deferreds. Executes a step expression f while a predicate p holds true.

iterate-while clj

(iterate-while stepf somef init)

Iteratively runs a step function stepf with an initial value init. After each step, checks the result of calling the predicate somef. Stops the loop when the predicate returns a falsy value. Both the step function and the initial value can be deferreds.

join clj

(join d)

Coerce 'deferred with deferred value' to deferred with plain value.

@(join (kd/future (kd/future (kd/future 1)))) ;; => 1

join1 clj

(join1 d)

Coerce 'deferred with deferred value' to deferred with plain value.

@(join1 (kd/future (kd/future 1))) ;; => 1

kd-await! clj macro

(kd-await! ls)

(kd-await! ls x1)

(kd-await! ls x1 & xs)

Internal macros used by yarns - prefer not to use it. Schedlue 0/1-arg function ls to execute after all deferreds are realized with values or at least one deferred is realized with an error. All deferreds must be instances of Knitty deferred (use wrap or wrap* for coercion).

kd-succeeded? clj macro

(kd-succeeded?)

(kd-succeeded? x1 & xs)

Internal macros used by yarns - prefer not to use it. Returns true when all instances of KDeferred are realized with values.

let-bind clj macro

(let-bind binds & body)

Monadic let. Unwraps deferreds during binding, returning result as deferred. Unlike manifold.deferred/let-flow all steps are resolved in sequential order, so use zip if parallel execution is required. Steps list can contain :let and :when special forms.

(let-bind [[x y] (zip (kd/future (rand) (kd/future (rand))) :when (< x y) ;; whole let-bind returns nil when condition is not met :let [d x] ;; bind without unwrapping z (kd/future (* x y))] (vector x y z @d))

listen! clj

(listen! x on-any)

(listen! x on-ok on-err)

Registers callback fns to run when deferred is realized. When only one callback is provided it shold be 0-arg fn. When both callbacks provided - they must accept 1 argument (value or error).

loop clj macro

(loop bindings & body)

A version of Clojure's loop which allows for asynchronous loops, via manifold.deferred/recur. loop will always return a deferred value, even if the body is synchronous. Note that loop does not coerce values to deferreds, actual Manifold deferreds must be used.

(loop [i 1e6] (chain (future i) #(if (p/zero? %) % (recur (p/dec %)))))

peel clj

(peel d)

(peel d fallback)

Get deferred value. Throws exeption when deferred is not realized or failed with an error.

recur clj macro

(recur & args)

A special recur that can be used with knitty.deferred/loop.

reduce clj

(reduce f initd xs)

Deferred-aware version of clojure.core/reduce. Step function f may return deferred values, xs may be sequence of deferreds.

revoke clj

(revoke d cancel-fn)

(revoke d cancel-fn err-handler)

Returns new deferred connected to the provided. When resulted deferred is realized but original is not - calls cancellation callback.

(let [f (java.core/future (Thread/sleep 1000) 1))
      x (-> (wrap* f)
            (bind inc)
            (revoke #(do
                       (println :operation-is-cancelled)
                       (clojure.core/future-cancel))))]
  (success! x 1)
  (assert (clojure.core/future-cancelled? f)))

Note that call to revoke should be last in a chain, because revokation is not propagated by binding fns.

(let [f (java.core/future (Thread/sleep 1000) 1))
      x (-> (wrap* f)
            (revoke #(do
                       (println :operation-is-cancelled)
                       (clojure.core/future-cancel)))
            (bind inc))]
  (success! x 1)
  ;; revokation was not pass through `(bind inc)` binding.
  (assert (not (clojure.core/future-cancelled? f))))

revoke-to clj

(revoke-to d rd)

(revoke-to d rd & rds)

Like revoke, but resolves rd with an exception instead of calling generic callback. Use it to bypass revokation to original deferred:

(let [x (call-some-function)]
  (-> x
     (bind #(process-x1 %))
     (bind #(procces-x2 %))
     (bind-err #(handle-error %))
     (revoke-to x)  ;; bypass cancelation to `x`
     ))

run! clj

(run! f xs)

Sequentially apply f to sequence of deferreds xs for side effects. Fn f may return deferreds.

sleep clj

(sleep time-ms)

(sleep value time-ms)

Creates a deferred that will be realized after a specified delay.

success! clj macro

(success! d x)

(success! d x token)

Puts a deferred into a realized state with provided value x.

timeout clj

(timeout d delay)

(timeout d delay timeout-value)

Adds a timeout to a deferred. If the deferred is not realized within the specified delay, it will be realized with a timeout exception or a provided timeout value.

timeout-exception clj

unwrap clj

(unwrap x)

Unwraps deferred (once). Returns realized value or the deferred itself.

unwrap1 clj

(unwrap1 x)

Unwraps deferred (once). Returns realized value or the deferred itself.

while clj macro

(while body)

(while pred & body)

Deferred-aware version of clojure.core/while. Both test and body experssion may result into deferred. Returns deferred realized to nil.

wrap clj

(wrap x)

Coerce type x from IDeferred to Knitty deferred or returns realized deferred with value x.

wrap* clj

(wrap* x)

Corece x into an instance of Knitty deferred. Converts non-deferred futures with manifold.deferred/->deferred.

wrap-err clj

(wrap-err e)

A realized deferred with error.

wrap-val clj

(wrap-val v)

A realized deferred with value.

zip clj

(zip)

(zip a)

(zip a b)

(zip a b c)

(zip a b c d)

(zip a b c d e)

(zip a b c d e f)

(zip a b c d e f g)

(zip a b c d e f g h)

(zip a b c d e f g h i)

(zip a b c d e f g h i j)

(zip a b c d e f g h i j k)

(zip a b c d e f g h i j k l)

(zip a b c d e f g h i j k l m)

(zip a b c d e f g h i j k l m n)

(zip a b c d e f g h i j k l m n o)

(zip a b c d e f g h i j k l m n o p)

(zip a b c d e f g h i j k l m n o p & z)

Takes several values and returns a deferred that will yield a vector of realized values.

zip* clj

(zip* vs)

(zip* a vs)

(zip* a b vs)

(zip* a b c vs)

(zip* a b c d vs)

(zip* a b c d e & vs)

Similar to (apply zip vs), returns a seq instead of vector.