RPC JSON - Stream transform protocol

RPC JSON - Server and Client Stream with JSON header and RAW body

$ npm install rpc-json

Run tests

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

$ npm test
# or
$ node test.js

Compare test results with travis run tests.

Include in your script

const rpc = require('rpc-json');

Define custom server async query function

async function query(response, head, body) {
    console.log('client-request', head, body.toString());
    // response back to client
    return await response('s-' + head, body);
}

Simple client-server stream pipe

const server = new rpc.server(query); // using custom 'query' request function
const client = new rpc.client;

// pipe: client (request to:) > server (response back to:) > client
client.pipe(server).pipe(client);

Promise exec calls

client.exec('head1', 'body1').
then(r1 => {
    console.log('log1', r1);
    return client.exec('head2', 'body2');
}).
then(r2 => {
    console.log('log2', r2);
    return client.exec('head3', 'body3');
}).
then(r3 => {
    console.log('log3', r3);
}).
catch(console.error);

/* Output
---------
client-request head1 body1
log1 { head: 's-head1', body: <Buffer 62 6f 64 79 31> }
client-request head2 body2
log2 { head: 's-head2', body: <Buffer 62 6f 64 79 32> }
client-request head3 body3
log3 { head: 's-head3', body: <Buffer 62 6f 64 79 33> }
*/

async/await exec calls

(async () => {
    const r1 = await client.exec('head1', 'body1');
    console.log('log1', r1);
    const r2 = await client.exec('head2', 'body2');
    console.log('log2', r2);
    const r3 = await client.exec('head3', 'body3');
    console.log('log3', r3);
})().catch(console.error);

/* Output
---------
client-request head1 body1
log1 { head: 's-head1', body: <Buffer 62 6f 64 79 31> }
client-request head2 body2
log2 { head: 's-head2', body: <Buffer 62 6f 64 79 32> }
client-request head3 body3
log3 { head: 's-head3', body: <Buffer 62 6f 64 79 33> }
*/

Simple client-server socket stream pipe

const net = require('net');

const srv = net.createServer(socket => { // on client connect
    socket.pipe(new rpc.server).pipe(socket); // create new 'rpc.server' object here, to reset data flow on each client
}).listen(function() { // server listen
    const a = this.address(); // get the server port and address
    net.connect(a.port, a.address, function() { // on client connect
        const cli = new rpc.client; // create new 'rpc.client' object here, to reset data flow on each client
        this.pipe(cli).pipe(this); // attach client to the server connection
        cli.exec('head', 'body'). // exec call
        then(r => { // response from server
            console.log('log', r);
            cli.push(null); // optional, end client connection
            srv.close(); // optional, close the socket server
        }).catch(console.error);
    }).on('end', () => console.log('socket client end'));
}).on('close', () => console.log('socket server close'));

/* Output
---------
log { head: 'head', body: <Buffer 62 6f 64 79> }
socket server close
socket client end
*/

Server async function request (<Promise> response, head, body)

• head - Value, can be any type (not a function) - deserialized with JSON
• body - Buffer or String
• this - Bind Server Object
• return - <Promise> response (head, body) - callback server response

Default server anonymous async request function will response back to client with the same request head and body values, like this: async (response, head, body) => await response(head, body)

Client Promise function exec (head, body)

• head - Value, can be any type (not a function)
• body - Buffer or String
• this - Bind Client Object
• return - Promise.resolve( { head, body } ) - body is Buffer

Custom stream error event names

• serverError - error event name for rpc.server
• clientError - error event name for rpc.client

server.on('serverError', e => console.log('onServerError', e));
client.on('clientError', e => console.log('onClientError', e));

For more info consult or run the test.js file.

RPC JSON is licensed under the MIT license. See the included LICENSE file for more details.