Live query pubsub adapter (#2902)

* Moves LiveQuery pub/sub to adapter folder

* Adds ability to provide custom adapter for LiveQuery pubsub

* Adds test for function based adapter

* Pass all options to createSubscriber

* nits

src/LiveQuery/EventEmitterPubSub.js

@@ -1,59 +0,0 @@
-import events from 'events';
-
-let emitter = new events.EventEmitter();
-
-class Publisher {
-  emitter: any;
-
-  constructor(emitter: any) {
-    this.emitter = emitter;
-  }
-
-  publish(channel: string, message: string): void {
-    this.emitter.emit(channel, message);
-  }
-}
-
-class Subscriber extends events.EventEmitter {
-  emitter: any;
-  subscriptions: any;
-
-  constructor(emitter: any) {
-    super();
-    this.emitter = emitter;
-    this.subscriptions = new Map();
-  }
-
-  subscribe(channel: string): void {
-    let handler = (message) => {
-      this.emit('message', channel, message);
-    }
-    this.subscriptions.set(channel, handler);
-    this.emitter.on(channel, handler);
-  }
-
-  unsubscribe(channel: string): void {
-    if (!this.subscriptions.has(channel)) {
-      return;
-    }
-    this.emitter.removeListener(channel, this.subscriptions.get(channel));
-    this.subscriptions.delete(channel);
-  }
-}
-
-function createPublisher(): any {
-  return new Publisher(emitter);
-}
-
-function createSubscriber(): any {
-  return new Subscriber(emitter);
-}
-
-let EventEmitterPubSub = {
-  createPublisher,
-  createSubscriber
-}
-
-export {
-  EventEmitterPubSub
-}

src/LiveQuery/ParseLiveQueryServer.js

@@ -54,9 +54,7 @@ class ParseLiveQueryServer {
     );
 
     // Initialize subscriber
-    this.subscriber = ParsePubSub.createSubscriber({
-      redisURL: config.redisURL
-    });
+    this.subscriber = ParsePubSub.createSubscriber(config);
     this.subscriber.subscribe('afterSave');
     this.subscriber.subscribe('afterDelete');
     // Register message handler for subscriber. When publisher get messages, it will publish message

src/LiveQuery/ParsePubSub.js

@@ -1,5 +1,11 @@
-import { RedisPubSub } from './RedisPubSub';
-import { EventEmitterPubSub } from './EventEmitterPubSub';
+import { loadAdapter } from '../Adapters/AdapterLoader';
+import {
+  EventEmitterPubSub
+} from '../Adapters/PubSub/EventEmitterPubSub';
+
+import {
+  RedisPubSub
+} from '../Adapters/PubSub/RedisPubSub';
 
 let ParsePubSub = {};
 
@@ -10,17 +16,25 @@ function useRedis(config: any): boolean {
 
 ParsePubSub.createPublisher = function(config: any): any {
   if (useRedis(config)) {
-    return RedisPubSub.createPublisher(config.redisURL);
+    return RedisPubSub.createPublisher(config);
   } else {
-    return EventEmitterPubSub.createPublisher();
+    let adapter = loadAdapter(config.pubSubAdapter, EventEmitterPubSub, config)
+    if (typeof adapter.createPublisher !== 'function') {
+      throw 'pubSubAdapter should have createPublisher()';
+    }
+    return adapter.createPublisher(config);
   }
 }
 
 ParsePubSub.createSubscriber = function(config: any): void {
   if (useRedis(config)) {
-    return RedisPubSub.createSubscriber(config.redisURL);
+    return RedisPubSub.createSubscriber(config);
   } else {
-    return EventEmitterPubSub.createSubscriber();
+    let adapter = loadAdapter(config.pubSubAdapter, EventEmitterPubSub, config)
+    if (typeof adapter.createSubscriber !== 'function') {
+      throw 'pubSubAdapter should have createSubscriber()';
+    }
+    return adapter.createSubscriber(config);
   }
 }
 

src/LiveQuery/RedisPubSub.js

@@ -1,18 +0,0 @@
-import redis from 'redis';
-
-function createPublisher(redisURL: string): any {
-  return redis.createClient(redisURL, { no_ready_check: true });
-}
-
-function createSubscriber(redisURL: string): any {
-  return redis.createClient(redisURL, { no_ready_check: true });
-}
-
-let RedisPubSub = {
-  createPublisher,
-  createSubscriber
-}
-
-export {
-  RedisPubSub
-}