feat: Automatically start streaming based on event handlers.

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,49 @@ 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)',
+    );
+  });
 });

packages/sdk/browser/src/BrowserClient.ts

@@ -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,11 +30,21 @@ 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,
@@ -164,12 +175,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,11 @@ 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;
+
   constructor(
     platform: Platform,
     flagManager: FlagManager,
@@ -41,6 +46,7 @@ export default class BrowserDataManager extends BaseDataManager {
       emitter,
       diagnosticsManager,
     );
+    this.forcedStreaming = browserConfig.streaming;
   }
 
   private debugLog(message: any, ...args: any[]) {
@@ -76,12 +82,37 @@ export default class BrowserDataManager extends BaseDataManager {
     }
   }
 
-  stopDataSource() {
+  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();
+    }
+  }
+
+  private stopDataSource() {
     this.updateProcessor?.close();
     this.updateProcessor = undefined;
   }
 
-  startDataSource() {
+  private startDataSource() {
     if (this.updateProcessor) {
       this.debugLog('Update processor already active. Not changing state.');
       return;

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

@@ -54,7 +54,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: DefaultFlagManager;
 
   private eventSendingEnabled: boolean = false;