0.0.11 • Public • Published

DBWrkr - A general use API for an easy to use pub-sub scheduler.


  • A pub-sub system makes it easy for multiple processes to communicate with each other by sending events. Subscribed processes will receive the events at the appropriate time.

  • This way, by creating (and reacting) to events you create loose coupling, little components that are easy to maintain, deploy and remove.

  • A storage-plugin system to store your events in your used technology like MongoDB, RethinkDB, Postgres, Redis, etc.

  • Although it could work pretty speedy, it's not build for performance, it's for ease of use, flexibility and introspection.


  • Currently this is in an early beta version. I'm using it in a production system and it seems to work pretty stable however, it could contain unplanned side-effects.

Build Status Coverage Status David Gitter

Storage engines

Backend Link Status
MongoDB https://github.com/whyhankee/dbwrkr-mongodb Build Status
RethinkDB https://github.com/whyhankee/dbwrkr-rethinkdb Build Status


npm install dbwrkr --save
npm install dbwrkr-<storage> --save

To publish events you need to

  • wrkr.connect() to connect to the DBWorker storage
  • wrkr.publish() an event
  • wrkr.disconnect() when your are done and want to exit

To process events to you need to:

  • wrkr.connect() to connect to the DBWorker storage
  • Get your self a queue from wrkr.queue()
  • subscribe to events using queue.subscribe()
  • use queue.on('event', fn) to setup your eventHandler (receives the events)
  • use wrkr.listen() to start receiving events
  • wrkr.disconnect() when your are done and want to exit

API - Setup

  • Require the modules (DBWrkr and storage module),
  • Create a storage object
  • Create the worker with the storage object
var wrkr = require('dbwrkr');
var DBWrkrMongodb = require('dbwrkr-mongodb');

var wrkr = new wrkr.DBWrkr({
  storage: new DBWrkrMongodb({
    dbName: 'dbwrkr'

wrkr.on('error', (errType, err, event) => {
    // do something with the error

Wrkr API

The Wrkr API is the main interface, it provides the main API but is also an EventEmitter.

It will Emit errors using the error event


Connect to the backend storage engine



Disconnect from the backend storage engine


wrkr.queue(queueName, callback)

Get a queue object representing a queue in the system.

wrkr.queue('queueName`, (err, q) => {
    // q = the Queue interface


Get a list of queues that are subscribed to the event.

wrkr.subscriptions(eventName, (err, queues) => {
  // queues is an array with the names of the queues

wrkr.publish(events, callback)

Publish new event(s). Events can be a single object or an array of objects. Events will we created for each queue that is subscribed to the event.


  • name - name of the event (required)
  • tid - target ID of the event (optional)
  • payload - Object with additional info (optional)
  • when - Date() when the event should be processed (optional, default immediate)
var events = [{
  name: 'yourapp.user.signup',
  tid: user.id,

wrkr.publish(events, (err, eventIds) => {
  // eventIds is an array with the ids the created events

wrkr.followUp(originalEvent, newEvent, callback)

FollowUp one event with another event. This will publish new event(s) with the parent set to the current event. This will help with the introspection system.

var newEvent = {
  name: 'yourapp.user.sentWelcomeMessage',
  tid: user.id,
wrkr.followUp(event, newEvent, (err, eventIds) => {
  // eventIds is an array with the created events

wrkr.retry(originalEvent, newEvent, callback)

Create a new retry event with the data of the current event also increasing the retryCount on the new event. You can set the when property to specify when the event should be retried, when defaults to an algorithm that should slowly increase to about 50 hours in 20 attempts.

When the retryCount becomes >= 20, an error will be returned.

wrkr.retry(event, when, (err, eventIds) => {
  // eventIds is an array with the created events

wrkr.find(findSpec, callback)

Find unprocessed events in the system.

var criteria = {
  name: eventName,
  tid: user.id,
wrkr.find(criteria, (err, events) => {
  // events is an array with matched events

wrkr.remove(eventSpec, callback)

Remove events in the system.

var criteria = {
  name: eventName,
  tid: user.id,
wrkr.remove(criteria, (err, events) => {
  // events is an array with matched events

Queue API

You will receive a queue interface using the wrkr.queue(queueName) method.

Listening for events

Events will be emitted to the Queue interace. You can handle the events by listening for the name of the event. Listening for * will receive the event if a specific event-listener was not available.

Listen to specific events

queue.on('eventName', function (event, callback) {
    console.log(event) 	// event is the processed event
    return callback(); 	// always call the callback the mark the event as done

queue.subscribe(eventName, callback)

Subscribe an event to a queue. The queue will emit the eventName

queue.subscribe(eventName, callback)

queue.unsubscribe(eventName, callback)

Unsubscribe an event from a queue. New events will no longer be created for this queue, although previously created qitems could still be received.

queue.unsubscribe(eventName, callback)

Testing, developing and debugging


  • The dbwrk package contains the tests. They are currently called from the storage engine, see the mongodb storage engine for more info.


DEBUG=wrkr* node ./example/example-mongodb.js

Event properties

id          Unique id of the event                          String  (indexed)
name        Name of the event                               String  (indexed)
queue       Name of the queue                               String  (indexed)
tid         Target id of the event (eg. userId/orderId)     String
payload     Object with extra properties for the event
parent      Id of the parent event (in case of a followUp)  String
created     Date created                                    Date
when        Date when the event should be processed         Date (sparse indexed)
done        Date when the event was processed               Date (sparse indexed)
retryCount  in case of an error followUp, the retryCount    Number



  • Statistics: Also store lastEventName/Tid


  • Statistics: No not overwrite .lastEvent* with undefined


  • Add q.statistics() to retrieve polling status info


  • fix: Invalid ‘error’ emit from __processNext


  • Update dependencies (debug, tape, eslint)
  • Map field (id, parent, tid) types to string, gives cleaner cross-db compatibility.
  • Fixed benchmarks in tests
  • DBWrkr.queue() - fix argument assert
  • Tinkering on debug() output


  • dbwrkr.queue() - opt argument is now optional


  • Update documentation
  • Use coveralls for coverage testing
  • Some API changes, mostly around wrkr.queue() method so we can use more than 1 queue per process
  • Rewrite tests using tape


  • Add to Travis-CI
  • Use mongodb storage for self-testing
  • Support for finding multiple ids in one call


  • Removed some es6 code


  • events are now emitted on the dbwrkr eventEmitter


  • Fix devDependency issue


  • Initial commit


npm i dbwrkr

DownloadsWeekly Downloads






Last publish


  • whyhankee