Neverending Perpetual Motion
    Share your code. npm Orgs help your team discover, share, and reuse code. Create a free org »


    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 Build Status
    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',
    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',
    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,
    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,
    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


    • avatar