Repository: boardgameio/boardgame.io
Pull request #978 (2021-08-07 23:22:50)

Adds pub-sub support for horizontally scaling bgio server
Usage:
```
import { Server, Origins, SocketIO, RedisPubSub } from 'boardgame.io/server';
import TicTacToe from './src/tic-tac-toe/game';
import Chess from './src/chess/game';
import redis from 'redis';
import { PostgresStore } from 'bgio-postgres';
 
const PORT = process.env.PORT || 8000;
const redisConfig = { host: '192.168.8.50' };
const pubClient = redis.createClient(redisConfig);
const subClient = redis.createClient(redisConfig);
const server = Server({
  games: [TicTacToe, Chess],
  db: new PostgresStore('vdf@192.168.8.50/vdf'),
  origins: [Origins.LOCALHOST],
  transport: new SocketIO({
    pubSub: new RedisPubSub(pubClient, subClient),
  }),
});
server.run(PORT, () => {
  console.log(`Serving at: http://localhost:${PORT}`);
});
```

Demo:
https://youtu.be/4BneuOrfs-0 

#### Checklist

- [x] Use a separate branch in your local repo (not `main`).
- [x] Test coverage is 100% (or you have a story for why it's ok).


vdfdev: Don't worry about it, I didn't mind. Good to keep my over-engineering in check sometimes hahaha
vdfdev: All checks have passed, I lost your approval but because I only did what you asked I will merge anyway

Patch:
diff --git a/packages/server.ts b/packages/server.ts
index e179a85..7b574b5 100644
--- a/packages/server.ts
+++ b/packages/server.ts
@@ -10,5 +10,6 @@ import { Server } from '../src/server';
 import { Origins } from '../src/server/cors';
 import { FlatFile } from '../src/server/db';
 import { SocketIO } from '../src/server/transport/socketio';
+import { GenericPubSub } from '../src/server/transport/pubsub/generic-pub-sub';
 
-export { Server, Origins, FlatFile, SocketIO };
+export { Server, Origins, FlatFile, SocketIO, GenericPubSub };
diff --git a/src/server/transport/pubsub/generic-pub-sub.ts b/src/server/transport/pubsub/generic-pub-sub.ts
new file mode 100644
index 0000000..c70b161
--- /dev/null
+++ b/src/server/transport/pubsub/generic-pub-sub.ts
@@ -0,0 +1,11 @@
+/** Generic interface for pub-subs.  */
+export interface GenericPubSub<T> {
+  // Publish an event for a match.
+  publish(channelId: string, payload: T);
+
+  // Subscribes to events related to a single match.
+  subscribe(channelId: string, callback: (payload: T) => void): void;
+
+  // Cleans up subscription for a given channel.
+  unsubscribeAll(channelId: string);
+}
diff --git a/src/server/transport/pubsub/in-memory-pub-sub.ts b/src/server/transport/pubsub/in-memory-pub-sub.ts
new file mode 100644
index 0000000..85c6cdb
--- /dev/null
+++ b/src/server/transport/pubsub/in-memory-pub-sub.ts
@@ -0,0 +1,28 @@
+import type { GenericPubSub } from './generic-pub-sub';
+
+export class InMemoryPubSub<T> implements GenericPubSub<T> {
+  callbacks: Map<string, ((payload: T) => void)[]> = new Map();
+
+  publish(channelId: string, payload: T) {
+    if (!this.callbacks.has(channelId)) {
+      return;
+    }
+    const allCallbacks = this.callbacks.get(channelId);
+    for (const callback of allCallbacks) {
+      callback(payload);
+    }
+  }
+
+  subscribe(channelId: string, callback: (payload: T) => void): void {
+    if (!this.callbacks.has(channelId)) {
+      this.callbacks.set(channelId, []);
+    }
+    this.callbacks.get(channelId).push(callback);
+  }
+
+  unsubscribeAll(channelId: string) {
+    if (this.callbacks.has(channelId)) {
+      this.callbacks.delete(channelId);
+    }
+  }
+}
diff --git a/src/server/transport/socketio.ts b/src/server/transport/socketio.ts
index a2724cc..6a916a6 100644
--- a/src/server/transport/socketio.ts
+++ b/src/server/transport/socketio.ts
@@ -17,76 +17,50 @@ import type {
 } from '../../master/master';
 import { getFilterPlayerView } from '../../master/filter-player-view';
 import type { Game, Server } from '../../types';
+import type { GenericPubSub } from './pubsub/generic-pub-sub';
+import type { IntermediateTransportData } from '../../master/master';
+import { InMemoryPubSub } from './pubsub/in-memory-pub-sub';
 
 const PING_TIMEOUT = 20 * 1e3;
 const PING_INTERVAL = 10 * 1e3;
 
+const emit = (socket: IOTypes.Socket, { type, args }: TransportData) => {
+  socket.emit(type, ...args);
+};
+
+function getPubSubChannelId(matchID: string): string {
+  return `MATCH-${matchID}`;
+}
+
 /**
  * API that's exposed by SocketIO for the Master to send
  * information to the clients.
  */
-export function TransportAPI(
+export const TransportAPI = (
   matchID: string,
   socket: IOTypes.Socket,
-  clientInfo: SocketIO['clientInfo'],
-  roomInfo: SocketIO['roomInfo'],
-  filterPlayerView: (string, IntermediateTransportData) => TransportData,
-  provisionalClient?: Client
-): MasterTransport {
-  /**
-   * Emit a socket.io event to the recipientID.
-   * If the recipient is the current socket, uses a basic emit, otherwise
-   * emits via the current socket’s `to` method.
-   */
-  const emit = (recipientID: string, { type, args }: TransportData) => {
-    const emitter = socket.id === recipientID ? socket : socket.to(recipientID);
-    emitter.emit(type, ...args);
-  };
-
-  /**
-   * Run a callback for each registered client for this match, including
-   * this transport’s provisionalClient if provided.
-   */
-  const forEachClient = (clientCallback: (client: Client) => void) => {
-    const clientIDs = roomInfo.get(matchID);
-    if (clientIDs) {
-      clientIDs.forEach((clientID) => {
-        const client = clientInfo.get(clientID);
-        clientCallback(client);
-      });
-    }
-    if (provisionalClient) {
-      clientCallback(provisionalClient);
-    }
-  };
-
-  /**
-   * Send a message to a specific client.
-   */
+  filterPlayerView: any,
+  pubSub: GenericPubSub<IntermediateTransportData>
+): MasterTransport => {
   const send: MasterTransport['send'] = ({ playerID, ...data }) => {
-    forEachClient((client) => {
-      if (client.playerID === playerID)
-        emit(client.socket.id, filterPlayerView(playerID, data));
-    });
+    emit(socket, filterPlayerView(playerID, data));
   };
 
   /**
    * Send a message to all clients.
    */
   const sendAll: MasterTransport['sendAll'] = (payload) => {
-    forEachClient((client) => {
-      const data = filterPlayerView(client.playerID, payload);
-      emit(client.socket.id, data);
-    });
+    pubSub.publish(getPubSubChannelId(matchID), payload);
   };
 
   return { send, sendAll };
-}
+};
 
 export interface SocketOpts {
   https?: HttpsOptions;
   socketOpts?: IOTypes.ServerOptions;
   socketAdapter?: any;
+  pubSub?: GenericPubSub<IntermediateTransportData>;
 }
 
 interface Client {
@@ -106,14 +80,16 @@ export class SocketIO {
   private readonly https: HttpsOptions;
   private readonly socketAdapter: any;
   private readonly socketOpts: IOTypes.ServerOptions;
+  protected pubSub: GenericPubSub<IntermediateTransportData>;
 
-  constructor({ https, socketAdapter, socketOpts }: SocketOpts = {}) {
+  constructor({ https, socketAdapter, socketOpts, pubSub }: SocketOpts = {}) {
     this.clientInfo = new Map();
     this.roomInfo = new Map();
     this.perMatchQueue = new Map();
     this.https = https;
     this.socketAdapter = socketAdapter;
     this.socketOpts = socketOpts;
+    this.pubSub = pubSub || new InMemoryPubSub();
   }
 
   /**
@@ -129,6 +105,7 @@ export class SocketIO {
     matchClients.delete(socketID);
     // If the match is now empty, delete its promise queue & client ID list.
     if (matchClients.size === 0) {
+      this.unsubscribePubSubChannel(matchID);
       this.roomInfo.delete(matchID);
       this.deleteMatchQueue(matchID);
     }
@@ -139,11 +116,12 @@ export class SocketIO {
   /**
    * Register client data for a socket.
    */
-  private addClient(client: Client): void {
+  private addClient(client: Client, game: Game): void {
     const { matchID, socket } = client;
     // Add client to list of connected sockets for this match.
     let matchClients = this.roomInfo.get(matchID);
     if (matchClients === undefined) {
+      this.subscribePubSubChannel(matchID, game);
       matchClients = new Set<string>();
       this.roomInfo.set(matchID, matchClients);
     }
@@ -152,6 +130,23 @@ export class SocketIO {
     this.clientInfo.set(socket.id, client);
   }
 
+  private subscribePubSubChannel(matchID: string, game: Game) {
+    const filterPlayerView = getFilterPlayerView(game);
+    const broadcast = (payload: IntermediateTransportData) => {
+      this.roomInfo.get(matchID).forEach((clientID) => {
+        const client = this.clientInfo.get(clientID);
+        const data = filterPlayerView(client.playerID, payload);
+        emit(client.socket, data);
+      });
+    };
+
+    this.pubSub.subscribe(getPubSubChannelId(matchID), broadcast);
+  }
+
+  private unsubscribePubSubChannel(matchID: string) {
+    this.pubSub.unsubscribeAll(getPubSubChannelId(matchID));
+  }
+
   init(
     app: Server.App & { _io?: IOTypes.Server },
     games: Game[],
@@ -185,13 +180,7 @@ export class SocketIO {
           const master = new Master(
             game,
             app.context.db,
-            TransportAPI(
-              matchID,
-              socket,
-              this.clientInfo,
-              this.roomInfo,
-              filterPlayerView
-            ),
+            TransportAPI(matchID, socket, filterPlayerView, this.pubSub),
             app.context.auth
           );
 
@@ -205,15 +194,12 @@ export class SocketIO {
           const [matchID, playerID, credentials] = args;
           socket.join(matchID);
           this.removeClient(socket.id);
-
-          const client = { socket, matchID, playerID, credentials };
+          const requestingClient = { socket, matchID, playerID, credentials };
           const transport = TransportAPI(
             matchID,
             socket,
-            this.clientInfo,
-            this.roomInfo,
             filterPlayerView,
-            client
+            this.pubSub
           );
           const master = new Master(
             game,
@@ -226,8 +212,8 @@ export class SocketIO {
           if (syncResponse && syncResponse.error === 'unauthorized') {
             return;
           }
+          this.addClient(requestingClient, game);
           await master.onConnectionChange(matchID, playerID, credentials, true);
-          this.addClient(client);
         });
 
         socket.on('disconnect', async () => {
@@ -238,13 +224,7 @@ export class SocketIO {
             const master = new Master(
               game,
               app.context.db,
-              TransportAPI(
-                matchID,
-                socket,
-                this.clientInfo,
-                this.roomInfo,
-                filterPlayerView
-              ),
+              TransportAPI(matchID, socket, filterPlayerView, this.pubSub),
               app.context.auth
             );
             await master.onConnectionChange(
@@ -263,13 +243,7 @@ export class SocketIO {
             const master = new Master(
               game,
               app.context.db,
-              TransportAPI(
-                matchID,
-                socket,
-                this.clientInfo,
-                this.roomInfo,
-                filterPlayerView
-              ),
+              TransportAPI(matchID, socket, filterPlayerView, this.pubSub),
               app.context.auth
             );
             master.onChatMessage(...args);


Test patch:
diff --git a/src/server/transport/pubsub/in-memory-pub-sub.test.ts b/src/server/transport/pubsub/in-memory-pub-sub.test.ts
new file mode 100644
index 0000000..25e6eee
--- /dev/null
+++ b/src/server/transport/pubsub/in-memory-pub-sub.test.ts
@@ -0,0 +1,47 @@
+import { InMemoryPubSub } from './in-memory-pub-sub';
+
+const CHANNEL_FOO = 'foo';
+
+describe('in-memory pubsub', () => {
+  it('should receive message from subscription', () => {
+    const pubSub = new InMemoryPubSub<string>();
+    const callback = jest.fn();
+    pubSub.subscribe(CHANNEL_FOO, callback);
+    const payload = 'hello world';
+    pubSub.publish(CHANNEL_FOO, payload);
+    expect(callback).toHaveBeenCalledWith(payload);
+  });
+
+  it('should receive message from two subscriptions', () => {
+    const pubSub = new InMemoryPubSub<string>();
+    const callback1 = jest.fn();
+    const callback2 = jest.fn();
+    pubSub.subscribe(CHANNEL_FOO, callback1);
+    pubSub.subscribe(CHANNEL_FOO, callback2);
+    const payload = 'hello world';
+    pubSub.publish(CHANNEL_FOO, payload);
+    expect(callback1).toHaveBeenCalledWith(payload);
+    expect(callback2).toHaveBeenCalledWith(payload);
+  });
+
+  it('should unsubscribe', () => {
+    const pubSub = new InMemoryPubSub<string>();
+    const callback = jest.fn();
+    pubSub.subscribe(CHANNEL_FOO, callback);
+    pubSub.unsubscribeAll(CHANNEL_FOO);
+    const payload = 'hello world';
+    pubSub.publish(CHANNEL_FOO, payload);
+    expect(callback).not.toHaveBeenCalled();
+  });
+
+  it('should ignore extra unsubscribe', () => {
+    const pubSub = new InMemoryPubSub<string>();
+    const callback = jest.fn();
+    pubSub.subscribe(CHANNEL_FOO, callback);
+    pubSub.unsubscribeAll(CHANNEL_FOO);
+    pubSub.unsubscribeAll(CHANNEL_FOO); // do nothing
+    const payload = 'hello world';
+    pubSub.publish(CHANNEL_FOO, payload);
+    expect(callback).not.toHaveBeenCalled();
+  });
+});
diff --git a/src/server/transport/socketio.test.ts b/src/server/transport/socketio.test.ts
index c974155..b70dc8c 100644
--- a/src/server/transport/socketio.test.ts
+++ b/src/server/transport/socketio.test.ts
@@ -36,6 +36,10 @@ class SocketIOTestAdapter extends SocketIO {
     this.clientInfo = clientInfo;
     this.roomInfo = roomInfo;
   }
+
+  public getPubSub() {
+    return this.pubSub;
+  }
 }
 
 jest.mock('../../master/master', () => {
@@ -156,6 +160,7 @@ describe('socketAdapter', () => {
 describe('TransportAPI', () => {
   let io;
   let api;
+  const matchID = 'matchID';
 
   beforeAll(() => {
     const auth = new Auth({ authenticateCredentials: () => true });
@@ -166,23 +171,23 @@ describe('TransportAPI', () => {
     const transport = new SocketIOTestAdapter({ clientInfo, roomInfo });
     transport.init(app, games);
     io = app.context.io;
+    const socket = io.socket;
     const filterPlayerView = getFilterPlayerView(games[0]);
     api = TransportAPI(
-      'matchID',
-      io.socket,
-      clientInfo,
-      roomInfo,
-      filterPlayerView
+      matchID,
+      socket,
+      filterPlayerView,
+      transport.getPubSub()
     );
   });
 
   beforeEach(async () => {
     io.socket.emit = jest.fn();
     io.socket.id = '0';
-    const args0: SyncArgs = ['matchID', '0', undefined, 2];
+    const args0: SyncArgs = [matchID, '0', undefined];
     await io.socket.receive('sync', ...args0);
     io.socket.id = '1';
-    const args1: SyncArgs = ['matchID', '1', undefined, 2];
+    const args1: SyncArgs = [matchID, '1', undefined];
     await io.socket.receive('sync', ...args1);
   });
 
@@ -192,12 +197,6 @@ describe('TransportAPI', () => {
     expect(io.socket.emit).toHaveBeenCalledWith('A');
   });
 
-  test('send to another player', () => {
-    io.socket.id = '0';
-    api.send({ type: 'A', playerID: '1', args: [] });
-    expect(io.socket.emit).toHaveBeenCalledWith('A');
-  });
-
   test('sendAll - function', () => {
     api.sendAll({ type: 'A', args: [] });
     expect(io.socket.emit).toHaveBeenCalledTimes(2);