Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/msolli/proletarian/llms.txt

Use this file to discover all available pages before exploring further.

start! and stop! manage the active lifecycle of a QueueWorker created by create-queue-worker. They are thin wrappers around the proletarian.protocols/QueueWorker protocol methods of the same names. Also documented here is process-next-job!, a lower-level function that is part of the internal polling machinery but is intentionally exposed for use in tests and the REPL. Namespace: proletarian.worker

​
start!

(start! queue-worker)
Sets up an internal scheduled thread pool with the number of threads configured via :proletarian/worker-threads (default 1). Each thread immediately begins polling the configured queue table at the interval set by :proletarian/polling-interval-ms. A small random jitter is added when scheduling subsequent threads to reduce lock contention on the job table at startup. If :proletarian/install-jvm-shutdown-hook? was set to true when the worker was created, the JVM shutdown hook is installed at this point. Returns true if the worker was successfully started, or nil if it was already running.

​
Usage

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

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

;; Start the worker — polling begins immediately
(worker/start! queue-worker)

​
stop!

(stop! queue-worker)
Initiates a graceful shutdown:
  1. Signals all worker threads to stop accepting new jobs.
  2. Waits up to :proletarian/await-termination-timeout-ms (default 10000 ms) for any in-flight jobs to complete.
  3. Calls the :proletarian/on-shutdown callback (if configured) after the thread pool has terminated.
  4. Removes the JVM shutdown hook (if one was installed).
In-flight jobs that are still running when the timeout expires will be interrupted. Because Proletarian uses database-level row locking (FOR UPDATE SKIP LOCKED), an interrupted job will remain in the job table and will be picked up again after the next start! or by another worker instance. Returns true if the worker was successfully stopped, or nil if it was already stopped.

​
Usage

;; Graceful shutdown — waits for in-flight jobs
(worker/stop! queue-worker)

​
Lifecycle management with a component system

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

;; Example using a simple atom-based component
(defn start-worker! [datasource]
  (let [w (worker/create-queue-worker datasource handle-job
                                      {:proletarian/worker-threads 2})]
    (worker/start! w)
    w))

(defn stop-worker! [queue-worker]
  (worker/stop! queue-worker))

​
process-next-job!

(process-next-job! data-source queue handler-fn log config)
Fetches the next eligible job from the database — within a transaction, using SELECT ... FOR UPDATE SKIP LOCKED — and invokes handler-fn. This is the core unit of work executed by each worker thread on every polling cycle. This function is part of the internal machinery, but is intentionally exposed as a public function for use in tests and at the REPL. It accepts no default values: you must supply all arguments explicitly. Creating a wrapper in your own codebase that provides sensible defaults is recommended.

​
Arguments

​
data-source
DataSource
required
A JDBC DataSource used to acquire a connection for this single poll-and-process cycle.
​
queue
keyword
required
The queue keyword to poll. Must match the value used when jobs were enqueued.
​
handler-fn
function
required
The job handler function. Called as (handler-fn job-type payload) by default, or (handler-fn job-map) in :advanced mode (see :proletarian.worker/handler-fn-mode in config).
​
log
function
required
A logger function (event-kw data-map). No default is provided — pass (fn [_ _]) to suppress output or supply your own logger.
​
config
map
required
A configuration map with internal keys (see proletarian.worker/create-queue-worker source for the full key set). Relevant keys include:
  • :proletarian.db/job-table — job table name
  • :proletarian.db/archived-job-table — archived job table name
  • :proletarian.db/serializer — Serializer instance
  • :proletarian.db/job-id-strategy — JobIdStrategy instance
  • :proletarian.worker/clock — java.time.Clock instance
  • :proletarian.worker/handler-fn-mode — :default or :advanced

​
Return Value

SituationReturn value
A job was found and processed without interrupttrue
A job was found but the thread was interrupted during processingfalse
No job was available in the queuenil

​
Testing & REPL example

(require '[proletarian.worker :as worker]
         '[proletarian.transit :as transit]
         '[proletarian.job-id-strategies :as job-id-strategies]
         '[proletarian.job :as job]
         '[next.jdbc :as jdbc])

;; A thin wrapper with test-friendly defaults
(defn process-one! [datasource]
  (worker/process-next-job!
    datasource
    :proletarian/default
    handle-job
    (fn [_event _data])  ; silent logger for tests
    {:proletarian.db/job-table          "proletarian.job"
     :proletarian.db/archived-job-table "proletarian.archived_job"
     :proletarian.db/serializer         (transit/create-serializer)
     :proletarian.db/job-id-strategy    (job-id-strategies/->postgresql-uuid-strategy)
     :proletarian.worker/clock          (java.time.Clock/systemUTC)
     :proletarian.worker/handler-fn-mode :default}))

;; In a test:
(jdbc/with-transaction [tx test-datasource]
  (job/enqueue! tx :myapp/send-email {:to "test@example.com"}))

(let [result (process-one! test-datasource)]
  (assert (true? result) "Expected one job to be processed"))
process-next-job! provides no default values for any argument. Creating a project-local wrapper function that supplies your application’s defaults makes test code more concise and ensures consistency with your production worker configuration.

Build docs developers (and LLMs) love

Get started for freeTalk to us