allow to pass a custom serializer and deserializer

Author: Quentin de Longraye

src/redis-pubsub.ts

@@ -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/test/tests.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,65 @@ 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('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('throws if you try to unsubscribe with an unknown id', () => {
     const pubSub = new RedisPubSub(mockOptions);
     return expect(() => pubSub.unsubscribe(123))