Merge branch 'master' into greenkeeper/@types/mocha-7.0.0

Author: davidyaha

.travis.yml

@@ -2,7 +2,6 @@ language: node_js
 node_js:
   - "node"
   - "lts/*"
-  - "6"
 
 cache:
   directories:

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "graphql-redis-subscriptions",
-  "version": "2.1.2",
+  "version": "2.2.0",
   "description": "A graphql-subscriptions PubSub Engine using redis",
   "main": "dist/index.js",
   "repository": {
@@ -47,10 +47,10 @@
   },
   "devDependencies": {
     "@types/chai": "^4.1.6",
-    "@types/ioredis": "4.14.1",
+    "@types/ioredis": "4.14.7",
     "@types/chai-as-promised": "7.1.1",
     "@types/mocha": "^7.0.0",
-    "@types/node": "12.12.19",
+    "@types/node": "13.5.3",
     "@types/simple-mock": "0.8.0",
     "chai": "^4.2.0",
     "chai-as-promised": "^7.1.1",

package.json

@@ -9,9 +9,14 @@ export interface PubSubRedisOptions {
   publisher?: RedisClient;
   subscriber?: RedisClient;
   reviver?: Reviver;
+  serializer?: Serializer;
+  deserializer?: Deserializer;
 }
 
 export class RedisPubSub implements PubSubEngine {
+  private readonly serializer?: Serializer;
+  private readonly deserializer?: Deserializer;
+
   constructor(options: PubSubRedisOptions = {}) {
     const {
       triggerTransform,
@@ -20,10 +25,19 @@ export class RedisPubSub implements PubSubEngine {
       subscriber,
       publisher,
       reviver,
+      serializer,
+      deserializer,
     } = options;
 
     this.triggerTransform = triggerTransform || (trigger => trigger as string);
+
+    if (reviver && deserializer) {
+      throw new Error("Reviver and deserializer can't be used together");
+    }
+
     this.reviver = reviver;
+    this.serializer = serializer;
+    this.deserializer = deserializer;
 
     if (subscriber && publisher) {
       this.redisPublisher = publisher;
@@ -53,15 +67,15 @@ export class RedisPubSub implements PubSubEngine {
     // handle messages received via psubscribe and subscribe
     this.redisSubscriber.on('pmessage', this.onMessage.bind(this));
     // partially applied function passes undefined for pattern arg since 'message' event won't provide it:
-    this.redisSubscriber.on('message', this.onMessage.bind(this, undefined)); 
+    this.redisSubscriber.on('message', this.onMessage.bind(this, undefined));
 
     this.subscriptionMap = {};
     this.subsRefsMap = {};
     this.currentSubscriptionId = 0;
   }
 
   public async publish<T>(trigger: string, payload: T): Promise<void> {
-    await this.redisPublisher.publish(trigger, JSON.stringify(payload));
+    await this.redisPublisher.publish(trigger, this.serializer ? this.serializer(payload) : JSON.stringify(payload));
   }
 
   public subscribe(
@@ -108,7 +122,7 @@ export class RedisPubSub implements PubSubEngine {
       // unsubscribe from specific channel and pattern match
       this.redisSubscriber.unsubscribe(triggerName);
       this.redisSubscriber.punsubscribe(triggerName);
-      
+
       delete this.subsRefsMap[triggerName];
     } else {
       const index = refs.indexOf(subId);
@@ -146,7 +160,7 @@ export class RedisPubSub implements PubSubEngine {
 
     let parsedMessage;
     try {
-      parsedMessage = JSON.parse(message, this.reviver);
+      parsedMessage = this.deserializer ? this.deserializer(message) : JSON.parse(message, this.reviver);
     } catch (e) {
       parsedMessage = message;
     }
@@ -174,3 +188,5 @@ export type TriggerTransform = (
   channelOptions?: Object,
 ) => string;
 export type Reviver = (key: any, value: any) => any;
+export type Serializer = (source: any) => string;
+export type Deserializer = (source: string) => any;

src/redis-pubsub.ts

@@ -1,6 +1,6 @@
 import * as chai from 'chai';
 import * as chaiAsPromised from 'chai-as-promised';
-import { spy, restore } from 'simple-mock';
+import { spy, restore, stub } from 'simple-mock';
 import { isAsyncIterable } from 'iterall';
 import { RedisPubSub } from '../redis-pubsub';
 import * as IORedis from 'ioredis';
@@ -272,6 +272,106 @@ describe('RedisPubSub', () => {
     });
   });
 
+  it('refuses custom reviver with a deserializer', done => {
+    const reviver = stub();
+    const deserializer = stub();
+
+    try {
+      expect(() => new RedisPubSub({...mockOptions, reviver, deserializer}))
+          .to.throw("Reviver and deserializer can't be used together");
+      done();
+    } catch (e) {
+      done(e);
+    }
+  });
+
+  it('allows to use a custom serializer', done => {
+    const serializer = stub();
+    const serializedPayload = `{ "hello": "custom" }`;
+    serializer.returnWith(serializedPayload);
+
+    const pubSub = new RedisPubSub({...mockOptions, serializer });
+
+    try {
+      pubSub.subscribe('TOPIC', message => {
+        try {
+          expect(message).to.eql({hello: 'custom'});
+          done();
+        } catch (e) {
+          done(e);
+        }
+      }).then(() => {
+        pubSub.publish('TOPIC', {hello: 'world'});
+      });
+    } catch (e) {
+      done(e);
+    }
+  });
+
+  it('custom serializer can throw an error', done => {
+    const serializer = stub();
+    serializer.throwWith(new Error('Custom serialization error'));
+
+    const pubSub = new RedisPubSub({...mockOptions, serializer });
+
+    try {
+      pubSub.publish('TOPIC', {hello: 'world'}).then(() => {
+        done(new Error('Expected error to be thrown upon publish'));
+     }, err => {
+        expect(err.message).to.eql('Custom serialization error');
+        done();
+      });
+    } catch (e) {
+      done(e);
+    }
+  });
+
+  it('allows to use a custom deserializer', done => {
+    const deserializer = stub();
+    const deserializedPayload = { hello: 'custom' };
+    deserializer.returnWith(deserializedPayload);
+
+    const pubSub = new RedisPubSub({...mockOptions, deserializer });
+
+    try {
+      pubSub.subscribe('TOPIC', message => {
+        try {
+          expect(message).to.eql({hello: 'custom'});
+          done();
+        } catch (e) {
+          done(e);
+        }
+      }).then(() => {
+        pubSub.publish('TOPIC', {hello: 'world'});
+      });
+    } catch (e) {
+      done(e);
+    }
+  });
+
+  it('unparsed payload is returned if custom deserializer throws an error', done => {
+    const deserializer = stub();
+    deserializer.throwWith(new Error('Custom deserialization error'));
+
+    const pubSub = new RedisPubSub({...mockOptions, deserializer });
+
+    try {
+      pubSub.subscribe('TOPIC', message => {
+        try {
+          expect(message).to.be.a('string');
+          expect(message).to.eql('{"hello":"world"}');
+          done();
+        } catch (e) {
+          done(e);
+        }
+      }).then(() => {
+        pubSub.publish('TOPIC', {hello: 'world'});
+      });
+    } catch (e) {
+      done(e);
+    }
+  });
+
   it('throws if you try to unsubscribe with an unknown id', () => {
     const pubSub = new RedisPubSub(mockOptions);
     return expect(() => pubSub.unsubscribe(123))