(ns manifold.deferred

{:author "Zach Tellman"

:doc "Methods for creating, transforming, and interacting with asynchronous values."}

(:refer-clojure :exclude [realized? loop future])

(:require

[clojure.set :as set]

[clojure.tools.logging :as log]

[manifold.debug :as debug]

[manifold.executor :as ex]

[manifold.time :as time]

[manifold.utils :as utils :refer [assert-some definterface+]]

[potemkin.types :refer [def-abstract-type reify+ defprotocol+ deftype+]]

[clj-commons.primitive-math :as p]

[riddley.compiler :as compiler]

[riddley.walk :as walk])

(:import

(clojure.lang

IPending

IBlockingDeref

IDeref)

(java.io

Writer)

(java.util

LinkedList)

(java.util.concurrent

Future

TimeoutException

TimeUnit

CountDownLatch

Executor

CompletionStage

CompletableFuture)

(java.util.concurrent.atomic

AtomicInteger

AtomicLong)

(java.util.concurrent.locks

Lock)

(java.util.function

Function

BiFunction

Consumer

BiConsumer)))

(set! *warn-on-reflection* true)

(set! *unchecked-math* true)

(defprotocol+ Deferrable

(^:private to-deferred [_] "Provides a conversion mechanism to manifold deferreds."))

;; implies IDeref, IBlockingDeref, IPending

(definterface+ IDeferred

(executor [])

(^boolean realized [])

(onRealized [on-success on-error])

(successValue [default])

(errorValue [default]))

(declare then-apply then-apply-async

then-accept then-accept-async

then-run then-run-async

then-combine then-combine-async

then-accept-both then-accept-both-async

run-after-both run-after-both-async

apply-to-either apply-to-either-async

accept-either accept-either-async

run-after-either run-after-either-async

then-compose then-compose-async

then-handle then-handle-async

then-exceptionally

to-completable-future

when-complete when-complete-async)

;; The potemkin abstract type for

;; implementations such as CompletionStage

^{:author "Renan Ribeiro"

:added "0.4.0"}

(def-abstract-type

ADeferred

CompletionStage

(thenApply [d f]

(then-apply d f))

(thenApplyAsync [d f]

(then-apply-async d f))

(thenApplyAsync [d f executor]

(then-apply-async d f executor))

(thenAccept [d f]

(then-accept d f))

(thenAcceptAsync [d f]

(then-accept-async d f))

(thenAcceptAsync [d f executor]

(then-accept-async d f executor))

(thenRun [d f]

(then-run d f))

(thenRunAsync [d f]

(then-run-async d f))

(thenRunAsync [d f executor]

(then-run-async d f executor))

(thenCombine [d other f]

(then-combine d other f))

(thenCombineAsync [d other f]

(then-combine-async d other f))

(thenCombineAsync [d other f executor]

(then-combine-async d other f executor))

(thenAcceptBoth [d other f]

(then-accept-both d other f))

(thenAcceptBothAsync [d other f]

(then-accept-both-async d other f))

(thenAcceptBothAsync [d other f executor]

(then-accept-both-async d other f executor))

(runAfterBoth [d other f]

(run-after-both d other f))

(runAfterBothAsync [d other f]

(run-after-both-async d other f))

(runAfterBothAsync [d other f executor]

(run-after-both-async d other f executor))

(applyToEither [d other f]

(apply-to-either d other f))

(applyToEitherAsync [d other f]

(apply-to-either-async d other f))

(applyToEitherAsync [d other f executor]

(apply-to-either-async d other f executor))

(acceptEither [d other f]

(accept-either d other f))

(acceptEitherAsync [d other f]

(accept-either-async d other f))

(acceptEitherAsync [d other f executor]

(accept-either-async d other f executor))

(runAfterEither [d other f]

(run-after-either d other f))

(runAfterEitherAsync [d other f]

(run-after-either-async d other f))

(runAfterEitherAsync [d other f executor]

(run-after-either-async d other f executor))

(thenCompose [d f]

(then-compose d f))

(thenComposeAsync [d f]

(then-compose-async d f))

(thenComposeAsync [d f executor]

(then-compose-async d f executor))

(handle [d f]

(then-handle d f))

(handleAsync [d f]

(then-handle-async d f))

(handleAsync [d f executor]

(then-handle-async d f executor))

(exceptionally [d f]

(then-exceptionally d f))

(toCompletableFuture [d]

(to-completable-future d))

(whenComplete [d f]

(when-complete d f))

(whenCompleteAsync [d f]

(when-complete-async d f))

(whenCompleteAsync [d f executor]

(when-complete-async d f executor)))

(definline realized?

"Returns true if the manifold deferred is realized."

[x]

`(.realized ~(with-meta x {:tag "manifold.deferred.IDeferred"})))

(definline ^:no-doc success-value [x default-value]

`(.successValue ~(with-meta x {:tag "manifold.deferred.IDeferred"}) ~default-value))

(definline ^:no-doc error-value [x default-value]

`(.errorValue ~(with-meta x {:tag "manifold.deferred.IDeferred"}) ~default-value))

(defmacro ^:no-doc success-error-unrealized

[deferred

success-value

success-clause

error-value

error-clause

unrealized-clause]

`(let [d# ~deferred

~success-value (success-value d# ::none)]

(if (identical? ::none ~success-value)

(let [~error-value (error-value d# ::none)]

(if (identical? ::none ~error-value)

~unrealized-clause

~error-clause))

~success-clause)))

(definline on-realized

"Registers callbacks with the manifold deferred for both success and error outcomes."

[x on-success on-error]

`(let [^manifold.deferred.IDeferred x# ~x]

(.onRealized x# ~on-success ~on-error)

x#))

(definline deferred?

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

[x]

`(instance? IDeferred ~x))

(def ^:private satisfies-deferrable?

(utils/fast-satisfies #'Deferrable))

(defn deferrable?

"Returns true if the object can be coerced to a Manifold deferred."

[x]

(or

(instance? IDeferred x)

(instance? Future x)

(and (instance? IPending x) (instance? clojure.lang.IDeref x))

(satisfies-deferrable? x)))

;; TODO: do some sort of periodic sampling so multiple futures can share a thread

(defn- register-future-callbacks [x on-success on-error]

(if (or

(when (instance? Future x)

(or (.isDone ^Future x) (.isCancelled ^Future x)))

(when (instance? IPending x)

(.isRealized ^IPending x)))

(try

(on-success @x)

(catch Throwable e

(on-error e)))

(utils/wait-for

(try

(on-success @x)

(catch Throwable e

(on-error e))))))

(defn ->deferred

"Transforms `x` into a deferred if possible, or returns `default-val`. If no default value

is given, an `IllegalArgumentException` is thrown."

([x]

(let [x' (->deferred x ::none)]

(if (identical? ::none x')

(throw

(IllegalArgumentException.

(str "cannot convert " (.getCanonicalName (class x)) " to deferred.")))

x')))

([x default-val]

(cond

(deferred? x)

x

(satisfies-deferrable? x)

(to-deferred x)

(instance? Future x)

(let [^Future x x]

(reify+

IDeref

(deref [_]

(.get x))

IBlockingDeref

(deref [_ time timeout-value]

(try

(.get x time TimeUnit/MILLISECONDS)

(catch TimeoutException e

timeout-value)))

IPending

(isRealized [this]

(realized? this))

ADeferred

IDeferred

(realized [_]

(or (.isDone x) (.isCancelled x)))

(onRealized [_ on-success on-error]

(register-future-callbacks x on-success on-error))

(successValue [_ default]

(if (.isDone x)

(try

(.get x)

(catch Throwable e

default))

default))

(errorValue [this default]

(if (.realized this)

(try

(.get x)

default

(catch Throwable e

e))

default))))

(and (instance? IPending x)

(instance? clojure.lang.IDeref x))

(reify+

IDeref

(deref [_]

(.deref ^IDeref x))

IBlockingDeref

(deref [_ time timeout-value]

(.deref ^IBlockingDeref x time timeout-value))

IPending

(isRealized [_]

(.isRealized ^IPending x))

ADeferred

IDeferred

(realized [_]

(.isRealized ^IPending x))

(onRealized [_ on-success on-error]

(register-future-callbacks x on-success on-error)

nil)

(successValue [_ default]

(if (.isRealized ^IPending x)

(try

(.deref ^IDeref x)

(catch Throwable e

default))

default))

(errorValue [this default]

(if (.isRealized ^IPending x)

(try

(.deref ^IDeref x)

default

(catch Throwable e

e))

default)))

:else

default-val)))

;;;

(definterface+ IDeferredListener

(onSuccess [x])

(onError [err]))

(deftype+ Listener [on-success on-error]

IDeferredListener

(onSuccess [_ x] (on-success x))

(onError [_ err] (on-error err))

(equals [this x] (identical? this x))

(hashCode [_] (System/identityHashCode on-success)))

(defn listener

"Creates a listener which can be registered or cancelled via `add-listener!` and `cancel-listener!`."

([on-success]

(listener on-success (fn [_])))

([on-success on-error]

(Listener. on-success on-error)))

(definterface+ IMutableDeferred

(success [x])

(success [x claim-token])

(error [x])

(error [x claim-token])

(claim [])

(addListener [listener])

(cancelListener [listener]))

(defn success!

"Equivalent to `deliver`, but allows a `claim-token` to be passed in."

([^IMutableDeferred deferred x]

(.success deferred x))

([^IMutableDeferred deferred x claim-token]

(.success deferred x claim-token)))

(defn error!

"Puts the deferred into an error state."

([^IMutableDeferred deferred x]

(.error deferred x))

([^IMutableDeferred deferred x claim-token]

(.error deferred x claim-token)))

(defn claim!

"Attempts to claim the deferred for future updates. If successful, a claim token is returned, otherwise returns `nil`."

[^IMutableDeferred deferred]

(.claim deferred))

(defn add-listener!

"Registers a listener which can be cancelled via `cancel-listener!`. Unless this is useful, prefer `on-realized`."

[^IMutableDeferred deferred listener]

(.addListener deferred listener))

(defn cancel-listener!

"Cancels a listener which has been registered via `add-listener!`."

[^IMutableDeferred deferred listener]

(.cancelListener deferred listener))

(defmacro ^:private set-deferred [val token success? claimed? executor]

`(if (utils/with-lock* ~'lock

(when (and

(identical? ~(if claimed? ::claimed ::unset) ~'state)

~@(when claimed?

`((identical? ~'claim-token ~token))))

(set! ~'val ~val)

(set! ~'state ~(if success? ::success ::error))

true))

(do

(clojure.core/loop []

(when-let [^IDeferredListener l# (.poll ~'listeners)]

(try

(if (nil? ~executor)

(~(if success? `.onSuccess `.onError) ^IDeferredListener l# ~val)

(.execute ~(with-meta executor {:tag "java.util.concurrent.Executor"})

(fn []

(try

(~(if success? `.onSuccess `.onError) l# ~val)

(catch Throwable e#

#_(.printStackTrace e#)

(log/error e# "error in deferred handler"))))))

(catch Throwable e#

#_(.printStackTrace e#)

(log/error e# "error in deferred handler")))

(recur)))

true)

~(if claimed?

`(throw (IllegalStateException.

(if (identical? ~'claim-token ~token)

"deferred isn't claimed"

"invalid claim-token")))

false)))

(defmacro ^:private throw' [val]

`(if (instance? Throwable ~val)

(throw ~val)

(throw (ex-info "" {:error ~val}))))

(defmacro ^:private deref-deferred [timeout-value & await-args]

(let [latch-sym (with-meta (gensym "latch") {:tag "java.util.concurrent.CountDownLatch"})]

`(condp identical? ~'state

::success ~'val

::error (throw' ~'val)

(let [~latch-sym (CountDownLatch. 1)

f# (fn [_#] (.countDown ~latch-sym))]

(on-realized ~'this f# f#)

(let [result# ~(if (empty? await-args)

`(do

(.await ~latch-sym)

true)

`(.await ~latch-sym ~@await-args))]

(if result#

(if (identical? ::success ~'state)

~'val

(throw' ~'val))

~timeout-value))))))

(defmacro ^:private both [body]

`(do

~(->> body

(map #(if (and (seq? %) (= 'either (first %)))

(nth % 1)

[%]))

(apply concat))

~(->> body

(map #(if (and (seq? %) (= 'either (first %)))

(nth % 2)

[%]))

(apply concat))))

(both

(deftype+ (either [LeakAwareDeferred] [Deferred])

[^:volatile-mutable val

^:volatile-mutable state

^:volatile-mutable claim-token

^Lock lock

^LinkedList listeners

^:volatile-mutable mta

^:volatile-mutable consumed?

^Executor executor]

;; we only want one in a small number of deferreds to incur this cost at finalization,

;; but it's an important sanity check since we might be getting errors bubbling up all

;; over without them reaching somewhere we'd notice

(either

[Object

(finalize [_]

(utils/with-lock lock

(when (and (identical? ::error state) (not consumed?))

(log/warn val "unconsumed deferred in error state, make sure you're using `catch`."))))]

nil)

clojure.lang.IReference

(meta [_] mta)

(resetMeta [_ m]

(utils/with-lock* lock

(set! mta m)))

(alterMeta [_ f args]

(utils/with-lock* lock

(set! mta (apply f mta args))))

IMutableDeferred

(claim [_]

(utils/with-lock* lock

(when (identical? state ::unset)

(set! state ::claimed)

(set! claim-token (Object.)))))

(addListener [_ listener]

(set! consumed? true)

(when-let [f (utils/with-lock* lock

(condp identical? state

::success #(.onSuccess ^IDeferredListener listener val)

::error #(.onError ^IDeferredListener listener val)

(do

(.add listeners listener)

nil)))]

(if executor

(.execute executor f)

(f)))

true)

(cancelListener [_ listener]

(utils/with-lock* lock

(let [state state]

(if (or (identical? ::unset state)

(identical? ::set state))

(.remove listeners listener)

false))))

(success [_ x]

(set-deferred x nil true false executor))

(success [_ x token]

(set-deferred x token true true executor))

(error [_ x]

(set-deferred x nil false false executor))

(error [_ x token]

(set-deferred x token false true executor))

clojure.lang.IFn

(invoke [this x]

(if (success! this x)

this

nil))

ADeferred

IDeferred

(executor [_]

executor)

(realized [_]

(let [state state]

(or (identical? ::success state)

(identical? ::error state))))

(onRealized [this on-success on-error]

(add-listener! this (listener on-success on-error)))

(successValue [this default-value]

(if (identical? ::success state)

(do

(set! consumed? true)

val)

default-value))

(errorValue [this default-value]

(if (identical? ::error state)

(do

(set! consumed? true)

val)

default-value))

clojure.lang.IPending

(isRealized [this] (realized? this))

clojure.lang.IDeref

clojure.lang.IBlockingDeref

(deref [this]

(set! consumed? true)

(deref-deferred nil))

(deref [this time timeout-value]

(set! consumed? true)

(deref-deferred timeout-value time TimeUnit/MILLISECONDS))))

(deftype+ SuccessDeferred

[val

^:volatile-mutable mta

^Executor executor]

clojure.lang.IReference

(meta [_] mta)

(resetMeta [_ m]

(set! mta m))

(alterMeta [this f args]

(locking this

(set! mta (apply f mta args))))

IMutableDeferred

(claim [_] false)

(addListener [_ listener]

(if (nil? executor)

(.onSuccess ^IDeferredListener listener val)

(.execute executor #(.onSuccess ^IDeferredListener listener val)))

true)

(cancelListener [_ listener] false)

(success [_ x] false)

(success [_ x token] false)

(error [_ x] false)

(error [_ x token] false)

clojure.lang.IFn

(invoke [this x] nil)

ADeferred

IDeferred

(executor [_] executor)

(realized [this] true)

(onRealized [this on-success on-error]

(if executor

(.execute executor #(on-success val))

(on-success val)))

(successValue [_ default-value]

val)

(errorValue [_ default-value]

default-value)

clojure.lang.IPending

(isRealized [this] (realized? this))

clojure.lang.IDeref

clojure.lang.IBlockingDeref

(deref [this] val)

(deref [this time timeout-value] val))

(deftype+ ErrorDeferred

[^Throwable error

^:volatile-mutable mta

^:volatile-mutable consumed?

^Executor executor]

Object

(finalize [_]

(when (and

(not consumed?)

debug/*dropped-error-logging-enabled?*)

(log/warn error "unconsumed deferred in error state, make sure you're using `catch`.")))

clojure.lang.IReference

(meta [_] mta)

(resetMeta [_ m]

(set! mta m))

(alterMeta [this f args]

(locking this

(set! mta (apply f mta args))))

IMutableDeferred

(claim [_] false)

(addListener [_ listener]

(set! consumed? true)

(.onError ^IDeferredListener listener error)

true)

(cancelListener [_ listener] false)

(success [_ x] false)

(success [_ x token] false)

(error [_ x] false)

(error [_ x token] false)

clojure.lang.IFn

(invoke [this x] nil)

ADeferred

IDeferred

(executor [_] executor)

(realized [_] true)

(onRealized [this on-success on-error]

(set! consumed? true)

(if (nil? executor)

(on-error error)

(.execute executor #(on-error error))))

(successValue [_ default-value]

default-value)

(errorValue [_ default-value]

(set! consumed? true)

error)

clojure.lang.IPending

(isRealized [this] (realized? this))

clojure.lang.IDeref

clojure.lang.IBlockingDeref

(deref [this]

(set! consumed? true)

(throw' error))

(deref [this time timeout-value]

(set! consumed? true)

(throw' error)))

(let [created (AtomicLong. 0)]

(defn deferred

"Equivalent to Clojure's `promise`, but also allows asynchronous callbacks to be registered

and composed via `chain`."

([]

(deferred (ex/executor)))

([executor]

(if (and (p/zero? (rem (.incrementAndGet created) 1024))

debug/*dropped-error-logging-enabled?*)

(LeakAwareDeferred. nil ::unset nil (utils/mutex) (LinkedList.) nil false executor)

(Deferred. nil ::unset nil (utils/mutex) (LinkedList.) nil false executor)))))

(def ^:no-doc true-deferred- (SuccessDeferred. true nil nil))

(def ^:no-doc false-deferred- (SuccessDeferred. false nil nil))

(def ^:no-doc nil-deferred- (SuccessDeferred. nil nil nil))

(defn success-deferred

"A deferred which already contains a realized value"

#_{:inline

(fn [val]

(cond

(p/true? val) 'manifold.deferred/true-deferred-

(p/false? val) 'manifold.deferred/false-deferred-

(nil? val) 'manifold.deferred/nil-deferred-

:else `(SuccessDeferred. ~val nil nil)))

:inline-arities #{1}}

([val]

(SuccessDeferred. val nil (ex/executor)))

([val executor]

(if (nil? executor)

(condp identical? val

true true-deferred-

false false-deferred-

nil nil-deferred-

(SuccessDeferred. val nil nil))

(SuccessDeferred. val nil executor))))

(defn error-deferred

"A deferred which already contains a realized error"

([error]

(ErrorDeferred. error nil false (ex/executor)))

([error executor]

(ErrorDeferred. error nil false executor)))

(declare chain)

(defn unwrap'

"Like unwrap, but does not coerce deferrable values."

[x]

(if (deferred? x)

(let [val (success-value x ::none)]

(if (identical? val ::none)

x

(recur val)))

x))

(defn unwrap

"Recursively unwraps a deferred or deferrable until either 1) a non-deferred

value is reached, or 2) an unrealized deferrable is reached."

[x]

(let [d (->deferred x nil)]

(if (nil? d)

x

(let [val (success-value d ::none)]

(if (identical? ::none val)

d

(recur val))))))

(defn connect

"Conveys the realized value of `a` into `b`."

[a b]

(assert (deferred? b) "sink `b` must be a Manifold deferred")

(let [a (unwrap a)]

(if (instance? IDeferred a)

(if (realized? b)

false

(do

(on-realized a

#(let [a' (unwrap %)]

(if (deferred? a')

(connect a' b)

(success! b a')))

#(error! b %))

true))

(success! b a))))

(defn onto

"Returns a deferred whose callbacks will be run on `executor`."

[^IDeferred d executor]

(if (identical? executor (.executor d))

d

(let [d' (deferred executor)]

(connect d d')

d')))

(defmacro future-with

"Equivalent to Clojure's `future`, but allows specification of the executor

and returns a Manifold deferred."

[executor & body]

`(let [d# (deferred)]

(manifold.utils/future-with ~executor

(when-not (realized? d#)

(try

(success! d# (do ~@body))

(catch Throwable e#

(error! d# e#)))))

d#))

(defmacro future

"Equivalent to Clojure's `future`, but returns a Manifold deferred."

[& body]

`(future-with (ex/execute-pool) ~@body))

;;;

(defn- unroll-chain

[unwrap-fn v & fs]

(let [fs' (repeatedly (count fs) #(gensym "f"))

v-sym (gensym "v")

idx-sym (gensym "idx")

idx-sym' (gensym "idx")]

`(let [~@(interleave fs' fs)]

((fn this# [d# v# ^long idx#]

(try

(clojure.core/loop [~v-sym v# ~idx-sym idx#]

(let [~v-sym (~unwrap-fn ~v-sym)]

(if (deferred? ~v-sym)

(let [d# (or d# (deferred))]

(on-realized ~v-sym

(fn [v#] (this# d# v# ~idx-sym))

(fn [e#] (error! d# e#)))

d#)

(let [~idx-sym' (unchecked-inc ~idx-sym)]

(case (unchecked-int ~idx-sym)

~@(apply concat

(map

(fn [idx f]

`(~idx (recur (~f ~v-sym) ~idx-sym')))

(range)

fs'))

~(count fs) (if (nil? d#)

(success-deferred ~v-sym)

(success! d# ~v-sym))

nil)))))

(catch Throwable e#

(if (nil? d#)

(error-deferred e#)

(error! d# e#)))))

nil

~v

0))))

(let [;; factored out for greater inlining joy

subscribe (fn

([this d x]

(let [d (or d (deferred))]

(on-realized x

#(this d %)

#(error! d %))

d))

([this d x f]

(let [d (or d (deferred))]

(on-realized x

#(this d % f)

#(error! d %))

d))

([this d x f g]

(let [d (or d (deferred))]

(on-realized x

#(this d % f g)

#(error! d %))

d)))]

(defn ^:no-doc chain'-

([d x]

(try

(let [x' (unwrap' x)]

(if (deferred? x')

(subscribe chain'- d x')

(if (nil? d)

(success-deferred x')

(success! d x'))))

(catch Throwable e

(if (nil? d)

(error-deferred e)

(error! d e)))))

([d x f]

(try

(let [x' (unwrap' x)]

(if (deferred? x')

(subscribe chain'- d x' f)

(let [x'' (f x')]

(if (deferred? x'')

(chain'- d x'')

(if (nil? d)

(success-deferred x'')

(success! d x''))))))

(catch Throwable e

(if (nil? d)

(error-deferred e)

(error! d e)))))

([d x f g]

(try

(let [x' (unwrap' x)]

(if (deferred? x')

(subscribe chain'- d x' f g)

(let [x'' (f x')]

(if (deferred? x'')

(chain'- d x'' g)

(let [x''' (g x'')]

(if (deferred? x''')

(chain'- d x''')

(if (nil? d)

(success-deferred x''')

(success! d x'''))))))))

(catch Throwable e

(if (nil? d)

(error-deferred e)

(error! d e)))))

([d x f g & fs]

(when (or (nil? d) (not (realized? d)))

(let [d (or d (deferred))]

(clojure.core/loop [x x, fs (list* f g fs)]

(if (empty? fs)

(success! d x)

(let [[f g & fs] fs

d' (if (nil? g)

(chain'- nil x f)

(chain'- nil x f g))]

(success-error-unrealized d'

val (recur val fs)

err (error! d err)

(on-realized d'

#(apply chain'- d % fs)

#(error! d %))))))

d)))))

(let [;; factored out for greater inlining joy

subscribe (fn

([this d x]

(let [d (or d (deferred))]

(on-realized x

#(this d %)

#(error! d %))

d))

([this d x f]

(let [d (or d (deferred))]

(on-realized x

#(this d % f)

#(error! d %))

d))

([this d x f g]

(let [d (or d (deferred))]

(on-realized x

#(this d % f g)

#(error! d %))

d)))]

(defn ^:no-doc chain-

([d x]

(let [x' (unwrap x)]

(if (deferred? x')

(subscribe chain- d x')

(if (nil? d)

(success-deferred x')

(success! d x')))))

([d x f]

(if (or (nil? d) (not (realized? d)))

(try

(let [x' (unwrap x)]

(if (deferred? x')

(subscribe chain- d x' f)

(let [x'' (f x')]

(if (deferrable? x'')

(chain- d x'')

(if (nil? d)

(success-deferred x'')

(success! d x''))))))

(catch Throwable e

(if (nil? d)

(error-deferred e)

(error! d e))))

d))

([d x f g]

(if (or (nil? d) (not (realized? d)))

(try

(let [x' (unwrap x)]

(if (deferred? x')

(subscribe chain- d x' f g)

(let [x'' (f x')]

(if (deferrable? x'')

(chain- d x'' g)

(let [x''' (g x'')]

(if (deferrable? x''')

(chain- d x''')

(if (nil? d)

(success-deferred x''')