Whisper through redis

src/channels/channel.ts

@@ -1,6 +1,6 @@
-import { PresenceChannel } from './presence-channel';
-import { PrivateChannel } from './private-channel';
-import { Log } from './../log';
+import {PresenceChannel} from './presence-channel';
+import {PrivateChannel} from './private-channel';
+import {Log} from './../log';
 
 export class Channel {
     /**
@@ -63,9 +63,7 @@ export class Channel {
             if (this.isClientEvent(data.event) &&
                 this.isPrivate(data.channel) &&
                 this.isInChannel(socket, data.channel)) {
-                this.io.sockets.connected[socket.id]
-                    .broadcast.to(data.channel)
-                    .emit(data.event, data.channel, data.data);
+                this.presence.clientEvent({socket, ...data});
             }
         }
     }

src/channels/presence-channel.ts

@@ -1,5 +1,8 @@
-import { Database } from './../database';
-import { Log } from './../log';
+import {Database} from './../database';
+import {Log} from './../log';
+import {Publisher} from "../publishers";
+import {PublisherFactory} from "../publishers/publisher-factory";
+
 var _ = require("lodash");
 
 export class PresenceChannel {
@@ -8,11 +11,14 @@ export class PresenceChannel {
      */
     db: Database;
 
+    publisher: Publisher;
+
     /**
      * Create a new Presence channel instance.
      */
     constructor(private io, private options: any) {
         this.db = new Database(options);
+        this.publisher = new PublisherFactory(io).create(options);
     }
 
     /**
@@ -141,16 +147,18 @@ export class PresenceChannel {
      * On join event handler.
      */
     onJoin(socket: any, channel: string, member: any): void {
-        this.io.sockets.connected[socket.id].broadcast
-            .to(channel)
-            .emit("presence:joining", channel, member);
+        this.publisher.publish(channel, "presence:joining",{
+            data: {member}
+        });
     }
 
     /**
      * On leave emitter.
      */
     onLeave(channel: string, member: any): void {
-        this.io.to(channel).emit("presence:leaving", channel, member);
+        this.publisher.publish(channel, "presence:leaving", {
+            data: {member}
+        });
     }
 
     /**
@@ -159,4 +167,8 @@ export class PresenceChannel {
     onSubscribed(socket: any, channel: string, members: any[]) {
         this.io.to(socket.id).emit("presence:subscribed", channel, members);
     }
+
+    clientEvent(data) {
+        this.publisher.publish(data.channel, data.event, data);
+    }
 }

src/channels/private-channel.ts

@@ -1,6 +1,5 @@
 let request = require('request');
 let url = require('url');
-import { Channel } from './channel';
 import { Log } from './../log';
 
 export class PrivateChannel {

src/echo-server.ts

@@ -1,11 +1,11 @@
-import { HttpSubscriber, RedisSubscriber, Subscriber } from './subscribers';
-import { Channel } from './channels';
-import { Server } from './server';
-import { HttpApi } from './api';
-import { Log } from './log';
-import * as fs from 'fs';
+import {HttpSubscriber, RedisSubscriber, Subscriber} from './subscribers';
+import {Channel} from './channels';
+import {Server} from './server';
+import {HttpApi} from './api';
+import {Log} from './log';
+
 const packageFile = require('../package.json');
-const { constants } = require('crypto');
+const {constants} = require('crypto');
 
 /**
  * Echo server class.
@@ -75,7 +75,8 @@ export class EchoServer {
     /**
      * Create a new instance.
      */
-    constructor() { }
+    constructor() {
+    }
 
     /**
      * Start the Echo Server.

src/publishers/index.ts

@@ -0,0 +1,2 @@
+export * from './redis-publisher';
+export * from './publisher';

src/publishers/io-publisher.ts

@@ -0,0 +1,35 @@
+import {Publisher} from "./publisher";
+
+export class IoPublisher implements Publisher {
+    constructor(private io) {
+    }
+
+    publish(channel: string, event: string, data: any): Promise<any> {
+        if (event === "presence:leaving") {
+            this.io
+                .to(channel)
+                .emit(event, data.data.member);
+
+            return Promise.resolve(undefined);
+        }
+
+        if (event === "presence:joining") {
+            this.io
+                .sockets
+                .connected[data.socket.id]
+                .broadcast
+                .to(channel)
+                .emit(event, data.data.member);
+
+            return Promise.resolve(undefined);
+        }
+
+        this.io.sockets
+            .connected[data.socket.id]
+            .broadcast
+            .to(channel)
+            .emit(event, channel, data.data);
+
+        return Promise.resolve(undefined);
+    }
+}

src/publishers/publisher-factory.ts

@@ -0,0 +1,14 @@
+import {Publisher} from "./publisher";
+import {RedisPublisher} from "./redis-publisher";
+import {IoPublisher} from "./io-publisher";
+
+export class PublisherFactory {
+    public constructor(private io) { }
+
+    public create(options: any): Publisher {
+        if (options.subscribers.redis) {
+            return new RedisPublisher(options);
+        }
+        return new IoPublisher(this.io);
+    }
+}

src/publishers/publisher.ts

@@ -0,0 +1,3 @@
+export interface Publisher {
+    publish(channel: string, event: string, data: any): Promise<any>;
+}

src/publishers/redis-publisher.ts

@@ -0,0 +1,47 @@
+var Redis = require('ioredis');
+import {Publisher} from './publisher';
+
+var _ = require("lodash");
+
+export class RedisPublisher implements Publisher {
+    /**
+     * Redis pub/sub client.
+     *
+     * @type {object}
+     */
+    private _redis: any;
+
+    /**
+     *
+     * KeyPrefix for used in the redis Connection
+     *
+     * @type {String}
+     */
+    private readonly _keyPrefix: string;
+
+    /**
+     * Create a new instance of subscriber.
+     *
+     * @param {any} options
+     */
+    constructor(private options) {
+        this._keyPrefix = options.databaseConfig.redis.keyPrefix || '';
+        this._redis = new Redis(options.databaseConfig.redis);
+    }
+
+    /**
+     * Subscribe to events to broadcast.
+     *
+     * @return {Promise<any>}
+     */
+    publish(channel: string, event: string, data: any): Promise<any> {
+        return new Promise((resolve, reject) => {
+            try {
+                this._redis.publish(this._keyPrefix + channel, JSON.stringify({event, ..._.omit(data, ['socket'])}));
+                resolve();
+            } catch (e) {
+                reject(e);
+            }
+        });
+    }
+}