d-core.libs.workers — org.clojars.cinguilherme/d-core-std 0.2.0

bump! clj

(bump! stats path)

Increment a stats counter at the given nested path.

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 an unbuffered channel.

command! clj

(command! system worker-id msg)

Send a message to an exposed worker using its configured :put-mode.

:put-mode can be:

• :async (default) -> async/put!
• :block -> blocking >!!
• :drop -> non-blocking offer! with drop event

deliver-output! clj

(deliver-output! out-chan value {:keys [worker-id output-chan-id stats emit]})

Attempt to deliver a worker result to an output channel without blocking.

When the output channel is full, records a drop and emits an event.

emit! clj

(emit! emit event)

Emit an event through the provided emitter, if any.

error-event clj

(error-event ctx msg ex)

Build a structured error event from worker context + exception.

exposed-chan clj

(exposed-chan system worker-id)

Return the underlying channel for an exposed worker.

exposed-channels clj

(exposed-channels definition channels)

Return a map of exposed worker-id -> {:channel ... :channel-id ... :put-mode ...}.

guarded-go-call clj

(guarded-go-call worker-fn ctx msg {:keys [guard-ms]})

Run a worker function inside a go block and fail if it exceeds guard-ms.

This must be invoked from within a go context.

handle-worker-error! clj

(handle-worker-error! ctx msg ex {:keys [stats emit fail-chan fail-chan-id]})

Handle a worker error by emitting a structured event and optionally forwarding it to a fail channel.

make-channels clj

(make-channels channels)

Build a map of channel-id -> core.async channel from a channels config map.

record-drop! clj

(record-drop! stats emit {:keys [worker-id channel-id reason]})

Record a drop counter and emit a :workers.drop event.

record-error! clj

(record-error! stats emit event)

Record an error counter and emit a :workers.error event.

request! clj

(request! system worker-id msg)

Send a message with a reply-chan and return the reply-chan.

resolve-emitter clj

(resolve-emitter components)

Resolve an emitter function from components.

Accepted shapes:

• components {:observability <fn|{:emit fn}>}
• components {:obs <fn|{:emit fn}>}
• components {:logger <fn|{:log fn}>}

start-command-worker clj

(start-command-worker worker ctx {:keys [dev-guard? guard-ms stats emit]})

Start a command worker that consumes from :in and invokes :worker-fn.

Supported keys on worker:

• :dispatch (:go | :thread, default :go)
• :output-chan (optional channel-id for results)
• :fail-chan (optional channel-id for error events)

start-ticker-worker clj

(start-ticker-worker worker ctx stop-chan)

Start a ticker worker that emits ticks to :out at :interval-ms until stop.

Tick delivery is best-effort and non-blocking; when the output channel is full, ticks may be dropped and a drop event recorded (if stats/emit are provided).

start-worker-by-kind clj multimethod

Dispatch worker startup by :kind

start-workers clj

(start-workers definition components)

(start-workers definition components {:keys [dev-guard? guard-ms]})

Start a workers runtime.

Returns {:channels ... :exposed ... :stats ... :emit ... :stop! ...}.

stats-snapshot clj

(stats-snapshot system)

Return a snapshot of stats counters for drops and errors.

worker-ctx clj

(worker-ctx components channels worker-id worker)

Build the worker context map passed to worker functions.