// Connecting
const Jackd = require('jackd')
const beanstalkd = new Jackd()
await beanstalkd.connect()
// Producing
await beanstalkd.put('Hello!')
// Consuming
const job = await beanstalkd.reserve() // => { id: '1', payload: 'Hello!' }
// ...process the job... then:
await beanstalkd.delete(job.id)
$ npm i jackd
⚠️ If you're usingjackd
in production, you should upgradejackd
to version 2.x. Read more here.
Most beanstalkd
clients don't support promises (fivebeans, nodestalker) and the ones that do have too many dependencies (node-beanstalkd-client). We wanted to make a package that has:
- A concise and easy to use API
- Native promise support
- No dependencies
- Protocol accuracy/completeness
beanstalkd
is a simple and blazing fast work queue. Producers connected through TCP sockets (by default on port 11300
) send in jobs to be processed at a later time by a consumer.
If you don't have experience using beanstalkd
, it's a good idea to read the beanstalkd
protocol before using this library.
const Jackd = require('jackd')
const beanstalkd = new Jackd()
await beanstalkd.connect() // Connects to localhost:11300
await beanstalkd.connect({ host, port })
await beanstalkd.disconnect() // You can also use beanstalkd.quit; it's an alias
You can add jobs to a tube by using the put
command, which accepts a payload and returns a job ID.
beanstalkd
job payloads are byte arrays. Passing in a Buffer
will send the payload as-is.
// This is a byte array of a UTF-8 encoded string
const jobId = await beanstalkd.put(Buffer.from('my long running job'))
You can also pass in a String
or an Object
and jackd
will automatically convert these values into byte arrays.
const jobId = await beanstalkd.put('my long running job') // Buffer.from(string)
const jobId = await beanstalkd.put({ foo: 'bar' }) // Buffer.from(JSON.stringify(object))
All jobs sent to beanstalkd have a priority, a delay, and TTR (time-to-run) specification. By default, all jobs are published with 0
priority, 0
delay, and 60
TTR, which means consumers will have 60 seconds to finish the job after reservation. You can override these defaults:
await beanstalkd.put(
{ foo: 'bar' },
{
delay: 2, // Two second delay
priority: 10,
ttr: 600 // Ten minute delay
}
)
Jobs with lower priorities are handled first. Refer to the protocol specs for more information on job options.
All jobs by default are added to the default
tube. You can change where you produce jobs with the use
command.
const tubeName = await beanstalkd.use('awesome-tube') // => 'awesome-tube'
await beanstalkd.put({ foo: 'bar' })
Consumers reserve jobs from tubes. Using await
on a reserve
command is a blocking operation and execution will stop until a job has been reserved.
const { id, payload } = await beanstalkd.reserve() // wait until job incoming
console.log({ id, payload }) // => { id: '1', payload: Buffer }
jackd
will return the payload as-is. This means you'll have to handle the encoding yourself. For instance, if you sent in an Object
, you'll need to first convert the Buffer
to a String
and then parse the JSON.
const { id, payload } = await beanstalkd.reserve()
const object = JSON.parse(payload.toString())
If you passed in a String
, you'll need to convert the incoming Buffer
to a string.
const { id, payload } = await beanstalkd.reserve()
const message = payload.toString()
Once you've reserved a job, there are several operations you can perform on it. The most common operation will be deleting the job after the consumer is finished processing it.
await beanstalkd.delete(id)
Consumers can also give up their reservation by releasing the job. You'll usually want to release the job if an error occurred on the consumer and you want to put it back in the queue immediately.
// Release immediately with high priority (0) and no delay (0)
await beanstalkd.release(id)
// You can also specify the priority and the delay
await beanstalkd.release(id,
priority: 10
delay: 10
})
However, you may want to bury the job to be processed later under certain conditions, such as a recurring error or a job that can't be processed. Buried jobs will not be processed until they are kicked.
await beanstalkd.bury(id)
// ... some time later ...
await beanstalkd.kickJob(id)
You'll notice that the kick operation is suffixed by Job
. This is because there is a kick
command in beanstalkd
which will kick a certain number of jobs back into the tube.
await beanstalkd.kick(10) // 10 buried jobs will be moved to a ready state
Consumers will sometimes need additional time to run jobs. You can touch
those jobs to let beanstalkd
know you're still processing them.
await beanstalkd.touch(id)
By default, all consumers will watch the default
tube only. Consumers can elect what tubes they want to watch.
const numberOfTubesWatched = await beanstalkd.watch('my-special-tube')
// => 2
Consumers can also ignore tubes.
const numberOfTubesWatched = await beanstalkd.ignore('default')
// => 1
Be aware that attempting to ignore the only tube being watched will return an error.
beanstalkd
has a number of commands that return YAML payloads. These commands mostly return statistics regarding the current beanstalkd
instance. jackd
, on purpose, does not ship with a YAML parser. This is to:
- Avoid dependencies
- Stay close to the protocol spec
- Let callers decide how to parse YAML
jackd
has full support for all commands, so you can expect to find these YAML commands in the API.
const stats = await beanstalkd.stats()
/* =>
---
current-jobs-urgent: 0
current-jobs-ready: 0
current-jobs-reserved: 0
current-jobs-delayed: 0
current-jobs-buried: 0
*/
You can then pipe this result through a YAML parser to get the actual contents of the YAML file.
const YAML = require('yaml')
const stats = await beanstalkd.executeMultiPartCommand('stats\r\n')
const { 'total-jobs': totalJobs } = YAML.parse(stats)
console.log(totalJobs)
// => 0
You may be looking to design a process that does nothing else but consume jobs. You can accomplish this with one jackd
client using async/await
. Here's an example implementation.
/* consumer.js */
const Jackd = require('jackd')
const beanstalkd = new Jackd()
start()
async function start() {
// Might want to do some error handling around connections
await beanstalkd.connect()
while (true) {
try {
const { id, payload } = await beanstalkd.reserve()
/* ... process job here ... */
await beanstalkd.delete(id)
} catch (err) {
// Log error somehow
console.error(err)
}
}
}
MIT