Home

Awesome

PostgreSQL queue for Node.js:

Examples

const { QlobberPG } = require('qlobber-pg');
const qpg = new QlobberPG({
    name: 'example1',
    db: {
        host: '/var/run/postgresql',
        database: 'qlobber-pg'
    }
});
qpg.subscribe('foo.*', (data, info) => {
    console.log(info.topic, data.toString());
    const assert = require('assert');
    assert.equal(info.topic, 'foo.bar');
    assert.equal(data, 'hello');
});
qpg.publish('foo.bar', 'hello');

You can publish messages using a separate process if you like:

const { QlobberPG } = require('qlobber-pg');
const qpg = new QlobberPG({
    name: 'example2',
    db: {
        host: '/var/run/postgresql',
        database: 'qlobber-pg'
    }
});
qpg.publish('foo.bar', 'hello', qpg.stop);

Or use the streaming interface to read and write messages:

const { QlobberPG } = require('qlobber-pg');
const qpg = new QlobberPG({
    name: 'example3',
    db: {
        host: '/var/run/postgresql',
        database: 'qlobber-pg'
    }
});
function handler(stream, info) {
    const data = [];
    stream.on('readable', function () {
        let chunk;
        while (chunk = this.read()) {
            data.push(chunk);
        }
    });
    stream.on('end', function () {
        const s = Buffer.concat(data).toString();
        console.log(info.topic, s);
        const assert = require('assert');
        assert.equal(info.topic, 'foo.bar');
        assert.equal(s, 'hello');
    });
}
handler.accept_stream = true;
qpg.subscribe('foo.*', handler);
qpg.publish('foo.bar').end('hello');

The API is described here.

Installation

To install the module:

npm install qlobber-pg

You need to create a database on your PostgreSQL server. You can use an administration tool (e.g. pgAdmin) or the command line, for example:

psql -c 'create database "qlobber-pg";'

Then you need to run migrations on your database to create the table that qlobber-pg uses:

npm run migrate

Note: The database is assumed to be named qlobber-pg. If you created a database with a different name, you’ll need to change it in config/default.json.

Limitations

How it works

Publishing a message creates a row in a table in the database. The columns for each message are:

The database is periodically queried for new messages and a trigger is optionally created to invoke a check as soon as a message is published.

The query made against the table is constructed from the topics to which the QlobberPG instance is subscribed.

Licence

MIT

Test

To run the default tests (including stress tests):

npm test

To run the multi-process tests (each process publishing and subscribing to different messages):

npm run test-multi [-- --queues=<number of queues>]

If you omit --queues then one process will be created per core.

To run the distributed tests (one process per remote host, each one publishing and subscribing to different messages):

npm run test-remote [-- --remote=<host1> --remote=<host2> ...]

You can specify as many remote hosts as you like. The test uses cp-remote to run a module on each remote host. Make sure on each host:

Please note the distributed tests don’t run on Windows.

Lint

npm run lint

Code Coverage

npm run coverage

c8 results are available here.

Coveralls page is here.

Benchmarking

To run the benchmark:

npm run bench -- --rounds=<number of rounds> \
                 --size=<message size> \
                 --ttl=<message time-to-live in seconds> \
                 (--queues=<number of queues> | \
                  --remote=<host1> --remote=<host2> ...)

If you provide at least one --remote=<host> argument then the benchmark will be distributed across multiple hosts using cp-remote. Make sure on each host: