Representation of a worker process with many concurrent job processors. Works at the concurrency set in options during construction. Will hold at most concurrency jobs in-memory while processing at any one time. Listens for signals to quiet or shutdown. Should not be started more than once per-process, nor should more than one worker be started per-process.

Typedefs

HI : object: An after-connect initial message from the server to handshake the connection
JobFunction : function: A function that executes work
Jobtype : string: Discriminator used by a worker to decide how to execute a job. This will be the name you used during register.
JobFunction : function: A function that executes work
Registry : Object.<Jobtype, JobFunction>: A lookup table holding the jobtype constants mapped to their job functions
ContextProvider : function: A function returned by a job function that will be called with the job context as its only argument and awaited. This exists to allow you to define simple job functions that only accept their job args, but in many cases you might need the job's custom properties or stateful connections (like a database connection) in your job and want to attach a connection for your job function to use without having to create it itself.
Context : object: A context object passed through middleware and to a job thunk
ContextProvider : function: A function returned by a job function that will be called with the job context as its only argument and awaited. This exists to allow you to define simple job functions that only accept their job args, but in many cases you might need the job's custom properties or stateful connections (like a database connection) in your job and want to attach a connection for your job function to use without having to create it itself.
HELLO : object: The client's response to the server's HI to initiate a connection
JobPayload : Object: A work unit that can be scheduled by the faktory work server and executed by clients
RejectedJobFromPushBulk : Object: A lookup table holding the jobtype constants mapped to their job functions
RejectedJobsFromPushBulk : Array: A lookup table holding the jobtype constants mapped to their job functions
RFC3339_DateTime : string: An RFC3339-format datetime string

API

faktory

creates faktory singletons

faktory
- .use(fn) ⇒ FaktoryControl
- .register(name, fn) ⇒ FaktoryControl
- .connect(...args) ⇒ Client
- .work(options) ⇒ Promise
- .stop() ⇒ promise

faktory.use(fn) ⇒ `FaktoryControl`

Adds a middleware function to the stack

Kind: instance method of faktory
Returns: FaktoryControl - this
See: koa middleware

Param	Type	Description
fn	`function`	koa-compose-style middleware function

Example

faktory.use(async (ctx, next) => {
  // a pool you created to hold database connections
  pool.use(async (conn) => {
    ctx.db = conn;
    await next();
  });
});

faktory.register(name, fn) ⇒ `FaktoryControl`

Adds a JobFunction to the Registry

Kind: instance method of faktory
Returns: FaktoryControl - this

Param	Type	Description
name	`Jobtype`	string descriptor for the jobtype
fn	`JobFunction`

Example

faktory.register('MyJob', (...args) => {
  // some work
});

faktory.connect(...args) ⇒ `Client`

Creates a new Client

Kind: instance method of faktory

Param	Type	Description
...args	`*`	args forwarded to Client

Example

const client = await faktory.connect();

await client.push(job);

faktory.work(options) ⇒ `Promise`

Starts a worker. Resolves after the worker is started. Only call this once per-process.

Kind: instance method of faktory
Returns: Promise - the Worker.work promise

Param	Type	Description
options	`object`	options to Worker

Example

// this keeps the process open and can be `await`ed
faktory.work();

faktory.stop() ⇒ `promise`

Stops the worker previously started.

Kind: instance method of faktory
Returns: promise - promise returned by Worker.stop
Example

// previously
faktory.work();

faktory.stop();

Client

A client connection handle for interacting with the faktory server. Holds a pool of 1 or more underlying connections. Safe for concurrent use and tolerant of unexpected connection terminations. Use this object for all interactions with the factory server.

Kind: global class

Client
- new Client([options])
- .connect() ⇒ Promise.<Client>
- .close() ⇒ Promise.<undefined>
- .job(jobtype, ...args) ⇒ Job
- .send(...args)
- .fetch(...queues) ⇒ Promise.<(object|null)>
- .beat() ⇒ Promise.<string>
- .push(job) ⇒ Promise.<string>
- .pushBulk(jobs) ⇒ Promise.<RejectedJobsFromPushBulk>
- .flush() ⇒ Promise.<string>
- .info() ⇒ Promise.<object>
- .ack(jid) ⇒ Promise.<string>
- .fail(jid, e) ⇒ Promise.<string>

new Client([options])

Creates a Client with a connection pool

Param	Type	Default	Description
[options]	`object`
[options.url]	`string`	`"tcp://127.0.0.1:7419"`	connection string for the faktory server (checks for FAKTORY_PROVIDER and FAKTORY_URL)
[options.host]	`string`	`"127.0.0.1"`	host string to connect to
[options.port]	`number` \| `string`	`7419`	port to connect to faktory server on
[options.password]	`string`		faktory server password to use during HELLO
[options.wid]	`string`		optional wid that should be provided to the server (only necessary for a worker process consuming jobs)
[options.labels]	`Array.<string>`	`[]`	optional labels to provide the faktory server for this client
[options.poolSize]	`number`	`10`	the maxmimum size of the connection pool

Example

const client = new Client();

const job = await client.fetch('default');

client.connect() ⇒ `Promise.<Client>`

Explicitly opens a connection and then closes it to test connectivity. Under normal circumstances you don't need to call this method as all of the communication methods will check out a connection before executing. If a connection is not available, one will be created. This method exists to ensure connection is possible if you need to do so. You can think of this like sqlx#MustConnect

Kind: instance method of Client
Returns: Promise.<Client> - resolves when a connection is opened

client.close() ⇒ `Promise.<undefined>`

Closes the connection to the server

Kind: instance method of Client

client.job(jobtype, ...args) ⇒ `Job`

Creates a new Job object to build a job payload

Kind: instance method of Client
Returns: Job - a job builder with attached Client for PUSHing
See: Job

Param	Type	Description
jobtype	`String`	name of the job function
...args	`*`	arguments to the job function

client.send(...args)

Borrows a connection from the connection pool, forwards all arguments to Connection.send, and checks the connection back into the pool when the promise returned by the wrapped function is resolved or rejected.

Kind: instance method of Client
See: Connection.send

Param	Type	Description
...args	`*`	arguments to Connection.send

client.fetch(...queues) ⇒ `Promise.<(object|null)>`

Fetches a job payload from the server from one of ...queues

Kind: instance method of Client
Returns: Promise.<(object|null)> - a job payload if one is available, otherwise null

Param	Type	Description
...queues	`String`	list of queues to pull a job from

client.beat() ⇒ `Promise.<string>`

Sends a heartbeat for this.wid to the server

Kind: instance method of Client
Returns: Promise.<string> - string 'OK' when the heartbeat is accepted, otherwise may return a state string when the server has a signal to send this client (quiet, terminate)

client.push(job) ⇒ `Promise.<string>`

Pushes a job payload to the server

Kind: instance method of Client
Returns: Promise.<string> - the jid for the pushed job

Param	Type	Description
job	`Job` \| `Object`	job payload to push

client.pushBulk(jobs) ⇒ `Promise.<RejectedJobsFromPushBulk>`

Pushes multiple jobs to the server and return map containing failed job submissions if any

Kind: instance method of Client
Returns: Promise.<RejectedJobsFromPushBulk> - response from the faktory server

Param	Type	Description
jobs	`Array.<Job>` \| `Array.<Object>`	jobs payload to push

client.flush() ⇒ `Promise.<string>`

Sends a FLUSH to the server

Kind: instance method of Client
Returns: Promise.<string> - resolves with the server's response text

client.info() ⇒ `Promise.<object>`

Sends an INFO command to the server

Kind: instance method of Client
Returns: Promise.<object> - the server's INFO response object

client.ack(jid) ⇒ `Promise.<string>`

Sends an ACK to the server for a particular job ID

Kind: instance method of Client
Returns: Promise.<string> - the server's response text

Param	Type	Description
jid	`String`	the jid of the job to acknowledge

client.fail(jid, e) ⇒ `Promise.<string>`

Sends a FAIL command to the server for a particular job ID with error information

Kind: instance method of Client
Returns: Promise.<string> - the server's response text

Param	Type	Description
jid	`String`	the jid of the job to FAIL
e	`Error`	an error object that caused the job to fail

Job

A class wrapping a JobPayload

Creating and pushing a job is typically accomplished by using a faktory client, which implements .job and automatically sets the client for the job when calling .push on the job later.

You do not need to use this class directly.`

Kind: global class

Job
- new Job(jobtype, [client])
- instance
  - .jid
  - .queue
  - .args
  - .priority
  - .retry
  - .at
  - .reserveFor
  - .custom
  - .toJSON() ⇒ object
  - .push() ⇒ string
- static
  - .jid() ⇒ string

new Job(jobtype, [client])

Creates a job

Param	Type	Description
jobtype	`string`	Jobtype string
[client]	`Client`	a client to use for communicating to the server (if calling push)

Example (with a faktory client)

// with a client
const client = await faktory.connect();
const job = client.job('SendWelcomeEmail', id);