feat(*): add RedisHub for Redis support with sseHub()

package.json

@@ -27,6 +27,8 @@
     "@types/mocha": "^2.2.41",
     "@types/node": "^8.0.13",
     "chai": "^4.0.0",
+    "cors": "^2.8.4",
+    "express": "^4.15.4",
     "mocha": "^3.4.2",
     "semantic-release": "^6.3.6",
     "tslint": "^5.2.0",
@@ -36,6 +38,8 @@
     "express": ">=4.0.0"
   },
   "dependencies": {
-    "compose-middleware": "^2.2.0"
+    "@types/ioredis": "^0.0.24",
+    "compose-middleware": "^2.2.0",
+    "ioredis": "^3.1.1"
   }
 }

src/index.ts

@@ -1,4 +1,5 @@
 export * from './hub';
+export * from './redis_hub';
 export { ISseMiddlewareOptions } from './sse_handler_middleware';
 export * from './sse_middleware';
 export * from './sse_hub_middleware';

src/redis_hub.ts

@@ -0,0 +1,48 @@
+import * as Redis from 'ioredis';
+import { Hub } from './hub';
+import * as fmt from './sse_formatter';
+import { ISseFunctions } from './sse_middleware';
+
+interface ISerializedSseMessage {
+    type: keyof ISseFunctions;
+    args: any[];
+}
+
+export class RedisHub extends Hub {
+    public constructor(
+        public readonly channel: string,
+        private readonly pub = new Redis(),
+        private readonly sub = new Redis()
+    ) {
+        super();
+
+        sub.subscribe(channel);
+        sub.on('message', (chan, message) => this.handleRedisMessage(message));
+    }
+
+    public data(data: fmt.SseValue, id?: string): void {
+        this.publish('data', arguments);
+    }
+
+    public event(event: string, data: fmt.SseValue, id?: string): void {
+        this.publish('event', arguments);
+    }
+
+    public comment(comment: string): void {
+        this.publish('comment', arguments);
+    }
+
+    private publish(type: keyof ISseFunctions, args: IArguments): void {
+        const message: ISerializedSseMessage = {
+            type, args: Array.from(args)
+        };
+
+        this.pub.publish(this.channel, JSON.stringify(message));
+    }
+
+    private handleRedisMessage(jsonMessage: string): void {
+        const { type, args } = JSON.parse(jsonMessage) as ISerializedSseMessage;
+
+        super[type].apply(this, args);
+    }
+}