feat: Browser-SDK Automatically start streaming based on event handlers. (#592)

Author: kinyoklion

packages/sdk/browser/__tests__/BrowserDataManager.test.ts

@@ -128,7 +128,7 @@ describe('given a BrowserDataManager with mocked dependencies', () => {
       off: jest.fn(),
     } as unknown as jest.Mocked<FlagManager>;
 
-    browserConfig = validateOptions({ streaming: false }, logger);
+    browserConfig = validateOptions({}, logger);
     baseHeaders = {};
     emitter = {
       emit: jest.fn(),
@@ -262,7 +262,7 @@ describe('given a BrowserDataManager with mocked dependencies', () => {
     await dataManager.identify(identifyResolve, identifyReject, context, identifyOptions);
 
     expect(platform.requests.createEventSource).not.toHaveBeenCalled();
-    dataManager.startDataSource();
+    dataManager.setForcedStreaming(true);
     expect(platform.requests.createEventSource).toHaveBeenCalled();
   });
 
@@ -277,8 +277,8 @@ describe('given a BrowserDataManager with mocked dependencies', () => {
     await dataManager.identify(identifyResolve, identifyReject, context, identifyOptions);
 
     expect(platform.requests.createEventSource).not.toHaveBeenCalled();
-    dataManager.startDataSource();
-    dataManager.startDataSource();
+    dataManager.setForcedStreaming(true);
+    dataManager.setForcedStreaming(true);
     expect(platform.requests.createEventSource).toHaveBeenCalledTimes(1);
     expect(logger.debug).toHaveBeenCalledWith(
       '[BrowserDataManager] Update processor already active. Not changing state.',
@@ -287,10 +287,66 @@ describe('given a BrowserDataManager with mocked dependencies', () => {
 
   it('does not start a stream if identify has not been called', async () => {
     expect(platform.requests.createEventSource).not.toHaveBeenCalled();
-    dataManager.startDataSource();
+    dataManager.setForcedStreaming(true);
     expect(platform.requests.createEventSource).not.toHaveBeenCalledTimes(1);
     expect(logger.debug).toHaveBeenCalledWith(
       '[BrowserDataManager] Context not set, not starting update processor.',
     );
   });
+
+  it('starts a stream on demand when not forced on/off', async () => {
+    const context = Context.fromLDContext({ kind: 'user', key: 'test-user' });
+    const identifyOptions: LDIdentifyOptions = { waitForNetworkResults: false };
+    const identifyResolve = jest.fn();
+    const identifyReject = jest.fn();
+
+    flagManager.loadCached.mockResolvedValue(false);
+
+    await dataManager.identify(identifyResolve, identifyReject, context, identifyOptions);
+
+    expect(platform.requests.createEventSource).not.toHaveBeenCalled();
+    dataManager.setAutomaticStreamingState(true);
+    expect(platform.requests.createEventSource).toHaveBeenCalled();
+    expect(logger.debug).toHaveBeenCalledWith('[BrowserDataManager] Starting update processor.');
+    expect(logger.debug).toHaveBeenCalledWith(
+      '[BrowserDataManager] Updating streaming state. forced(undefined) automatic(true)',
+    );
+  });
+
+  it('does not start a stream when forced off', async () => {
+    const context = Context.fromLDContext({ kind: 'user', key: 'test-user' });
+    const identifyOptions: LDIdentifyOptions = { waitForNetworkResults: false };
+    const identifyResolve = jest.fn();
+    const identifyReject = jest.fn();
+
+    dataManager.setForcedStreaming(false);
+
+    flagManager.loadCached.mockResolvedValue(false);
+
+    await dataManager.identify(identifyResolve, identifyReject, context, identifyOptions);
+
+    expect(platform.requests.createEventSource).not.toHaveBeenCalled();
+    dataManager.setAutomaticStreamingState(true);
+    expect(platform.requests.createEventSource).not.toHaveBeenCalled();
+    expect(logger.debug).toHaveBeenCalledWith(
+      '[BrowserDataManager] Updating streaming state. forced(false) automatic(true)',
+    );
+  });
+
+  it('starts streaming on identify if the automatic state is true', async () => {
+    const context = Context.fromLDContext({ kind: 'user', key: 'test-user' });
+    const identifyOptions: LDIdentifyOptions = { waitForNetworkResults: false };
+    const identifyResolve = jest.fn();
+    const identifyReject = jest.fn();
+
+    dataManager.setForcedStreaming(undefined);
+    dataManager.setAutomaticStreamingState(true);
+    expect(platform.requests.createEventSource).not.toHaveBeenCalled();
+
+    flagManager.loadCached.mockResolvedValue(false);
+
+    await dataManager.identify(identifyResolve, identifyReject, context, identifyOptions);
+
+    expect(platform.requests.createEventSource).toHaveBeenCalled();
+  });
 });

packages/sdk/browser/src/BrowserClient.ts

@@ -1,9 +1,9 @@
 import {
   AutoEnvAttributes,
   base64UrlEncode,
-  BasicLogger,
   LDClient as CommonClient,
   Configuration,
+  createSafeLogger,
   Encoding,
   FlagManager,
   internal,
@@ -14,6 +14,7 @@ import {
   Platform,
 } from '@launchdarkly/js-client-sdk-common';
 import { LDIdentifyOptions } from '@launchdarkly/js-client-sdk-common/dist/api/LDIdentifyOptions';
+import { EventName } from '@launchdarkly/js-client-sdk-common/dist/LDEmitter';
 
 import BrowserDataManager from './BrowserDataManager';
 import GoalManager from './goals/GoalManager';
@@ -29,24 +30,40 @@ export type LDClient = Omit<
   CommonClient,
   'setConnectionMode' | 'getConnectionMode' | 'getOffline'
 > & {
-  setStreaming(streaming: boolean): void;
+  /**
+   * 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 {@link LDClient.on}).
+   *
+   * This can also be set as the `streaming` property of {@link LDOptions}.
+   */
+  setStreaming(streaming?: boolean): void;
 };
 
 export class BrowserClient extends LDClientImpl {
   private readonly goalManager?: GoalManager;
+
   constructor(
     private readonly clientSideId: string,
     autoEnvAttributes: AutoEnvAttributes,
     options: BrowserOptions = {},
     overridePlatform?: Platform,
   ) {
     const { logger: customLogger, debug } = options;
+    // Overrides the default logger from the common implementation.
     const logger =
       customLogger ??
-      new BasicLogger({
-        level: debug ? 'debug' : 'info',
+      createSafeLogger({
+        // eslint-disable-next-line no-console
+        debug: debug ? console.debug : () => {},
+        // eslint-disable-next-line no-console
+        info: console.info,
         // eslint-disable-next-line no-console
-        destination: console.log,
+        warn: console.warn,
+        // eslint-disable-next-line no-console
+        error: console.error,
       });
 
     // TODO: Use the already-configured baseUri from the SDK config. SDK-560
@@ -59,7 +76,7 @@ export class BrowserClient extends LDClientImpl {
       clientSideId,
       autoEnvAttributes,
       platform,
-      filterToBaseOptions(options),
+      filterToBaseOptions({ ...options, logger }),
       (
         flagManager: FlagManager,
         configuration: Configuration,
@@ -164,12 +181,27 @@ export class BrowserClient extends LDClientImpl {
     this.goalManager?.startTracking();
   }
 
-  setStreaming(streaming: boolean): void {
+  setStreaming(streaming?: boolean): void {
+    // With FDv2 we may want to consider if we support connection mode directly.
+    // Maybe with an extension to connection mode for 'automatic'.
     const browserDataManager = this.dataManager as BrowserDataManager;
-    if (streaming) {
-      browserDataManager.startDataSource();
-    } else {
-      browserDataManager.stopDataSource();
-    }
+    browserDataManager.setForcedStreaming(streaming);
+  }
+
+  private updateAutomaticStreamingState() {
+    const browserDataManager = this.dataManager as BrowserDataManager;
+    // This will need changed if support for listening to individual flag change
+    // events it added.
+    browserDataManager.setAutomaticStreamingState(!!this.emitter.listenerCount('change'));
+  }
+
+  override on(eventName: EventName, listener: Function): void {
+    super.on(eventName, listener);
+    this.updateAutomaticStreamingState();
+  }
+
+  override off(eventName: EventName, listener: Function): void {
+    super.off(eventName, listener);
+    this.updateAutomaticStreamingState();
   }
 }

packages/sdk/browser/src/BrowserDataManager.ts

@@ -18,6 +18,22 @@ import { ValidatedOptions } from './options';
 const logTag = '[BrowserDataManager]';
 
 export default class BrowserDataManager extends BaseDataManager {
+  // If streaming is forced on or off, then we follow that setting.
+  // Otherwise we automatically manage streaming state.
+  private forcedStreaming?: boolean = undefined;
+  private automaticStreamingState: boolean = false;
+
+  // +-----------+-----------+---------------+
+  // |  forced   | automatic |     state     |
+  // +-----------+-----------+---------------+
+  // | true      | false     | streaming     |
+  // | true      | true      | streaming     |
+  // | false     | true      | not streaming |
+  // | false     | false     | not streaming |
+  // | undefined | true      | streaming     |
+  // | undefined | false     | not streaming |
+  // +-----------+-----------+---------------+
+
   constructor(
     platform: Platform,
     flagManager: FlagManager,
@@ -41,6 +57,7 @@ export default class BrowserDataManager extends BaseDataManager {
       emitter,
       diagnosticsManager,
     );
+    this.forcedStreaming = browserConfig.streaming;
   }
 
   private debugLog(message: any, ...args: any[]) {
@@ -71,17 +88,43 @@ export default class BrowserDataManager extends BaseDataManager {
       identifyReject(e);
     }
 
-    if (this.browserConfig.streaming) {
-      this.setupConnection(context);
+    this.updateStreamingState();
+  }
+
+  setForcedStreaming(streaming?: boolean) {
+    this.forcedStreaming = streaming;
+    this.updateStreamingState();
+  }
+
+  setAutomaticStreamingState(streaming: boolean) {
+    this.automaticStreamingState = streaming;
+    this.updateStreamingState();
+  }
+
+  private updateStreamingState() {
+    const shouldBeStreaming =
+      this.forcedStreaming || (this.automaticStreamingState && this.forcedStreaming === undefined);
+
+    this.debugLog(
+      `Updating streaming state. forced(${this.forcedStreaming}) automatic(${this.automaticStreamingState})`,
+    );
+
+    if (shouldBeStreaming) {
+      this.startDataSource();
+    } else {
+      this.stopDataSource();
     }
   }
 
-  stopDataSource() {
+  private stopDataSource() {
+    if (this.updateProcessor) {
+      this.debugLog('Stopping update processor.');
+    }
     this.updateProcessor?.close();
     this.updateProcessor = undefined;
   }
 
-  startDataSource() {
+  private startDataSource() {
     if (this.updateProcessor) {
       this.debugLog('Update processor already active. Not changing state.');
       return;
@@ -132,5 +175,4 @@ export default class BrowserDataManager extends BaseDataManager {
     const uri = getPollingUri(this.config.serviceEndpoints, path, parameters);
     return new Requestor(this.platform.requests, uri, headers, method, body);
   }
-  // TODO: Automatically start streaming if event handlers are registered.
 }

packages/shared/sdk-client/__tests__/LDClientImpl.test.ts

@@ -268,7 +268,9 @@ describe('sdk-client object', () => {
     const carContext2: LDContext = { kind: 'car', key: 'test-car-2' };
     await ldc.identify(carContext2);
 
-    expect(emitter.listenerCount('change')).toEqual(1);
+    // No default listeners. This is important for clients to be able to determine if there are
+    // any listeners and act on that information.
+    expect(emitter.listenerCount('change')).toEqual(0);
     expect(emitter.listenerCount('error')).toEqual(1);
   });
 

packages/shared/sdk-client/src/LDClientImpl.ts

@@ -53,7 +53,7 @@ export default class LDClientImpl implements LDClient {
 
   private eventFactoryDefault = new EventFactory(false);
   private eventFactoryWithReasons = new EventFactory(true);
-  private emitter: LDEmitter;
+  protected emitter: LDEmitter;
   private flagManager: FlagManager;
 
   private eventSendingEnabled: boolean = false;
@@ -105,15 +105,13 @@ export default class LDClientImpl implements LDClient {
       this.diagnosticsManager,
     );
     this.emitter = new LDEmitter();
-    this.emitter.on('change', (c: LDContext, changedKeys: string[]) => {
-      this.logger.debug(`change: context: ${JSON.stringify(c)}, flags: ${changedKeys}`);
-    });
     this.emitter.on('error', (c: LDContext, err: any) => {
       this.logger.error(`error: ${err}, context: ${JSON.stringify(c)}`);
     });
 
     this.flagManager.on((context, flagKeys) => {
       const ldContext = Context.toLDContext(context);
+      this.logger.debug(`change: context: ${JSON.stringify(ldContext)}, flags: ${flagKeys}`);
       this.emitter.emit('change', ldContext, flagKeys);
     });
 

packages/shared/sdk-client/src/LDEmitter.ts

@@ -2,6 +2,12 @@ import { LDLogger } from '@launchdarkly/js-sdk-common';
 
 export type EventName = 'error' | 'change';
 
+/**
+ * Implementation Note: There should not be any default listeners for change events in a client
+ * implementation. Default listeners mean a client cannot determine when there are actual
+ * application developer provided listeners. If we require default listeners, then we should add
+ * a system to allow listeners which have counts independent of the primary listener counts.
+ */
 export default class LDEmitter {
   private listeners: Map<EventName, Function[]> = new Map();