verbo.sokka.worker
DEFAULT-TASK-KEEPALIVE-TIMEclj
execute!clj
(execute! {:keys [taskq topic pid pfn keepalive-ms timeout-ms]} ctrl task)execute!*clj
(execute!* {:keys [abort-chan close-chan] :as ctrl} pfn)failedclj
(failed)(failed {:keys [error-message]})keepalive!*clj
(keepalive!* {:keys [abort-chan close-chan] :as ctrl} keepalive-ms keepalive-fn)Spawns a thread that monitors the async pipeline, does regular
keepalive pings every keepalive-ms by calling the keepalive-fn,
signals termination by delivering value to a promise and does
housekeeping when things go wrong. Returns a record that implements
TaskCtrl.
Spawns a thread that monitors the async pipeline, does regular keepalive pings every `keepalive-ms` by calling the `keepalive-fn`, signals termination by delivering value to a promise and does housekeeping when things go wrong. Returns a record that implements `TaskCtrl`.
keepalive-fn!clj
(keepalive-fn! taskq id pid)Extends lease of task with the given id and pid with exponential
backoff retry.
Extends lease of task with the given `id` and `pid` with exponential backoff retry.
okclj
(ok)reserve!clj
(reserve! taskq topic pid)Reserve next message from the tasks service, with exponential backoff retries. This function will return nil if reservation was not successful after the set number of retries.
Reserve next message from the tasks service, with exponential backoff retries. This function will return nil if reservation was not successful after the set number of retries.
snoozedclj
(snoozed)(snoozed {:keys [snooze-time]})workerclj
(worker {:keys [taskq topic pid pfn lease-time-ms keepalive-ms timeout-ms
max-poll-interval-ms]
:as opts})Polls the task service on the given topic for a task and when
available, obtains a lease for the task, and calls processor
function pfn in a separate thread, passing the task as an
argument. It also spins up a sidecar thread (keepalive) to extend
the lease of the task periodically, while pfn is being executed.
processor-fn - should accept task as an argument, perform thecleanup-leased-tasks
operation corresponding to the task and return a tuple containing
[event-name opts]. valid event-names and args are:
[:sokka/completed nil], [:sokka/failed, {:keys [error-message]}]
[:sokka/snoozed, {:keys [snooze-time]}].
; keepalive-ms - keepalive timeout, will be lease-time * 0.7 ; task-timeout - how long do we expect the task to run. this can be overridden at the task level, but set at the worker level
The worker will wait until worker-timeout-ms has passed
for the task to complete, if the task isn't complete by then, it
closes the task, acknowledges the task with status = :failed and
carries on. If the task completes successfully, it acknowledges the
task with status = :ok and carries on.
The worker polls for tasks using an exponentially increasing sleeper function to prevent the worker from flooding the queue with requests during inactivity.
Polls the task service on the given `topic` for a task and when
available, obtains a lease for the task, and calls processor
function `pfn` in a separate thread, passing the task as an
argument. It also spins up a sidecar thread (keepalive) to extend
the lease of the task periodically, while pfn is being executed.
`processor-fn` - should accept task as an argument, perform thecleanup-leased-tasks
operation corresponding to the task and return a tuple containing
[event-name opts]. valid event-names and args are:
[:sokka/completed nil], [:sokka/failed, {:keys [error-message]}]
[:sokka/snoozed, {:keys [snooze-time]}].
; keepalive-ms - keepalive timeout, will be lease-time * 0.7
; task-timeout - how long do we expect the task to run. this can be
overridden at the task level, but set at the worker level
The worker will wait until `worker-timeout-ms` has passed
for the task to complete, if the task isn't complete by then, it
closes the task, acknowledges the task with status = :failed and
carries on. If the task completes successfully, it acknowledges the
task with status = :ok and carries on.
The worker polls for tasks using an exponentially increasing
sleeper function to prevent the worker from flooding the queue with
requests during inactivity.cljdoc is a website building & hosting documentation for Clojure/Script libraries
× close