FastMQ

High performance message broker for node.js with multiple network transports support.

Table of Contents

• FastMQ
  • Overview
  • Features
  • Installation
  • Examples
    • Simple REQUEST/RESPONSE server and client
    • Simple REQUEST/RESPONSE pattern between two clients
    • Simple PUSH/PULL pattern, one PUSH, two PULL workers
    • Simple PUBLISH/SUBSCRIBE pattern, one PUBLISH, two SUBSCRIBE channels
• API
  • FastMQ.Server.create(name)
  • FastMQ.Server.create(name, port[, host])
  • Class: FastMQ.Server
    • server.start()
    • server.stop()
    • server.request(target, topic, payload = {}, contentType = 'json')
    • server.response(topic, listener)
    • server.onError(handler)
    • server.onSocketError(handler)
  • FastMQ.Client.connect(channelName, path[, connectListener])
  • FastMQ.Client.connect(channelName, port[, host][, connectListener])
  • FastMQ.Client.connect(channelName, options[, connectListener])
  • Class: FastMQ.Channel
    • channel.onError(listener)
    • channel.onReconnect(listener)
    • channel.disconnect(graceful)
    • channel.request(target, topic, data = {}, contentType = 'json')
    • channel.response(topic, listener)
    • channel.push(target, topic, items, contentType = 'json')
    • channel.pull(topic, options, listener)
    • channel.publish(target, topic, payload, contentType = 'json')
      • channel.subscribe(topic, listener)
      • channel.getChannels(name[, type])
      • channel.watchChannels(name, callback)
  • Class: FastMQ.Message
    • message.header
    • message.payload
    • message.setError(code)
    • message.isError(value)
  • Class: FastMQ.Response
    • response.send(payload, contentType)
  • List of Error Codes

Overview

FastMQ is a node.js based message broker aims to let programmer easy to commuicate between different processes or machines. It is designed for high performance which can achieve to over 30000 message delivery per second with 64KB message payload, and with small size header overhead.

It support both local socket(Unix domain socket and Windows pipe) for local process communication, and also supports reliable TCP connections between machines.

Which makes FastMQ suitable as backend service for IPC, clusters, and micro service applications.

Features

• Pure javascript and asynchronous message broker, compatiable with node.js >= 4.x
• Easy to communicate between proceeses and machines
• All major functions support Promise for better async. programming
• Small extra size overhead with packed binary header
• Support various of payload formats: JSON, Raw Binary and Plain Text
• Support REQUEST/RESPONSE pattern for executing asynchronous Remote Procedure Call (RPC)
• Support PUSH/PULL pattern for distributing large number of tasks among workers
• Support PUBLISH/SUBSCRIBE pattern for sending messages to many consumers at once
• Support Unix domain socket / Windows pipe and TCP socket
• GLOB topic expression to specify message routing

Installation

Install FastMQ via npm or yarn

npm install fastmq

yarn add fastmq

Examples

Simple REQUEST/RESPONSE server and client

Server.js: Simple server handle 'test_cmd' topic request

const FastMQ = require('fastmq');

// Create message broker server with 'master' channel name
const server = FastMQ.Server.create('master');

// Register topic: 'test_cmd', receive message and response back to client requester
server.response('test_cmd', (msg, res) => {
    console.log('Server receive request payload:', msg.payload);
    // echo request data back;
    let resData = {data: msg.payload.data};
    res.send(resData, 'json');
});

// start server
server.start().then(() => {
    console.log('Message Broker server started');
});

Simple client send 'test_cmd' topic to server('master' channel)

const FastMQ = require('fastmq');

var requestChannel;
// create a client with 'requestChannel' channel name and connect to server.
FastMQ.Client.connect('requestChannel', 'master')
.then((channel) => {
    // client connected
    requestChannel = channel;

    // send request to 'master' channel  with topic 'test_cmd' and JSON format payload.
    let reqPayload = {data: 'reqeust data'};
    return requestChannel.request('master', 'test_cmd', reqPayload, 'json');
})
.then((result) => {
    console.log('Got response from master, data:' + result.payload.data);
    // client channel disconnect
    requestChannel.disconnect();
})
.catch((err) => {
    console.log('Got error:', err.stack);
});

Simple REQUEST/RESPONSE pattern between two clients

Client1.js: Connect to server, register 'responseChannel' channel, and handle 'test_cmd' topic

const FastMQ = require('fastmq');

var responseChannel;
// create a client with 'requestChannel' channel name and connect to server.
FastMQ.Client.connect('responseChannel', 'master')
.then((channel) => {
    // client connected
    responseChannel = channel;
    responseChannel.response('test_cmd', (msg, res) => {
        console.log('Receive request payload:', msg.payload);
        // echo request data back;
        let resData = {data: msg.payload.data};
        res.send(resData, 'json');
    });

})
.catch((err) => {
    console.log('Got error:', err.stack);
});

Client2.js: Connect to server, register 'requestChannel' channel, and send 'test_cmd' request

const FastMQ = require('fastmq');

var requestChannel;
// create a client with 'requestChannel' channel name and connect to server.
FastMQ.Client.connect('requestChannel', 'master')
.then((channel) => {
    // client connected
    requestChannel = channel;
    let reqPayload = {data: 'reqeust data'};

    // send request to 'responseChannel' channel  with topic 'test_cmd' and JSON format payload.
    return requestChannel.request('responseChannel', 'test_cmd', reqPayload, 'json');
})
.then((result) => {
    console.log('Got response from master, data:' + result.payload.data);
    // client channel disconnect
    requestChannel.disconnect();
})
.catch((err) => {
    console.log('Got error:', err.stack);
});

Simple PUSH/PULL pattern, one PUSH, two PULL workers

push_client.js: Push 'add_job' message to all channels match 'worker*' GLOB expression

const FastMQ = require('fastmq');

var pushChannel;
// create a client with 'pushChannel' channel name and connect to server.
FastMQ.Client.connect('pushChannel', 'master')
.then((channel) => {
    pushChannel = channel;

    // generate 10 tasks
    var data = [];
    for (let i = 1; i <= 10; i++)
        data.push({id: 'job' + i});

    // push tasks to 'worker*' channels
    return pushChannel.push('worker*', 'add_job', data);
})
.then(() => {
    console.log('Push 10 tasks to worker*:add_job');
    pushChannel.disconnect();
})
.catch((err) => {
    console.log('Got error:', err.stack);
});

worker1.js: Pull 'add_job'

const FastMQ = require('fastmq');

var workerChannel;
// create a client with 'worker1' channel name and connect to server.
FastMQ.Client.connect('worker1', 'master')
.then((channel) => {
    workerChannel = channel;

    // handle 'add_job' push message
    workerChannel.pull('add_job', (msg) => {
        console.log('Worker1 receive job:', msg.payload.id);
    });
})
.catch((err) => {
    console.log('Got error:', err.stack);
});

worker2.js: Pull 'add_job'

const FastMQ = require('fastmq');

var workerChannel;
// create a client with 'worker2' channel name and connect to server.
FastMQ.Client.connect('worker2', 'master')
.then((channel) => {
    workerChannel = channel;

    // handle 'add_job' push message
    workerChannel.pull('add_job', (msg) => {
        console.log('Worker2 receive job:', msg.payload.id);
    });
})
.catch((err) => {
    console.log('Got error:', err.stack);
});

Simple PUBLISH/SUBSCRIBE pattern, one PUBLISH, two SUBSCRIBE channels

publish.js: Publish 'log' message to to all channels match 'console.*' GLOB expression

const FastMQ = require('fastmq');

var pubChannel;
FastMQ.Client.connect('publisher', 'master')
.then((channel) => {
    pubChannel = channel;
    return pubChannel.publish('console.*', 'log', {data: 'a example log'}, 'json');
})
.then(() => {
    console.log('Push log to console.*');
    pubChannel.disconnect();
})
.catch((err) => {
    console.log('Got error:', err.stack);
});

console1.js: 'console.1' channel, subscribe 'log' topic

const FastMQ = require('fastmq');

var subChannel;
FastMQ.Client.connect('console.1', 'master')
.then((channel) => {
    su