Service Adapter

Service Adapter - stream transform protocol with queue control

$ npm install service-adapter

Run tests

Browse module (e.g. node_modules/service-adapter) install directory, and run tests:

$ npm test
# or
$ node test.js

Compare test results with travis run tests.

Include in your script

const adapter = require('service-adapter');

Socket adapter type

• AaS - Adapter as Server
• AaC - Adapter as Client

Diagram example

HTTP Server          DB Server
-----------          ---------
 AaC | AaC <----------> AaS
  ^                      ^
  |        Log Server    |
  |        ----------    |
  --------> AaS | AaC <---

Network router

A socket connection can only be made between two different socket adapter type, see the diagram above. You can use the same set of functions for different adapters.

const functions = {
    func1: function(head, body) {
        console.log(this.name, 'func1 call');
        // call `AaC.func2`
        this.next('func2');
    },
    func2: function(head, body) {
        console.log(this.name, 'func2 call');
        // call `AaS.func3`
        this.next('func3');
        // optional, end client connection
        if (this.name === 'AaC') { this.client.end(); }
    },
    func3: function(head, body) {
        console.log(this.name, 'func3 call');
        // optional, close the server
        if (this.name === 'AaS') { this.server.close(); }
    }
    // ... and so on
};

require('net').createServer(function(socket) {

    // create the `AaS` adapter
    const AaS = new adapter(functions);
    // optional, adapter name
    AaS.name = 'AaS';
    // optional, create `this.server`, see `func3`
    AaS.server = this;
    // pipe `AaS` into `socket` stream
    socket.pipe(AaS).pipe(socket);

}).listen('/tmp/AaS.sock', () => {

    require('net').connect('/tmp/AaS.sock', function() {

        // create the `AaC` adapter
        const AaC = new adapter(functions);
        // optional, adapter name
        AaC.name = 'AaC';
        // optional, create `this.client`, see `func2`
        AaC.client = this;
        // pipe `AaC` into `this` client stream
        this.pipe(AaC).pipe(this);
        // start data flow
        AaC.next('func1'); // call `AaS.func1`

    });

});
// the `AaS` is listening on unix socket `/tmp/AaS.sock` and `AaC` is connecting to it
/*
Output:
------
AaS func1 call
AaC func2 call
AaS func3 call
*/

Basic router

Adapters can be use to create routers between internal app functions, without connecting to any socket. This way, you can split the code between multiple micro services (adapters) in the same application for individual maintenance and debugging.

// object functions for adapter1
const fc1 = {
    test1: (head, body) => console.log('test1 call', head, body.toString())
};
// object functions for adapter2
const fc2 = {
    test2: function(head, body) {
        console.log('test2 call', head);
        // `adapter1` is next on the pipe, after `adapter2`
        // call function `test1` from `adapter1`
        this.next('test1', head, 'back');
    }
};
// adapters
const adapter1 = new adapter(fc1);
const adapter2 = new adapter(fc2);

// create a router, data flow logic
adapter1.pipe(adapter2).pipe(adapter1);

// call function `test2` from `adapter2`
adapter2.exec('test2', '_welcome');

// `adapter2` is next on the pipe, after `adapter1`
// call function `test2` from `adapter2`
adapter1.next('test2', 'welcome_');

// `adapter2.exec` has the same result as `adapter1.next` for the router created
/*
Output:
------
test2 call _welcome
test1 call _welcome back
test2 call welcome_
test1 call welcome_ back
*/

Adapter constructor new adapter (functions[, options])

• functions - Object functions list
• options - Object {queue:Boolean, error:String, limit:Number, qumax:Number}
• queue - enable queue, default: false (disabled)
• error - error event name, default: err
• limit - limit parser bytes, default: Infinity
• qumax - max queue jobs, default: Infinity

Adapter prototype function

• exec (func[, head[, body]]) - call function func from this adapter
• next (func[, head[, body]]) - call function func from next adapter on the pipe
• head - Value, can be any type (not function)
• body - Buffer or String
• exeq () - execute all queue jobs
• delq () - delete all queue jobs
• jobs () - return all queue jobs

Function adapter ([head,[ body]])

• head - Value, can be any type (not function)
• body - Buffer

Enable queue

The adapter has an internal queue system, that is turned off by default. When queue is enabled, each adapter function execution 'adapter.exec' (and pipe(adapter) exec) is considered as a job. The queue can be enabled on adapter init new adapter(functions, {queue:true}) or later adapter.queue=true.

Attention, call this.done() at the end of each job (adapter function), in order to execute the next queue job, otherwise, if more jobs are added in queue, the queue list will increase until app is out of memory.

const functions = {
    job1: function(head, body) {
        console.log('job1 call');
        // set a task for 1s
        const t = this;
        setTimeout(() => {
            // the task is done
            console.log('job1 done');
            // exec the next job on the queue
            t.done();
        }, 1000);
    },
    job2: function(head, body) {
        console.log('job2 call');
        // add a job in the queue
        this.exec('job1');
        // exec the next job on the queue
        this.done();
    }
};

const adapter1 = new adapter(functions, { queue: false });
// add two jobs
adapter1.exec('job1').exec('job2');
/*
Output:
------
job1 call
job2 call
job1 call
job1 done
job1 done
*/

// {queue:true} - enable queue
const adapter2 = new adapter(functions, { queue: true });
// add jobs
adapter2.exec('job1').exec('job1').exec('job1');
/*
Output:
------
job1 call
job1 done
job2 call
job1 call
job1 done
*/

Disable queue

If queue is enabled on the adapter, there are two methods to disable it:

1. Safe. Async execute all queue jobs adapter.exeq() (this may take a while, depending on how many jobs are in the queue), then disable the queue adapter.queue=false.
2. Unsafe . Delete all queue jobs adapter.delq() (this is unsafe), then disable the queue adapter.queue=false.

Read queue

If queue is enabled, function adapter.jobs(), will return an array with all queue jobs. Use adapter.jobs().length to see how many jobs are in the queue, at a given time.

Error handling

The adapter emit, by default, the event named err for errors, to ensure the data flow (non-blocking state). For blocking state (no data flow), name it error on constructor options {error:'error'}, or later adapter.error='error'.

// default adapter `err` error event name
new adapter(functions, {error:'err'}).
// adapter error `err` event, non-blocking mode
on('err', e => console.log('adapter onErr', e)).
// stream standard `error` event, blocking mode
on('error', e => console.log('adapter onError', e));

For more informations, consult or run the test.js file.

Service Adapter is licensed under the MIT license. See the included LICENSE file for more details.