interlinked v0.1.7

Interlinked

Rx-enabled peer-to-peer RPC for JavaScript.

No dependencies other than RxJS. Use any transport and serialization you like.

Installation

yarn add interlinked

or

npm install -S interlinked

Usage

Interlinked's API is a single function that returns an Observable.

import interlinked from 'interlinked'

const api = {
  numbers: Observable.interval(1000).take(10),
  hello: () => 'interlinked'
}

someTransport.on('connection', (socket, req) => {
  const input  = Observable.fromEvent(socket, 'message').map(JSON.parse)
  const output = x => socket.send(JSON.stringify(x))

  interlinked(input, output, api)
})

Then, on a connected peer:

• Use remote observables like they are local.
• Functions always return a promise that resolve once the remote function has a return value.

// On some other process, or machine...

interlinked(input, output).subscribe(async remote => {
  remote.numbers.subscribe(console.log)
  // 0... 1... 2... 3...
  console.log(await remote.hello())
  // interlinked
})

Here, only one side exposes an API. However, Interlinked is peer-to-peer: both sides can expose an interface of functions and observables.

API

interlinked(input: Observable, output: Function|Subject, api: Object): Observable

Returns an Observable that emits the remote interface when available.

Errors

Errors from the remote observables or promises are caught and sent to the calling peer. Observables will complete with an error, and promises will be rejected, as you'd expect.

Functions that return an Observable

Remote functions may return an observable.

interlinked(input, output, {
  countTo: n => Observable.interval(1000).take(n).map(x => x + 1)
})

On a connected peer, the function's promise will resolve to an observable that can be subscribed to as if it were local.

interlinked(input, output).subscribe(async remote => {
  const obs = await remote.countTo(3)
  obs.subscribe(console.log)
  // 1... 2... 3
})

Note: since the remote can't know when a peer will subscribe to a returned observable, all returned observables are held in memory until the peer disconnects (or when the input observable otherwise completes). Use with care.

Full Example

See the example directory for a working sample of an express server and a browser connected via websocket.

Protocol

Interlinked's protocol has some similarities to JSON-RPC, but it is not intended to be a superset.

Publishing the API

{publish: {...}}

Sent when an interlinked API is first created.

Methods

{id: txId, method: 'fn'}

{id: txId, result: data}

{id: txId, result: data, observable: true}

Observables

{id: txId, subscribe: obsId}

{unsubscribe: txId}

{complete: txId}

[txId, data]

Errors

{id: txId, error: {message: }}

TODO

• Timeouts
• More documentation
• Allow observable constructors other than RxJS 5?
• Optionally surface errors on the remote also (middleware?)