Pom/server prototype (#4)

Server prototype

.travis.yml

@@ -0,0 +1,9 @@
+language: node_js
+
+node_js:
+  - "8"
+
+script:
+- cd peer-exchange
+- yarn
+- yarn test

package.json

@@ -2,9 +2,7 @@
   "name": "snex-peering",
   "version": "0.1.0",
   "description": "A generic WebRTC library and peering server.",
-  "main": "index.js",
   "scripts": {
-    "start": "node server/server.js",
     "test": "jest"
   },
   "repository": {

peer-exchange/package.json

@@ -0,0 +1,19 @@
+{
+  "name": "@snex/peer-exchange",
+  "version": "0.1.0",
+  "description": "Peer Exchange server for WebRTC",
+  "main": "index.js",
+  "scripts": {
+    "start": "node src/server.js",
+    "test": "jest"
+  },
+  "keywords": [],
+  "author": "Pontus Alexander <pontus.alexander@gmail.com>",
+  "license": "ISC",
+  "dependencies": {
+    "ws": "^3.3.2"
+  },
+  "devDependencies": {
+    "jest": "^21.2.1"
+  }
+}

peer-exchange/src/__tests__/handler.test.js

@@ -0,0 +1,40 @@
+const {Client} = require('../models/client.js');
+const {createConnectionHandler} = require('../handler.js');
+
+describe('createConnectionHandler', () => {
+    it('listens to messages', () => {
+        const regMock = {
+            handleMessage: jest.fn(),
+            handleClose: jest.fn(),
+        };
+
+        const handler = createConnectionHandler(regMock);
+
+        const connMock = {
+            on: jest.fn(),
+        };
+
+        handler(connMock);
+
+        expect(connMock.on).toHaveBeenCalledTimes(2);
+        expect(connMock.on.mock.calls[0][0]).toBe('message');
+        expect(connMock.on.mock.calls[0][1]).toBeInstanceOf(Function);
+        expect(connMock.on.mock.calls[1][0]).toBe('close');
+        expect(connMock.on.mock.calls[1][1]).toBeInstanceOf(Function);
+
+        const messageCallback = connMock.on.mock.calls[0][1];
+        expect(regMock.handleMessage).toHaveBeenCalledTimes(0);
+        messageCallback('arbitrary message');
+        expect(regMock.handleMessage).toHaveBeenCalledTimes(1);
+        expect(regMock.handleMessage.mock.calls[0][0]).toBeInstanceOf(Client);
+        expect(regMock.handleMessage.mock.calls[0][0].conn).toBe(connMock);
+        expect(regMock.handleMessage.mock.calls[0][1]).toBe('arbitrary message');
+
+        const disconnectCallback = connMock.on.mock.calls[1][1];
+        expect(regMock.handleClose).toHaveBeenCalledTimes(0);
+        disconnectCallback();
+        expect(regMock.handleClose).toHaveBeenCalledTimes(1);
+        expect(regMock.handleClose.mock.calls[0][0]).toBeInstanceOf(Client);
+        expect(regMock.handleClose.mock.calls[0][0].conn).toBe(connMock);
+    });
+});

peer-exchange/src/handler.js

@@ -0,0 +1,19 @@
+const {Client} = require('./models/client.js');
+
+function createConnectionHandler(registry) {
+    return function handleConnection(conn) {
+        const client = new Client(conn);
+
+        conn.on('message', message => {
+            registry.handleMessage(client, message);
+        });
+
+        conn.on('close', () => {
+            registry.handleClose(client);
+        });
+    };
+}
+
+module.exports = {
+    createConnectionHandler,
+};

peer-exchange/src/models/__tests__/channel-test.js

@@ -0,0 +1,35 @@
+const {Channel} = require('../channel.js');
+const {Client} = require('../client.js');
+
+describe('Channel', () => {
+    it('when joined by two parties broadcasting sends signal to each except sender', () => {
+        const channel = new Channel();
+        const client1 = new Client();
+        const client2 = new Client();
+        client1.send = jest.fn();
+        client2.send = jest.fn();
+        channel.join(client1);
+        channel.join(client2);
+        const data = {my: 'data'};
+        channel.broadcast(client1, data);
+        expect(client1.send).toHaveBeenCalledTimes(0);
+        expect(client2.send).toHaveBeenCalledTimes(1);
+        expect(client2.send).toHaveBeenLastCalledWith(data);
+
+        const client3 = new Client();
+        client3.send = jest.fn();
+        channel.join(client3);
+        channel.broadcast(client1, data);
+        expect(client1.send).toHaveBeenCalledTimes(0);
+        expect(client2.send).toHaveBeenCalledTimes(2);
+        expect(client3.send).toHaveBeenCalledTimes(1);
+        expect(client2.send).toHaveBeenLastCalledWith(data);
+        expect(client2.send).toHaveBeenLastCalledWith(data);
+
+        channel.leave(client2);
+        channel.broadcast(client3, data);
+        expect(client1.send).toHaveBeenCalledTimes(1);
+        expect(client1.send).toHaveBeenLastCalledWith(data);
+        expect(client3.send).toHaveBeenCalledTimes(1);
+    });
+});

peer-exchange/src/models/__tests__/client-test.js

@@ -0,0 +1,14 @@
+const {ack, Client} = require('../client.js');
+
+describe('Client', () => {
+    it('send messages encoded as JSON to connection', () => {
+        const connMock = {
+            send: jest.fn(),
+        };
+
+        const client = new Client(connMock);
+        client.send({my: 'data'});
+
+        expect(connMock.send).toBeCalledWith('{"my":"data"}', ack);
+    });
+})

peer-exchange/src/models/__tests__/registry-test.js

@@ -0,0 +1,30 @@
+const {Registry} = require('../registry.js');
+const {Client} = require('../client.js');
+
+describe('Registry', () => {
+    it('broadcasts message to everyone with same id', () => {
+        const reg = new Registry();
+        const client1 = new Client();
+        const client2 = new Client();
+
+        client1.send = jest.fn();
+        client2.send = jest.fn();
+
+        reg.handleMessage(client1, JSON.stringify({channelId: 'x', type: "greet"}));
+        reg.handleMessage(client2, JSON.stringify({channelId: 'x', type: "offer"}));
+
+        expect(client1.send).toHaveBeenCalledTimes(1);
+        expect(client1.send).lastCalledWith({"channelId": "x", "type": "offer"});
+        expect(client2.send).toHaveBeenCalledTimes(0);
+    });
+
+    it('cleans up when leaving', () => {
+        const reg = new Registry();
+        const client = new Client();
+
+        reg.handleMessage(client, JSON.stringify({channelId: 'x'}));
+        reg.handleDisconnect(client);
+
+        expect(reg.channels.size).toBe(0);
+    });
+});

peer-exchange/src/models/channel.js

@@ -0,0 +1,24 @@
+class Channel
+{
+    constructor() {
+        this.clients = new Set();
+    }
+
+    broadcast(originClient, data) {
+        [...this.clients]
+            .filter(candidateClient => candidateClient !== originClient)
+            .forEach(targetClient => targetClient.send(data));
+    }
+
+    join(client) {
+        this.clients.add(client);
+    }
+
+    leave(client) {
+        this.clients.delete(client);
+    }
+}
+
+module.exports = {
+    Channel,
+};

peer-exchange/src/models/client.js

@@ -0,0 +1,22 @@
+function ack(err) {
+    if (err) {
+        console.log('Error sending message', msg, err);
+    }
+}
+
+class Client
+{
+    constructor(conn) {
+        this.conn = conn;
+    }
+
+    send(data) {
+        const msg = JSON.stringify(data);
+        this.conn.send(msg, ack);
+    }
+}
+
+module.exports = {
+    ack,
+    Client,
+};

peer-exchange/src/models/registry.js

@@ -0,0 +1,33 @@
+const {Channel} = require('./channel');
+
+class Registry {
+    constructor() {
+        this.channels = new Map();
+    }
+
+    handleDisconnect(client) {
+        this.channels.forEach((channel, id) => {
+            channel.leave(client);
+            if (channel.clients.size === 0) {
+                this.channels.delete(id);
+            }
+        });
+    }
+
+    handleMessage(client, message) {
+        const data = JSON.parse(message);
+        const {channelId} = data;
+
+        if (!this.channels.has(channelId)) {
+            this.channels.set(channelId, new Channel());
+        }
+
+        const channel = this.channels.get(channelId);
+        channel.join(client);
+        channel.broadcast(client, data);
+    }
+}
+
+module.exports = {
+    Registry,
+};

peer-exchange/src/random.js

@@ -0,0 +1,20 @@
+const crypto = require('crypto');
+
+const ID_LENGTH = 12;
+const URL_SAFE = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789';
+
+function generateRandomString(len = 6, chars = URL_SAFE) {
+  let cursor = 0;
+  return crypto.randomBytes(len).reduce((string, byte) => {
+    cursor += byte;
+    return string + chars[cursor % chars.length];
+  }, '');
+}
+
+function createId() {
+    return generateRandomString(ID_LENGTH);
+}
+
+module.exports = {
+    createId,
+};

peer-exchange/src/server.js

@@ -0,0 +1,71 @@
+const WebSocketServer = require('ws').Server;
+const Session = require('./models/session');
+const Client = require('./models/client');
+const { createId } = require('./random');
+
+const PORT = process.env.PORT || 9000;
+
+const server = new WebSocketServer({port: 9000});
+
+const sessions = new Map;
+
+function createClient(conn, id = createId()) {
+    return new Client(conn, id);
+}
+
+function createSession(id = createId()) {
+    if (sessions.has(id)) {
+        throw new Error(`Session ${id} already exists`);
+    }
+
+    const session = new Session(id);
+    console.log('Creating session', session);
+
+    sessions.set(id, session);
+
+    return session;
+}
+
+function getSession(id) {
+    return sessions.get(id);
+}
+
+server.on('connection', conn => {
+    console.log('Connection established');
+    const client = createClient(conn);
+
+    conn.on('message', msg => {
+        console.log('Message received', msg);
+        const data = JSON.parse(msg);
+
+        if (data.type === 'create-session') {
+            const session = createSession();
+            session.join(client);
+
+            client.send({
+                type: 'session-created',
+                id: session.id,
+            });
+        } else if (data.type === 'join-session') {
+            const session = getSession(data.id) || createSession(data.id);
+            session.join(client);
+        } else {
+            client.broadcast(data);
+        }
+    });
+
+    conn.on('close', () => {
+        console.log('Connection closed');
+        const session = client.session;
+        if (session) {
+            session.leave(client);
+            if (session.clients.size === 0) {
+                sessions.delete(session.id);
+            }
+        }
+
+        console.log(sessions);
+    });
+});
+
+console.log(`Running on port ${PORT}`);