ReSynkd

Observable pattern through WebSockets.

Motivation

ReSynkd puts together the amazing RxJS library and the WebSocket protocol.

Although RxJS provides the WebSocketSubject class, it seems that anyone can subscribe to but also send through the same webSocketSubject, which I did not like. I wanted a way to put an Observable on the server and subscribe to it from the clients, but also vice versa.

Documentation

The typedoc documentation is available at https://cope.github.io/resynkd/docs/

Demo

See the ./demo/ folder in this repo.

Watch the demo video:

Usage

Keep in mind that the socketsSendMethod is a placeholder for whatever the send method is in the websocket implementation used, and thus this function varies. Most often it is simply socket.send, as is in the demo, but it can also be connection.socket.send or e.target.send, and so on.

Endpoint1 (Observable)

import ReSynkd from "resynkd";
const resynkd = new ReSynkd();
socket.onmessage(message => {
	let consumed = resynkd.message(message, socketsSendMethod.bind(socket));	// IMPORTANT: don't forget to bind!
	if(!consumed) console.log(message); // ...handle non-resynkd messages...
});

const subject = new Subject();
resynkd.addSubject('unique_subject_name', subject);
subject.next('value 1');
subject.next('value 25');
subject.next('value 17');
// etc.

Endpoint2 (Observer)

import ReSynkd from "resynkd";
const resynkd = new ReSynkd();
socket.onmessage(message => {
	let consumed = resynkd.message(message, socketsSendMethod.bind(socket));	// IMPORTANT: don't forget to bind!
	if(!consumed) console.log(message); // ...handle non-resynkd messages...
});

resynkd.subscribe({
	socketId: 'socket-id',					  // must be same value on both socket endpoints
	subjectId: 'unique_subject_name',		  
	send: socketsSendMethod.bind(socket),		// IMPORTANT: don't forget to bind!
	observer: {
		next: (value) => {   // * required!
			// handle value...
		},
		error: (err) => {   // * optional
			// handle error...
		},
		complete: () => {   // * optional
			// handle complete...
		}
   	}
});

The on message processing

The main part of any websocket communication is the on message event. Whether it is socket.on('message', ..., or socket.addEventListener('message', ..., in either case, the handler method receives the websocket message and it must include the resynkd processing in it.

import ReSynkd from "resynkd";
const resynkd = new ReSynkd();
let handler = (message) => {
	let consumed = resynkd.message(message, socketsSendMethod.bind(socket));	// IMPORTANT: don't forget to bind!
	if(!consumed) console.log(message); // ...handle non-resynkd messages...
}

Putting it all together, the most common example looks like below...

Server side (using fastify)

import ReSynkd from "resynkd";
const resynkd = new ReSynkd();
fastify.get('/ws', {websocket: true}, (connection, req) => {
	connection.socket.on('message', message => {
		let consumed = resynkd.message(message, connection.socket.send.bind(connection.socket));   // IMPORTANT: don't forget to bind!
		if(!consumed) console.log(message); // ...handle non-resynkd messages...
	});
});

Client side

const ReSynkd = require('resynkd');
const resynkd = new ReSynkd();

let socket = new WebSocket(YOUR_WS_URI);
socket.onmessage = (e) => {
	let { data: message } = e;
	let consumed = resynkd.message(message, e.target.send.bind(e.target));	 // IMPORTANT: don't forget to bind!
	if(!consumed) console.log(message); // ...handle non-resynkd messages...
};

TODO

See the Kanban board

• Peer-to-Peer (client-to-client) subscriptions
• Observables broadcast their subjects- so that observers know what they can subscribe to