Grenache Node.JS WebSocket implementation

Grenache is a micro-framework for connecting microservices. Its simple and optimized for performance.

Internally, Grenache uses Distributed Hash Tables (DHT, known from Bittorrent) for Peer to Peer connections. You can find more details how Grenche internally works at the Main Project Homepage

• Setup
• Examples
• API

Setup

Install

npm install --save grenache-nodejs-ws

Other Requirements

Install Grenache Grape: https://github.com/bitfinexcom/grenache-grape:

npm i -g grenache-grape

// Start 2 Grapes
grape --dp 20001 --aph 30001 --bn '127.0.0.1:20002'
grape --dp 20002 --aph 40001 --bn '127.0.0.1:20001'

Examples

RPC Server / Client

This RPC Server example announces a service called rpc_test on the overlay network. When a request from a client is received, it replies with world. It receives the payload hello from the client.

The client sends hello and receives world from the server.

Internally the DHT is asked for the IP of the server and then the request is done as Peer-to-Peer request via websockets.

Grape:

grape --dp 20001 --aph 30001 --bn '127.0.0.1:20002'
grape --dp 20002 --aph 40001 --bn '127.0.0.1:20001'

Server:

const link = new Link({
  grape: 'http://127.0.0.1:30001'
})
link.start()

const peer = new PeerRPCServer(link, {})
peer.init()

const service = peer.transport('server')
service.listen(_.random(1000) + 1024)

setInterval(function () {
  link.announce('rpc_test', service.port, {})
}, 1000)

service.on('request', (rid, key, payload, handler) => {
  console.log(payload) // hello
  handler.reply(null, 'world')
})

Client:

const link = new Link({
  grape: 'http://127.0.0.1:30001'
})
link.start()

const peer = new PeerRPCClient(link, {})
peer.init()

peer.request('rpc_test', 'hello', { timeout: 10000 }, (err, data) => {
  if (err) {
    console.error(err)
    process.exit(-1)
  }
  console.log(data) // world
})

Code Server Code Client

API

Class: Link

new Link(options)

• options <Object> Options for the link
  • grape <String> Address of the Grenache Grape instance. Communication is done via WebSocket or HTTP.
  • requestTimeout <Number> Default timeout for requests to Grape,
  • pingTimeout <Number> Ping connection timeout to Grape (triggers reconnect attempt),
  • lruMaxSizeLookup <Number> Maximum size of the cache, checked by applying the length function to all values in the cache
  • lruMaxAgeLookup <Number> Maximum cache age in ms.

link.start()

Sets up a connection to the DHT. Emits a connect event on successful connection.

link.stop()

Stops the connection to the DHT. Emits a disconnect event on successful disconnection.

link.announce(name)

• name <String> Name of the service, used to find the service from other peers

Used to announce a service, e.g. a RPC Server.

link.put(options)

• options
  • v: <String> value to store
• callback <function>

Puts a value into the DHT. Example.

link.get(hash, callback)

• hash <String> Hash used for lookup
• callback <function>

Retrieves a stored value from the DHT via a hash <String>. Callback returns err <Object> and data <Object>. Example.

Class: PeerRPCServer

Event: 'request'

Emitted when a request from a RPC client is received.

• rid unique request id
• key name of the service
• payload Payload sent by client
• handler Handler object, used to reply to a client.

service.on('request', (rid, key, payload, handler) => {
  handler.reply(null, 'world')
})

new PeerRPCServer(link, [options])

• link <Object> Instance of a Link Class
• options <Object>

Creates a new instance of a PeerRPCServer, which connects to the DHT using the passed link.

peer.init()

Sets the peer active. Must get called before we get a transport to set up a server.

peer.transport('server')

Must get called after the peer is active. Sets peer into server- mode.

peer.listen(port)

Lets the PeerRPCServer listen on the desired port. The port is stored in the DHT.

peer.port

Port of the server (set by listen(port)).

Example

This RPC Server example announces a service called rpc_test on the overlay network. When a request from a client is received, it replies with world. It receives the payload hello from the client.

The client sends hello and receives world from the server.

Internally the DHT is asked for the IP of the server and then the request is done as Peer-to-Peer request via websockets.

Server:

const link = new Link({
  grape: 'http://127.0.0.1:30001'
})
link.start()

const peer = new PeerRPCServer(link, {})
peer.init()

const service = peer.transport('server')
service.listen(_.random(1000) + 1024)

setInterval(function () {
  link.announce('rpc_test', service.port, {})
}, 1000)

service.on('request', (rid, key, payload, handler) => {
  console.log(payload) // hello
  handler.reply(null, 'world')
})

Client:

const link = new Link({
  grape: 'http://127.0.0.1:30001'
})
link.start()

const peer = new PeerRPCClient(link, {})
peer.init()

peer.request('rpc_test', 'hello', { timeout: 10000 }, (err, data) => {
  if (err) {
    console.error(err)
    process.exit(-1)
  }
  console.log(data) // world
})

Server Client

Class: PeerRPCClient

new PeerRPCClient(link, [options])

• link <Object> Instance of a Link Class
• options <Object>
  • maxActiveKeyDests <Number>
  • maxActiveDestTransports <Number>

Creates a new instance of a PeerRPCClient, which connects to the DHT using the passed link.

A PeerRPCClient can communicate with multiple Servers and map work items over them. With maxActiveKeyDests you can limit the maximum amount of destinations. Additionally, you can limit the amount of transports with maxActiveDestTransports.

peer.init()

Sets the peer active. Must get called before we start to make requests.

peer.map(name, payload, [options], callback)

• name <String> Name of the service to address
• payload <String> Payload to send
• options <Object> Options for the request
  • timeout <Number> timeout in ms
  • limit <Number> maximum requests per available worker
• callback <function>

Maps a number of requests over the amount of registered workers / PeerRPCServers. Example.

peer.request(name, payload, [options], callback)

• name <String> Name of the service to address
• payload <String> Payload to send
• options <Object> Options for the request
  • timeout <Number> timeout in ms
• callback <function>

Sends a single request to a RPC server/worker. Example.

Class: PeerPub

new PeerPub(link, [options])

• link <Object> Instance of a Link Class
• options <Object>

peer.init()

Sets the peer active. Must get called before we get a transport to set up a server.

peer.transport('server')

Must get called after the peer is active. Sets peer into server- mode.

peer.listen(port)

Lets the PeerRPCServer listen on the desired port. The port is stored in the DHT.

peer.pub(payload)

• payload <String> Payload to send

Sends a message to all connected peers. Example.

Class: PeerSub

new PeerSub(link, [options])

• link <Object> Instance of a Link Class
• options <Object>

Creates a new instance of a PeerSub, which connects to the DHT using the passed link.

.sub(name, [options])

• name <String> Name of the Pub Channel to register
• options <Object> Options for the request
  • timeout <Number> timeout in ms

Registers as a receiver for messages. Example.

Event: 'message'

Emitted when a payload is received.