stanchionjs

StanchionJS

A simple & fast queue done right. backed by Redis, supports auto-reconnect, TypeScript, Promise and Rxjs.

Features

• Fast. Just BLPOP and RPUSH, no other fancy stuff, simple and fast.

• Auto-Reconnect.

• Works with Promise. Unleash the power of async / await, say bye-bye to callback.

• Better error handling.

• Written in TypeScript. No hacky code and provides better experience if you are using TypeScript.

• Reactive. If you don't like making promises, there're reactive APIs too, all APIs have two versions.

Installation

$ npm i stanchionjs

Examples

Initialize

const { Stanchion } = require('stanchionjs');

const stanchion = new Stanchion({
    redis: {
        host: '127.0.0.1',
        port: 6379,
        db: '1',
    },
    concurrency: 20,
});

How To Create a Job

const { Stanchion } = require('stanchionjs');

const stanchion = new Stanchion();

// Payload will be serialized using `JSON.stringify`
const job_1 = {
    thing: 'apple',
};
const job_2 = {
    thing: 'banana',
};

//
// Promise way:
//

stanchion.push(job_1, job_2).then(() => {
    console.log('jobs created.');
});

//
// Rxjs way:
//

stanchion.push$(job_1, job_2).subscribe(() => {
    console.log('jobs created.');
});

How To Process a Job

When a Job processing is done, you must tell StanchionJS so it can go fetching next Job for you. StanchionJS provides several ways to do that:

const { Stanchion } = require('stanchionjs');
const { Observable } = require('rxjs'); // Not required.

const stanchion = new Stanchion();

//
// Promise way:
//

stanchion.process(job => new Promise((resolve, reject) => {
    console.log('Got a job:', job);

    resolve();
}));

//
// Async / Await way:
//

stanchion.process(async job => {
    console.log('Got a job:', job);
});

//
// Rxjs way:
//

stanchion.process$(
    job => Observable.of(job)
        .map(job => {
            console.log('Got a job:', job);
        })
).subscribe(); // Don't forget to subscribe!

Error Handling

Every exception (including those from redis) can be obtained by attach handler to Stanchion instance:

const { Stanchion } = require('stanchionjs');

const stanchion = new Stanchion();

//
// Callback handler:
//

stanchion.onError(err => {
    console.log('error occurred', err);
});

//
// Rxjs stream:
//

stanchion.onError$().subscribe(err => {
    console.log('error occurred', err);
});

How To Exit

const { Stanchion } = require('stanchionjs');

const stanchion = new Stanchion();

//
// Promise way:
//

stanchion.shutdown().then(() => {
    console.log('Stanchion exited.');
});

//
// Rxjs way:
//

stanchion.shutdown$().subscribe(() => {
    console.log('Stanchion exited.');
});

When Exited

const { Stanchion } = require('stanchionjs');

const stanchion = new Stanchion();

//
// Promise way:
//

stanchion.onShutdowned(() => {
    console.log('Stanchion exited.');
});

//
// Rxjs way:
//

stanchion.onShutdowned$().subscribe(() => {
    console.log('Stanchion exited.');
});

Reference

Interfaces

ConstructOptions

Default value:

{
    // Redis configuration.
    redis: {
        host: '127.0.0.1',
        port: 6739,
    },

    // If you have lots of I/O intensive jobs, increase this may help.
    concurrency: 10,
    
    // Redis key for this queue.
    // Stanchion also support multiple keys, just use:
    //   rediskey: ['stanchion:queue1', 'stanchion:queue2'],
    redisKey: 'stanchion:queue',
    
    // How many times you want Stanchion to try reconnecting when connection is lost.
    retryAttempts: 6,

    // Optional configure, default is `true`
    shuffleKeysOnPop: true,
}

Stanchion

Stanchion#constructor

constructor(options: ConstructOptions)

Stanchion#push

push(...jobs: any[]): Promise<void>

Stanchion#push$

push$(...jobs: any[]): Observable<void>

Stanchion#getSize

getSize(): Promise<number>

Stanchion#getSize$

getSize$(): Observable<number>

Stanchion#onError

onError(handler: ErrorHandler): Subscription

Stanchion#onError$

onError$(): Subject<any>

Stanchion#process

process(processor: PromiseProcessor): Promise<void>

Stanchion#process$

process$(processor: ObservableProcessor): Observable<void>

Stanchion#shutdown

shutdown(): Promise<void>

Stanchion#shutdown$

shutdown$(): Observable<void>

Stanchion#isShutdowned

isShutdowned(): boolean

Stanchion#onShutdowned

onShutdowned(cb: VoidFunction): Subscription

Stanchion#onShutdowned$

onShutdowned$(): Observable<void>

TODOs

• Tests. Although Stanchion is bettle-tested, handling about 26K jobs per second.

• Multiple queue (Redis key) support.

• "unprocess" method.

• Key-sharding support for cluster usage.

Credits

Awesome icon by Freepik, licensed under the Creative Commons BY 3.0.

Inspired by Neamar/featureless-job-queue.