allow streaming mode to be decoupled from event subscription

src/EventEmitter.js

@@ -32,6 +32,14 @@ export default function EventEmitter() {
     }
   };
 
+  emitter.getEvents = function() {
+    return Object.keys(events);
+  };
+
+  emitter.getEventListenerCount = function(event) {
+    return events[event] ? events[event].length : 0;
+  };
+
   emitter.maybeReportError = function(error) {
     if (!error) {
       return;

src/__tests__/LDClient-streaming-test.js

@@ -44,37 +44,157 @@ describe('LDClient', () => {
 
   describe('streaming/event listening', () => {
     const streamUrl = 'https://clientstream.launchdarkly.com';
+    const fullStreamUrlWithUser = streamUrl + '/eval/' + envName + '/' + encodedUser;
 
     function streamEvents() {
-      return sources[`${streamUrl}/eval/${envName}/${encodedUser}`].__emitter._events;
+      return sources[fullStreamUrlWithUser].__emitter._events;
+    }
+
+    function expectStreamUrlIsOpen(url) {
+      expect(Object.keys(sources)).toEqual([url]);
+    }
+
+    function expectNoStreamIsOpen() {
+      expect(sources).toMatchObject({});
     }
 
     it('does not connect to the stream by default', done => {
       const client = LDClient.initialize(envName, user, { bootstrap: {} });
 
       client.on('ready', () => {
-        expect(sources).toMatchObject({});
+        expectNoStreamIsOpen();
         done();
       });
     });
 
-    it('connects to the stream when listening to global change events', done => {
-      const client = LDClient.initialize(envName, user, { bootstrap: {} });
+    it('connects to the stream if options.streaming is true', done => {
+      const client = LDClient.initialize(envName, user, { bootstrap: {}, streaming: true });
 
       client.on('ready', () => {
-        client.on('change', () => {});
-        expect(Object.keys(sources)).toEqual([streamUrl + '/eval/' + envName + '/' + encodedUser]);
+        expectStreamUrlIsOpen(fullStreamUrlWithUser);
         done();
       });
     });
 
-    it('connects to the stream when listening to change event for one flag', done => {
-      const client = LDClient.initialize(envName, user, { bootstrap: {} });
+    describe('setStreaming()', () => {
+      it('can connect to the stream', done => {
+        const client = LDClient.initialize(envName, user, { bootstrap: {} });
 
-      client.on('ready', () => {
-        client.on('change:flagkey', () => {});
-        expect(Object.keys(sources)).toEqual([streamUrl + '/eval/' + envName + '/' + encodedUser]);
-        done();
+        client.on('ready', () => {
+          client.setStreaming(true);
+          expectStreamUrlIsOpen(fullStreamUrlWithUser);
+          done();
+        });
+      });
+
+      it('can disconnect from the stream', done => {
+        const client = LDClient.initialize(envName, user, { bootstrap: {} });
+
+        client.on('ready', () => {
+          client.setStreaming(true);
+          expectStreamUrlIsOpen(fullStreamUrlWithUser);
+          client.setStreaming(false);
+          expectNoStreamIsOpen();
+          done();
+        });
+      });
+    });
+
+    describe('on("change")', () => {
+      it('connects to the stream if not otherwise overridden', done => {
+        const client = LDClient.initialize(envName, user, { bootstrap: {} });
+
+        client.on('ready', () => {
+          client.on('change', () => {});
+          expectStreamUrlIsOpen(fullStreamUrlWithUser);
+          done();
+        });
+      });
+
+      it('also connects if listening for a specific flag', done => {
+        const client = LDClient.initialize(envName, user, { bootstrap: {} });
+
+        client.on('ready', () => {
+          client.on('change:flagkey', () => {});
+          expectStreamUrlIsOpen(fullStreamUrlWithUser);
+          done();
+        });
+      });
+
+      it('does not connect if some other kind of event was specified', done => {
+        const client = LDClient.initialize(envName, user, { bootstrap: {} });
+
+        client.on('ready', () => {
+          client.on('error', () => {});
+          expectNoStreamIsOpen();
+          done();
+        });
+      });
+
+      it('does not connect if options.streaming is explicitly set to false', done => {
+        const client = LDClient.initialize(envName, user, { bootstrap: {}, streaming: false });
+
+        client.on('ready', () => {
+          client.on('change', () => {});
+          expectNoStreamIsOpen();
+          done();
+        });
+      });
+
+      it('does not connect if setStreaming(false) was called', done => {
+        const client = LDClient.initialize(envName, user, { bootstrap: {} });
+
+        client.on('ready', () => {
+          client.setStreaming(false);
+          client.on('change', () => {});
+          expectNoStreamIsOpen();
+          done();
+        });
+      });
+    });
+
+    describe('off("change")', () => {
+      it('disconnects from the stream if all event listeners are removed', done => {
+        const client = LDClient.initialize(envName, user, { bootstrap: {} });
+        const listener1 = () => {};
+        const listener2 = () => {};
+
+        client.on('ready', () => {
+          client.on('change', listener1);
+          client.on('change:flagkey', listener2);
+          client.on('error', () => {});
+          expectStreamUrlIsOpen(fullStreamUrlWithUser);
+
+          client.off('change', listener1);
+          expectStreamUrlIsOpen(fullStreamUrlWithUser);
+
+          client.off('change:flagkey', listener2);
+          expectNoStreamIsOpen();
+
+          done();
+        });
+      });
+
+      it('does not disconnect if setStreaming(true) was called', done => {
+        const client = LDClient.initialize(envName, user, { bootstrap: {} });
+        const listener1 = () => {};
+        const listener2 = () => {};
+
+        client.on('ready', () => {
+          client.setStreaming(true);
+
+          client.on('change', listener1);
+          client.on('change:flagkey', listener2);
+          expectStreamUrlIsOpen(fullStreamUrlWithUser);
+
+          client.off('change', listener1);
+          expectStreamUrlIsOpen(fullStreamUrlWithUser);
+
+          client.off('change:flagkey', listener2);
+          expectStreamUrlIsOpen(fullStreamUrlWithUser);
+
+          done();
+        });
       });
     });
 
@@ -83,7 +203,7 @@ describe('LDClient', () => {
 
       client.on('ready', () => {
         client.on('change:flagkey', () => {});
-        expect(Object.keys(sources)).toEqual([streamUrl + '/eval/' + envName + '/' + encodedUser + '?h=' + hash]);
+        expectStreamUrlIsOpen(fullStreamUrlWithUser + '?h=' + hash);
         done();
       });
     });
@@ -93,9 +213,7 @@ describe('LDClient', () => {
 
       client.on('ready', () => {
         client.on('change', () => {});
-        expect(Object.keys(sources)).toEqual([
-          streamUrl + '/eval/' + envName + '/' + encodedUser + '?withReasons=true',
-        ]);
+        expectStreamUrlIsOpen(fullStreamUrlWithUser + '?withReasons=true');
         done();
       });
     });
@@ -105,9 +223,7 @@ describe('LDClient', () => {
 
       client.on('ready', () => {
         client.on('change', () => {});
-        expect(Object.keys(sources)).toEqual([
-          streamUrl + '/eval/' + envName + '/' + encodedUser + '?h=' + hash + '&withReasons=true',
-        ]);
+        expectStreamUrlIsOpen(fullStreamUrlWithUser + '?h=' + hash + '&withReasons=true');
         done();
       });
     });

src/index.js

@@ -36,6 +36,8 @@ export function initialize(env, user, options = {}) {
   let goalTracker;
   let useLocalStorage;
   let goals;
+  let streamActive;
+  let streamForcedState;
   let subscribedToChangeEvents;
   let firstEvent = true;
 
@@ -180,7 +182,7 @@ export function initialize(env, user, options = {}) {
               updateSettings(settings);
             }
             resolve(utils.transformVersionedValuesToValues(settings));
-            if (subscribedToChangeEvents) {
+            if (streamActive) {
               connectStream();
             }
           });
@@ -299,6 +301,7 @@ export function initialize(env, user, options = {}) {
   }
 
   function connectStream() {
+    streamActive = true;
     if (!ident.getUser()) {
       return;
     }
@@ -403,9 +406,9 @@ export function initialize(env, user, options = {}) {
   }
 
   function on(event, handler, context) {
-    if (event.substr(0, changeEvent.length) === changeEvent) {
+    if (isChangeEventKey(event)) {
       subscribedToChangeEvents = true;
-      if (!stream.isConnected()) {
+      if (!streamActive && streamForcedState === undefined) {
         connectStream();
       }
       emitter.on.apply(emitter, [event, handler, context]);
@@ -415,13 +418,40 @@ export function initialize(env, user, options = {}) {
   }
 
   function off(event) {
-    if (event === changeEvent) {
-      if ((subscribedToChangeEvents = true)) {
+    emitter.off.apply(emitter, Array.prototype.slice.call(arguments));
+    if (isChangeEventKey(event)) {
+      let haveListeners = false;
+      emitter.getEvents().forEach(key => {
+        if (isChangeEventKey(key) && emitter.getEventListenerCount(key) > 0) {
+          haveListeners = true;
+        }
+      });
+      if (!haveListeners) {
         subscribedToChangeEvents = false;
+        if (streamActive && streamForcedState === undefined) {
+          streamActive = false;
+          stream.disconnect();
+        }
+      }
+    }
+  }
+
+  function setStreaming(state) {
+    const newState = state === null ? undefined : state;
+    if (newState !== streamForcedState) {
+      streamForcedState = newState;
+      let shouldBeStreaming = streamForcedState || (subscribedToChangeEvents && streamForcedState === undefined);
+      if (shouldBeStreaming && !streamActive) {
+        connectStream();
+      } else if (!shouldBeStreaming && streamActive) {
+        streamActive = false;
         stream.disconnect();
       }
     }
-    emitter.off.apply(emitter, Array.prototype.slice.call(arguments));
+  }
+
+  function isChangeEventKey(event) {
+    return event === changeEvent || event.substr(0, changeEvent.length + 1) === changeEvent + ':';
   }
 
   function handleMessage(event) {
@@ -565,6 +595,9 @@ export function initialize(env, user, options = {}) {
   }
 
   function signalSuccessfulInit() {
+    if (options.streaming !== undefined) {
+      setStreaming(options.streaming);
+    }
     emitter.emit(readyEvent);
     emitter.emit(successEvent); // allows initPromise to distinguish between success and failure
   }
@@ -631,6 +664,7 @@ export function initialize(env, user, options = {}) {
     track: track,
     on: on,
     off: off,
+    setStreaming: setStreaming,
     flush: flush,
     allFlags: allFlags,
   };

typings.d.ts

@@ -73,9 +73,10 @@ declare module 'ldclient-js' {
    * flag value, and the previous value. This is always accompanied by a general
    * "change" event as described above; you can listen for either or both.
    *
-   * The "change" and "change:FLAG-KEY" events have special behavior: the client
-   * will open a streaming connection to receive live changes if and only if you
-   * are listening for one of these events.
+   * The "change" and "change:FLAG-KEY" events have special behavior: by default, the
+   * client will open a streaming connection to receive live changes if and only if
+   * you are listening for one of these events. This behavior can be overridden by
+   * setting LDOptions.streaming or calling LDClient.setStreaming().
    */
   type LDEventSignature = (
     key: string,
@@ -128,6 +129,17 @@ declare module 'ldclient-js' {
      */
     streamUrl?: string;
 
+    /**
+     * Whether or not to open a streaming connection to LaunchDarkly for live flag updates.
+     *
+     * If this is true, the client will always attempt to maintain a streaming connection; if false,
+     * it never will. If you leave the value undefined (the default), the client will open a streaming
+     * connection if you subscribe to "change" or "change:flag-key" events (see LDClient.on()).
+     *
+     * This is equivalent to calling client.setStreaming() with the same value.
+     */
+    streaming?: boolean;
+
     /**
      * Whether or not to use the REPORT verb to fetch flag settings.
      *
@@ -424,6 +436,17 @@ declare module 'ldclient-js' {
      */
     variationDetail: (key: string, defaultValue?: LDFlagValue) => LDEvaluationDetail;
 
+    /**
+     * Specifies whether or not to open a streaming connection to LaunchDarkly for live flag updates.
+     *
+     * If this is true, the client will always attempt to maintain a streaming connection; if false,
+     * it never will. If you leave the value undefined (the default), the client will open a streaming
+     * connection if you subscribe to "change" or "change:flag-key" events (see LDClient.on()).
+     *
+     * This can also be set as the "streaming" property of the client options.
+     */
+    setStreaming: (value?: boolean) => void;
+
     /**
      * Registers an event listener. See LDEventSignature for the available event types
      * and the data that can be associated with them.