api.md

Contents

Classes

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.

Job

A class wrapping a JobPayload

Worker

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

JobThunk : 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

Registry : Object.<Jobtype, JobFunction>

A lookup table holding the jobtype constants mapped to their job functions

Command : Array.<string>

A command to send the server in array form

External

Jobtype : string

Discriminator used by a worker to decide how to execute a job. This will be the name you used during register.

timestamp : string

An RFC3339-format datetime string

JobPayload : object

A work unit that can be scheduled by the faktory work server and executed by clients

JobFunction : function

A function that executes work

HI : object

An after-connect initial message from the server to handshake the connection

HELLO : object

The client's response to the server's HI to initiate a connection

API

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
  • .close() ⇒ undefined
  • .job(jobtype, ...args) ⇒ Job
  • .send(...args)
  • .fetch(...queues) ⇒ Object | null
  • .beat() ⇒ String
  • .push(job) ⇒ String
  • .flush() ⇒ Promise
  • .info() ⇒ Object
  • .ack(jid) ⇒ String
  • .fail(jid, e) ⇒ String

new Client([options])

Creates a Client with a connection pool

Param	Type	Default	Description
[options]	object
[options.url]	string	"tcp://localhost:7419"	connection string for the faktory server (checks for FAKTORY_PROVIDER and FAKTORY_URL)
[options.host]	string	"localhost"	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

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 - resolves when a connection is opened

client.close() ⇒ 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) ⇒ Object | null

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

Kind: instance method of Client
Returns: 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() ⇒ String

Sends a heartbeat for this.wid to the server

Kind: instance method of Client
Returns: 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) ⇒ String

Pushes a job payload to the server

Kind: instance method of Client
Returns: String - the jid for the pushed job

Param	Type	Description
job	Job | Object	job payload to push

client.flush() ⇒ Promise

Sends a FLUSH to the server

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

client.info() ⇒ Object

Sends an INFO command to the server

Kind: instance method of Client
Returns: Object - the server's INFO response object

client.ack(jid) ⇒ String

Sends an ACK to the server for a particular job ID

Kind: instance method of Client
Returns: String - the server's response text

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

client.fail(jid, e) ⇒ String

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

Kind: instance method of Client
Returns: 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

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 client
const client = await faktory.connect();
const job = client.job('SendWelcomeEmail', id);

Example

// without a client
const job = new Job('SendWelcomeEmail');
job.args = [id];
job.queue = 'mailers';
console.log(job.toJSON());

job.jid

sets the jid

Kind: instance property of Job
See: external:JobPayload

Param	Type	Description
value	string	the >8 length jid

job.queue

sets the queue

Kind: instance property of Job
See: external:JobPayload

Param	Type	Description
value	string	queue name

job.args

sets the args

Kind: instance property of Job
See: external:JobPayload

Param	Type	Description
value	Array	array of positional arguments

job.priority

sets the priority of this job

Kind: instance property of Job
See: external:JobPayload

Param	Type	Description
value	number	0-9

job.retry

sets the retry count

Kind: instance property of Job
See: external:JobPayload

Param	Type	Description
value	number	{@see external:JobPayload}

job.at

sets the scheduled time

Kind: instance property of Job
See: external:JobPayload

Param	Type	Description
value	Date | string	the date object or RFC3339 timestamp string

job.reserveFor

sets the reserveFor parameter

Kind: instance property of Job
See: external:JobPayload

Param	Type
value	number

job.custom

sets the custom object property

Kind: instance property of Job
See: external:JobPayload

Param	Type	Description
value	object	the custom data

job.toJSON() ⇒ object

Generates an object from this instance for transmission over the wire

Kind: instance method of Job
Returns: object - the job as a serializable javascript object
Link: external:JobPayload|JobPayload}
See: external:JobPayload

job.push() ⇒ string

Pushes this job to the faktory server. Modifications after this point are not persistable to the server

Kind: instance method of Job
Returns: string - return of client.push(job)

Job.jid() ⇒ string

generates a uuid

Kind: static method of Job
Returns: string - a uuid/v4 string

Worker

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.

Kind: global class

• Worker
  • new Worker([options])
  • .inProgress ⇒ array
  • .work() ⇒ Worker
  • .quiet() ⇒ undefined
  • .stop() ⇒ promise

new Worker([options])

Param	Type	Default	Description
[options]	object
[options.wid]	String	uuid().slice(0, 8)	the wid the worker will use
[options.concurrency]	Number	20	how many jobs this worker can process at once
[options.shutdownTimeout]	Number	8	the amount of time in seconds that the worker may take to finish a job before exiting ungracefully
[options.beatInterval]	Number	15	the amount of time in seconds between each heartbeat
[options.queues]	Array.<string>	['default']	the queues this worker will fetch jobs from
[options.middleware]	Array.<function()>	[]	a set of middleware to run before performing each job in koa.js-style middleware execution signature
[options.registry]	Registry	Registry	the job registry to use when working

Example

const worker = new Worker({
  queues: ['critical', 'default', 'low'],
});

worker.work();

worker.inProgress ⇒ array

Returns an array of ids of processors with a job currently in progress

Kind: instance property of Worker
Returns: array - array of processor ids

worker.work() ⇒ Worker

starts the worker fetch loop and job processing

Kind: instance method of Worker
Returns: Worker - self, when working has been stopped by a signal or concurrent call to stop or quiet
See

• Worker.quiet
• Worker.stop

worker.quiet() ⇒ undefined

Signals to the worker to discontinue fetching new jobs and allows the worker to continue processing any currently-running jobs

Kind: instance method of Worker

worker.stop() ⇒ promise

stops the worker

Kind: instance method of Worker
Returns: promise - resolved when worker stops

JobThunk : 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.

Kind: global typedef
See: Context

Param	Type	Description
ctx	object	context object containing the job and any other data attached via userland-middleware

Example

// assumes you have middleware that attaches `db` to `ctx`

faktory.register('UserWelcomer', (...args) => (ctx) => {
  const [ id ] = args;
  const user = await ctx.db.users.find(id);
  const email = new WelcomeEmail(user);
  await email.deliver();
});

Context : object

A context object passed through middleware and to a job thunk

Kind: global typedef
Properties

Name	Type	Description
Context.job	object	the job payload
Context.fn	function	a reference to the job function

Registry : Object.<Jobtype, JobFunction>

A lookup table holding the jobtype constants mapped to their job functions

Kind: global typedef
See

• external:Jobtype
• external:JobFunction

Example

{
  SendWelcomeUser: (id) => {
    // job fn
  },
  GenerateThumbnail: (id, size) => {
    // job fn
  }
}

Command : Array.<string>

A command to send the server in array form

Kind: global typedef
Example

// multiple string arguments
['FETCH', 'critical', 'default']

// json string as an argument
['PUSH', '{"jid": "123"}']

// single string argument
['ACK', '123']

Jobtype : string

Discriminator used by a worker to decide how to execute a job. This will be the name you used during register.

Kind: global external
See: https://github.com/contribsys/faktory/wiki/The-Job-Payload
Example

// where `MyFunction` is the jobtype

faktory.register('MyFunction', () => {})

timestamp : string

An RFC3339-format datetime string

Kind: global external
Example

"2002-10-02T10:00:00-05:00"
"2002-10-02T15:00:00Z"
"2002-10-02T15:00:00.05Z"

new Date().toISOString();
// => '2019-02-11T15:59:15.593Z'

JobPayload : object

A work unit that can be scheduled by the faktory work server and executed by clients

Kind: global external
See

• https://github.com/contribsys/faktory/wiki/The-Job-Payload
• Faktory Protocol Specification - Work Units

Properties

Name	Type	Default	Description
[jid]	string	"uuid()"	globally unique ID for the job.
jobtype	Jobtype
[queue]	string	"default"	which job queue to push this job onto.
[args]	array	[]	parameters the worker should use when executing the job.
[priority]	number	5	higher priority jobs are dequeued before lower priority jobs.
[retry]	number	25	number of times to retry this job if it fails. 0 discards the failed job, -1 saves the failed job to the dead set.
[at]	timestamp		run the job at approximately this time; immediately if blank
[reserve_for]	number	1800	number of seconds a job may be held by a worker before it is considered failed.
custom	object		provides additional context to the worker executing the job.

JobFunction : function

A function that executes work

Kind: global external

Param	Type	Description
...args	*	arguments from the job payload

Example

function(...args) {
  // does something meaningful
}

HI : object

An after-connect initial message from the server to handshake the connection

Kind: global external
See: external:HELLO
Properties

Name	Type	Description
v	number	faktory server protocol version number
i	number	only present when password is required. number of password hash iterations. see HELLO.
s	string	only present when password is required. salt for password hashing. see HELLO.

HELLO : object

The client's response to the server's HI to initiate a connection

Kind: global external
See

• external:HI
• Faktory Protocol Specification

Properties

Name	Type	Description
v	string	the faktory client protocol version
hostname	string	name of the host that is running this worker
wid	string	globally unique identifier for this worker
pid	number	local process identifier for this worker on its host
labels	Array.<string>	labels that apply to this worker, to allow producers to target work units to worker types.
pwdhash	string	This field should be the hexadecimal representation of the ith SHA256 hash of the client password concatenated with the value in s.