create-queue-worker — Proletarian Queue Worker Factory

create-queue-worker constructs and returns a QueueWorker instance — the object responsible for polling the job table, deserializing payloads, and invoking your handler function. The worker is not started automatically after creation; you must call start! on it. This separation lets you wire the worker into your application’s component lifecycle (e.g. Integrant, Mount, or Component) before any polling begins. Namespace: proletarian.worker

Signatures

(create-queue-worker data-source handler-fn)
(create-queue-worker data-source handler-fn options)

Required Arguments

data-source

DataSource

required

A JDBC DataSource that Proletarian uses to acquire connections for polling. The worker obtains one connection per worker thread per poll cycle and manages connection lifecycle internally.

handler-fn

function

required

The function invoked when a job is dequeued. Its calling convention depends on :proletarian/handler-fn-mode:

:default mode (default): called as (handler-fn job-type payload) — an arity-2 function or multimethod where job-type is the keyword supplied to enqueue! and payload is the deserialized job data.
:advanced mode: called as (handler-fn job-map) — an arity-1 function receiving a map with keys :proletarian.job/job-type, :proletarian.job/payload, :proletarian.job/job-id, :proletarian.job/queue, :proletarian.job/enqueued-at, :proletarian.job/process-at, and :proletarian.job/attempts.

Options

All keys are optional. Pass options as a plain map in the third argument position.

Queue & Storage

:proletarian/queue

keyword

default:":proletarian/default"

The queue this worker consumes jobs from. Must match the :proletarian/queue used in enqueue! calls.

:proletarian/job-table

string

The fully-qualified table name for pending jobs. Override only if you renamed the table during schema installation.

:proletarian/archived-job-table

string

The fully-qualified table name for completed and failed jobs. Both successful and permanently-failed jobs are moved here after processing.

Serialization & ID Strategy

:proletarian/serializer

Serializer

default:"Transit JSON serializer"

An implementation of the Serializer protocol used to decode job payloads read from the database. Must be the same serializer used when enqueueing. Defaults to proletarian.transit/create-serializer.

:proletarian/job-id-strategy

JobIdStrategy

default:"PostgreSQL UUID strategy"

An implementation of the JobIdStrategy protocol used to decode job IDs read from the database. Must match the strategy used in enqueue!. Defaults to proletarian.job-id-strategies/->postgresql-uuid-strategy.

Handler Mode

:proletarian/handler-fn-mode

keyword

default:":default"

Controls how handler-fn is called. Accepted values:

:default — (handler-fn job-type payload)
:advanced — (handler-fn job-map) where job-map contains all job attributes

Retry & Failure

:proletarian/retry-strategy-fn

function

default:"(constantly nil)"

An arity-2 function (job-map exception) called when a job throws an exception. Should return a retry-strategy map:

{:retries 3              ; total number of retries
 :delays  [1000 5000 30000]} ; delay in ms before each retry attempt

If nil is returned (the default), the job is archived as failed without retrying.

:proletarian/failed-job-fn

function

default:"(constantly nil)"

An arity-2 function (job-map exception) called after a job has exhausted all retries and is being archived as permanently failed. Use it to trigger alerts or compensating actions. The return value is discarded.

Logging

:proletarian/log

function

default:"println-logger"

A logger function with signature (event-kw data-map). Proletarian calls this whenever a notable event occurs (job started, job finished, polling error, etc.). The default implementation prints every event to stdout via println. Supply your own to integrate with clojure.tools.logging, timbre, etc.

Worker Identity & Threading

:proletarian/queue-worker-id

string

default:"derived from queue name"

A human-readable identifier for this worker instance. Used as the thread-name prefix in the internal thread pool and included in every log event under the key :proletarian.worker/queue-worker-id. Defaults to "proletarian[<queue-name>]".

:proletarian/polling-interval-ms

integer

default:"100"

Milliseconds to wait between finishing one job and polling for the next one. A small jitter is applied between worker threads to reduce lock contention on the job table.

:proletarian/worker-threads

integer

default:"1"

Number of concurrent worker threads in the thread pool. Each thread independently polls for and processes jobs.

Error Handling & Shutdown

:proletarian/on-polling-error

function

default:"(constantly true)"

An arity-1 function (throwable) called when an unhandled Throwable is caught in the polling loop. If it returns a truthy value, the worker is stopped. The default always returns true, meaning any unexpected polling error stops the worker.

:proletarian/await-termination-timeout-ms

integer

default:"10000"

Maximum milliseconds to wait for in-flight jobs to complete when stop! is called before the thread pool is forcibly terminated. Default is 10 seconds.

:proletarian/install-jvm-shutdown-hook?

boolean

default:"false"

When true, Proletarian registers a JVM shutdown hook that calls stop! automatically when the JVM exits. Useful for standalone applications; consider managing lifecycle explicitly in server frameworks.

:proletarian/on-shutdown

function

default:"(fn [])"

A zero-arity function called after the worker has fully shut down. Use it to release resources or notify a health-check system. Return value is discarded.

Time

:proletarian/clock

Clock

default:"Clock/systemUTC"

The java.time.Clock used when computing timestamps for retry scheduling and archiving. Inject a fixed or offset clock in tests to control time deterministically.

Return Value

Returns an implementation of proletarian.protocols/QueueWorker. Call proletarian.worker/start! on the returned value to begin polling.

Examples

Minimal usage

(require '[proletarian.worker :as worker])

(defmulti handle-job (fn [job-type _payload] job-type))

(defmethod handle-job :myapp/send-welcome-email [_job-type {:keys [email]}]
  (send-email! email "Welcome!"))

(def queue-worker
  (worker/create-queue-worker datasource handle-job))

;; Start polling
(worker/start! queue-worker)

Full example with multiple options

(require '[proletarian.worker :as worker]
         '[proletarian.transit :as transit]
         '[proletarian.job-id-strategies :as job-id-strategies]
         '[clojure.tools.logging :as log])

(def queue-worker
  (worker/create-queue-worker
    datasource
    handle-job
    {:proletarian/queue                    :myapp/default
     :proletarian/job-table                "myapp.job"
     :proletarian/archived-job-table       "myapp.archived_job"
     :proletarian/serializer               (transit/create-serializer)
     :proletarian/job-id-strategy          (job-id-strategies/->postgresql-uuid-strategy)
     :proletarian/worker-threads           4
     :proletarian/polling-interval-ms      200
     :proletarian/await-termination-timeout-ms 15000
     :proletarian/install-jvm-shutdown-hook? true
     :proletarian/retry-strategy-fn        (fn [_job _ex]
                                             {:retries 3
                                              :delays  [1000 5000 30000]})
     :proletarian/failed-job-fn            (fn [job ex]
                                             (log/error ex "Job permanently failed" job))
     :proletarian/log                      (fn [event data]
                                             (log/info event data))
     :proletarian/on-shutdown              (fn [] (log/info "Worker shut down."))}))

create-queue-worker only creates the worker — it does not start polling. You must call (worker/start! queue-worker) before any jobs are processed.

proletarian.job

proletarian.worker

proletarian.protocols

proletarian.transit

Documentation Index

​Signatures

​Required Arguments

​Options

​Queue & Storage

​Serialization & ID Strategy

​Handler Mode

​Retry & Failure

​Logging

​Worker Identity & Threading

​Error Handling & Shutdown

​Time

​Return Value

​Examples

​Minimal usage

​Full example with multiple options

Build docs developers (and LLMs) love

Signatures

Required Arguments

Options

Queue & Storage

Serialization & ID Strategy

Handler Mode

Retry & Failure

Logging

Worker Identity & Threading

Error Handling & Shutdown

Time

Return Value

Examples

Minimal usage

Full example with multiple options