d-core.libs.pipeline — org.clojars.cinguilherme/d-core-std 0.4.0

Railway-oriented CSP pipeline for Clojure core.async.

Stages execute in sequence; successful results pass to the next stage's input channel, while any failure (thrown exception or error return value) is immediately routed to a consolidated error channel. Returning nil filters/drops the item.

channel-buffer clj

(channel-buffer {:keys [buffer buffer-type]})

Return a core.async buffer for channel config.

Supported keys:

• :buffer (size, default 0)
• :buffer-type (:fixed | :dropping | :sliding, default :fixed) A :buffer of 0 returns nil (unbuffered / rendezvous).

default-error? clj

(default-error? x)

Return true if x represents an error. Matches any Throwable, or a map with a non-nil, non-false :error key.

error-envelope clj

(error-envelope stage-name input err)

Build a structured error envelope map.

into-pipeline! clj

(into-pipeline! items stages)

(into-pipeline! items stages opts)

Feed a collection of items through a pipeline and synchronously collect the results as {:ok [...] :err [...]}.

Supported options:

• :timeout-ms (default 5000): abort if execution exceeds this duration
• any options supported by pipeline

make-channel clj

(make-channel)

(make-channel buf-cfg)

Create a core.async channel with optional buffer configuration.

make-intermediate-channels clj

(make-intermediate-channels normalized-stages
                            default-buffer
                            default-buffer-type)

Build intermediate channels between consecutive stages.

normalize-stage clj

(normalize-stage idx stage default-dispatch default-buffer default-buffer-type)

Normalize a stage definition (fn or map) into a standard stage map.

pipe-> clj macro

(pipe-> in-chan & stages-or-opts)

Convenience macro to construct a pipeline.

Usage: (pipe-> in-chan stage-fn1 stage-fn2) (pipe-> in-chan {:buffer 10} stage-fn1 stage-fn2)

pipeline clj

(pipeline in-chan stages)

(pipeline in-chan stages opts)

Construct and start a railway-oriented CSP pipeline.

Arguments:

• in-chan: input core.async channel
• stages: vector/seq of stage functions or maps:
  • fn: (fn [msg] ...)
  • map: {:fn f, :name :custom-name, :dispatch :go|:thread, :buffer ...}
• opts: optional map:
  • :out-chan: output channel for success results (defaults to unbuffered chan)
  • :err-chan: error channel for failures (defaults to unbuffered chan)
  • :buffer: default intermediate buffer size (default 0)
  • :buffer-type: default buffer type (:fixed, :dropping, :sliding)
  • :dispatch: default dispatch mode (:go or :thread, default :go)
  • :close-out?: whether to close out-chan when pipeline finishes (default true)
  • :close-err?: whether to close err-chan when all stages terminate (default true)

Returns: {:ok ok-chan, :err err-chan, :in in-chan, :stop! stop-fn}

safe-apply-stage clj

(safe-apply-stage stage-fn input error-pred)

Apply a stage function to input safely. Returns:

• [:drop nil] when result is nil
• [:err err] when an exception is thrown or error? matches
• [:ok res] on successful return value.

start-go-stage clj

(start-go-stage {:keys [fn name error?] :as _stage}
                in-chan
                out-chan
                err-chan
                stop-chan
                should-close-out?
                on-done!)

Start a stage running in a core.async go-loop.

start-stage clj

(start-stage stage
             in-chan
             out-chan
             err-chan
             stop-chan
             should-close-out?
             on-done!)

Dispatch stage execution based on its :dispatch strategy.

start-thread-stage clj

(start-thread-stage {:keys [fn name error?] :as _stage}
                    in-chan
                    out-chan
                    err-chan
                    stop-chan
                    should-close-out?
                    on-done!)

Start a stage running on a dedicated core.async thread.