RxWS SocketIO

Reactive Web Sockets on top of SocketIO

Status:

RxWS is a RESTful reactive JavaScript implementation on top of web sockets. This includes, GET, POST, PUT, REMOVE (DELETE), PATCH, and HEAD. RxWS guarantees message delivery by generating a correlation id for each message (to and from the server). Both the server and client automatically send an acknowledgement response for each request. If there is no acknowledgement after a timeout, an error is thrown.

RxWS implements a RESTful protocol. You can use any web socket server as long as it implements the same protocol.

Client library

The client library is available at https://github.com/CanopyTax/rxws

Setup

npm install --save rxws-socketio socket.io

Example

// Server
import io from 'socket.io';
import rxws from 'rxws-socketio';

let server = io(3000);

function rxHandler(rxSocket) {
	rxSocket.get('questions')
		.subscribe(({req, res, next}) => {
			res.send([
				{
					id: 1,
					name: 'userName',
					label: 'What is your name?'
				}
			])
		});
}

server.on('connection', function(socket) {
	rxHandler(rxws(socket));
});

// Client
import rxws from 'rxws';
import SocketIOBackend from 'rxws-socketio/lib/SocketIOBackend';

rxws.setBackend({
	backend: SocketIOBackend.default,
	url: 'http://localhost:3000'
});

setInterval(function() {
	rxws.get('questions')
		.subscribe(function(resp) {
			console.log(resp.questions);
		});
}, 1000);

API

/**
 * Middleware handler - The middleware is executed in the order that it is defined.
 *
 * @param {object} options An options object passed to the middleware
 * @return {Observable} An obserable with the request and response object which may
 *                be manipulated.
 */
rxSocket.use(options)
	.subscribe(({req, res, next}) => {
		req.requestTime = Date.now();
		next();
	});

/**
 * Error middleware handler
 *
 * @param {object} options An options object. Unused at the moment.
 * @return {Observable} Observable is passed an object with an error object, the
 *                 request, response, and next method. Calling next will route
 *                 to the next error handling middleware.
 */
rxSocket.useError(options)
	.subscribe(({err, req, res, next}) => {
		console.error(err.stack);
		res.status(500).send('Something broke!');
	});

/**
 * RESTful methods are available on rxSocket
 *
 * @param {string} Resource a dot separated path to the resource being requested.
 *                 The only available parameters for a given resource are parameters
 *                 that match a given resource. For example, the resource 'questions'
 *                 can only take a 'questions' parameter. Anything else will error.
 *                 Nested resources need to have each resource represented with a
 *                 parameter except for the last resource, which is optional.
 * @return {Observable} The observable can only be passed a single value, that value
 *                 must be destructured to get both the request and response.
 */
rxSocket.get('questions')
	.subscribe(({req, res, next}) => {
		/** Req API **/
		req.headers
		req.resource
		req.method
		req.parameters
		req.query
		req.body

		/** Res API **/
		res.headers
		res.body
		res.statusCode = 404;
		res.setHeader('name', 'value');
		res.getHeader('name');
		res.removeHeader('name');
		res.send({data: ""});

		res.status(500).send('Something broke!');
	})

/** RestFul Listeners **/
rxSocket.put('questions').subscribe(({req, res, next}) => {});
rxSocket.post('questions').subscribe(({req, res, next}) => {});
rxSocket.patch('questions').subscribe(({req, res, next}) => {});
rxSocket.delete('questions').subscribe(({req, res, next}) => {});
rxSocket.head('questions').subscribe(({req, res, next}) => {});

/**
 * Server sent messages - possible other verbs? send, emit, write
 *
 * @param {string} messageKey A unique message key which the client can subscribe to
 * @param {any} data An arbitrary data value sent to the client
 */
rxSocket.push('newPost', {some: 'data'});