proletarian.worker — msolli/proletarian 1.0.41-alpha

create-queue-worker clj

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

(create-queue-worker data-source
                     handler-fn
                     {:proletarian/keys
                        [queue job-table archived-job-table serializer 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
                           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!.

Arguments

• data-source – a javax.sql.DataSource factory for creating connections to the PostgreSQL database.
• handler-fn – the function that will be called when a job is pulled off the queue. It should be an arity-2 function or multimethod. The first argument is the job type (as provided to proletarian.job/enqueue!). The second argument is the job's payload (again, as provided to proletarian.job/enqueue!).
• options – an optional map describing configuration options, see below.

Options

The optional third 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!.
• :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]].

process-next-job! clj

(process-next-job! data-source queue handler-fn log config)

Gets the next job from the database table and runs it.

This function is part of the internal machinery of the Proletarian worker, but is being exposed as a public function for use in testing scenarios and in the REPL. No default values are provided for any of the arguments or configuration options. See the documentation for, and implementation of, create-queue-worker for what those default values are. It might be a good idea to create a wrapper function around this function, for use in your own application, that provides sensible values for all the arguments and config.

Arguments

• data-source – a javax.sql.DataSource factory for creating connections to the PostgreSQL database.
• queue – a keyword with the name of the queue.
• handler-fn – the function that will be called when a job is pulled off the queue. It should be an arity-2 function or multimethod. The first argument is the job type (as provided to proletarian.job/enqueue!). The second argument is the job's payload (again, as provided to proletarian.job/enqueue!).
• 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.
• config – a map describing configuration options, see below.

Config

• :proletarian.db/job-table – which PostgreSQL table to write the job to.
• :proletarian.db/archived-job-table – which PostgreSQL table to write archived jobs to.
• :proletarian.db/serializer – an implementation of the proletarian.protocols/Serializer protocol. You should use the same serializer for proletarian.job/enqueue!.
• :proletarian.worker/clock – a java.time.Clock instance to use for getting the current time.

Returns true if there was a job to be run, and the current thread did not receive an interrupt while handling the job. Returns false if there was an interrupt. Returns nil if there was no job to be run.

start! clj

(start! queue-worker)

Start the Queue Worker.

stop! clj

(stop! queue-worker)

Stop the Queue Worker.