proletarian.worker — msolli/proletarian 1.0.32-alpha

create-queue-worker clj

(create-queue-worker data-source)

(create-queue-worker data-source
                     {:proletarian/keys
                        [queue job-table archived-job-table serializer
                         context-fn log queue-worker-id polling-interval-ms
                         worker-threads await-termination-timeout-ms
                         install-jvm-shutdown-hook? on-shutdown clock]
                      :or {clock (Clock/systemUTC)
                           on-shutdown (fn [])
                           log log/println-logger
                           await-termination-timeout-ms 10000
                           polling-interval-ms 100
                           archived-job-table db/DEFAULT_ARCHIVED_JOB_TABLE
                           install-jvm-shutdown-hook? false
                           queue db/DEFAULT_QUEUE
                           context-fn (constantly {})
                           serializer (transit/create-serializer)
                           worker-threads 1
                           job-table db/DEFAULT_JOB_TABLE}})

Create and return a Queue Worker, which is an instance of proletarian.protocols/QueueWorker. After creation, the Queue Worker must be started using start!, and can be stopped using stop!.

data-source is a [[javax.sql.DataSource]] factory for creating connections to the PostgreSQL database.

Options

The optional second argument is an options map with the following keys, all optional with default values:
:queue - A keyword with the name of the queue. The default value is :proletarian/default.
:job-table - Which PostgreSQL table to write the job to. The default is proletarian.job. You should only have to override this if you changed the default table name during installation.
:archived-job-table - Which PostgreSQL table to write archived jobs to. The default is proletarian.archived_job. You should only have to override this if you changed the default table name during installation.
:serializer - An implementation of the proletarian.protocols/Serializer protocol. The default is a Transit serializer (see proletarian.transit/create-serializer). If you override this, you should use the same serializer for proletarian.job/enqueue!.
:context-fn - A function that Proletarian calls to provide the first argument to the handle-job! multimethod. It takes no arguments, and should return a map with data that provides useful "context" for the job being run.
:log - A logger function that Proletarian calls whenever anything interesting happens during operation. It takes one or two arguments: The first is a keyword identifying the event being logged. The second is a map with data describing the event. The default logging function is simply a println-logger that will print every event using println.
:queue-worker-id - A string identifying this Queue Worker. It is used as a thread prefix for names of threads in the thread pool. It is also added to the log event data under the key :proletarian.worker/queue-worker-id. The default value is computed from the name of the queue that this worker is getting jobs from.
:polling-interval-ms - The time in milliseconds to wait after a job is finished before polling for a new one. The default value is 100 milliseconds.
:worker-threads - The number of worker threads that work in parallel. The default value is 1.
:await-termination-timeout-ms - The time in milliseconds to wait for jobs to finish before throwing an error when shutting down the thread pool. The default value is 10000 (10 seconds).
:install-jvm-shutdown-hook? - Should Proletarian install a JVM shutdown hook that tries to stop the Queue Worker (using stop!) when the JVM is shut down? The default is false.
:on-shutdown - A function that Proletarian calls after the Queue Worker has shut down successfully. It takes no arguments, and the return value is discarded. The default function is a no-op.
:clock - The [[java.time.Clock]] to use for getting the current time. Used in testing. The default is [[java.time.Clock/systemUTC]].

start! clj

(start! queue-worker)

Start the given Queue Worker.

stop! clj

(stop! queue-worker)

Stop the given Queue Worker.