feat: adds polling synchronizer support (#816)

**Requirements**

- [x] I have added test coverage for new or changed functionality

- [x] I have followed the repository's [pull request submission
guidelines](../blob/main/CONTRIBUTING.md#submitting-pull-requests)
- [ ] I have validated my changes against all supported platform
versions
This will be done at a later time on the target branch during
integration testing.

**Related issues**

SDK-858 and SDK-851

**Describe the solution you've provided**

Refactors payload processing to be reusable between polling data sources
and streaming data sources.
Adds OneShotInitializers and PollingSynchronizer.
Adds support to contract tests for sdk-test-harness/feat/fdv2 branch.

---------

Co-authored-by: Todd Anderson <[email protected]>

Author: tanderson-ld

contract-tests/sdkClientEntity.js

@@ -23,6 +23,7 @@ export function makeSdkConfig(options, tag) {
 
   const maybeTime = (seconds) =>
     seconds === undefined || seconds === null ? undefined : seconds / 1000;
+
   if (options.streaming) {
     cf.streamUri = options.streaming.baseUri;
     cf.streamInitialReconnectDelay = maybeTime(options.streaming.initialRetryDelayMs);
@@ -33,7 +34,7 @@ export function makeSdkConfig(options, tag) {
   if (options.polling) {
     cf.stream = false;
     cf.baseUri = options.polling.baseUri;
-    cf.pollInterface = options.polling.pollIntervalMs / 1000;
+    cf.pollInterval = options.polling.pollIntervalMs / 1000;
     if (options.polling.filter) {
       cf.payloadFilterKey = options.polling.filter;
     }
@@ -81,6 +82,61 @@ export function makeSdkConfig(options, tag) {
       cf.wrapperVersion = options.wrapper.version;
     }
   }
+  if (options.dataSystem) {
+    const dataSourceStreamingOptions = options.dataSystem.synchronizers?.primary?.streaming ?? options.dataSystem.synchronizers?.secondary?.streaming;
+    const dataSourcePollingOptions = options.dataSystem.synchronizers?.primary?.polling ?? options.dataSystem.synchronizers?.secondary?.polling;
+    
+    if (dataSourceStreamingOptions) {
+      cf.streamUri = dataSourceStreamingOptions.baseUri;
+      cf.streamInitialReconnectDelay = maybeTime(dataSourceStreamingOptions.initialRetryDelayMs);
+      if (dataSourceStreamingOptions.filter) {
+        cf.payloadFilterKey = dataSourceStreamingOptions.filter;
+      }
+    }
+    if (dataSourcePollingOptions) {
+      cf.stream = false;
+      cf.baseUri = dataSourcePollingOptions.baseUri;
+      cf.pollInterval = dataSourcePollingOptions.pollIntervalMs / 1000;
+      if (dataSourcePollingOptions.filter) {
+        cf.payloadFilterKey = dataSourcePollingOptions.filter;
+      }
+    }
+
+    let dataSourceOptions;
+    if (dataSourceStreamingOptions && dataSourcePollingOptions) {
+      dataSourceOptions = {
+        type: 'standard',
+        ...(dataSourceStreamingOptions.initialRetryDelayMs != null && 
+          { streamInitialReconnectDelay: maybeTime(dataSourceStreamingOptions.initialRetryDelayMs) }),
+       ...(dataSourcePollingOptions.pollIntervalMs != null && 
+          { pollInterval: dataSourcePollingOptions.pollIntervalMs }),
+      }
+    } else if (dataSourceStreamingOptions) {
+      dataSourceOptions = {
+        type: 'streamingOnly',
+        ...(dataSourceStreamingOptions.initialRetryDelayMs != null && 
+          { streamInitialReconnectDelay: maybeTime(dataSourceStreamingOptions.initialRetryDelayMs) }),
+      }
+    } else if (dataSourcePollingOptions) {
+      dataSourceOptions = {
+        type: 'pollingOnly',
+        ...(dataSourcePollingOptions.pollIntervalMs != null && 
+          { pollInterval: dataSourcePollingOptions.pollIntervalMs }),
+      }
+    } else {
+      // No data source options were specified
+      dataSourceOptions = undefined;
+    }
+
+    if (options.dataSystem.payloadFilter) {
+      cf.payloadFilterKey = options.dataSystem.payloadFilter;
+    }
+
+    cf.dataSystem = {
+      dataSource: dataSourceOptions,
+    }
+  }
+
   return cf;
 }
 

contract-tests/testharness-suppressions.txt

@@ -1,2 +1,11 @@
 streaming/validation/drop and reconnect if stream event has malformed JSON
 streaming/validation/drop and reconnect if stream event has well-formed JSON not matching schema
+
+streaming/fdv2/reconnection state management/initializes from polling initializer
+streaming/fdv2/reconnection state management/initializes from 2 polling initializers
+
+streaming/fdv2/reconnection state management/saves previously known state
+streaming/fdv2/reconnection state management/replaces previously known state
+streaming/fdv2/reconnection state management/updates previously known state
+streaming/fdv2/ignores model version
+streaming/fdv2/can discard partial events on errors

packages/shared/common/__tests__/internal/fdv2/PayloadReader.test.ts renamed to packages/shared/common/__tests__/internal/fdv2/PayloadStreamReader.test.ts

@@ -1,5 +1,7 @@
 import { EventListener, EventName, LDLogger } from '../../../src/api';
-import { EventStream, Payload, PayloadReader } from '../../../src/internal/fdv2/payloadReader';
+import { Payload } from '../../../src/internal/fdv2/payloadProcessor';
+import { EventStream, PayloadStreamReader } from '../../../src/internal/fdv2/payloadStreamReader';
+
 
 class MockEventStream implements EventStream {
   private _listeners: Record<EventName, EventListener> = {};
@@ -16,7 +18,7 @@ class MockEventStream implements EventStream {
 it('it sets basis to true when intent code is xfer-full', () => {
   const mockStream = new MockEventStream();
   const receivedPayloads: Payload[] = [];
-  const readerUnderTest = new PayloadReader(mockStream, {
+  const readerUnderTest = new PayloadStreamReader(mockStream, {
     mockKind: (it) => it, // obj processor that just returns the same obj
   });
   readerUnderTest.addPayloadListener((it) => {
@@ -38,7 +40,7 @@ it('it sets basis to true when intent code is xfer-full', () => {
 it('it sets basis to false when intent code is xfer-changes', () => {
   const mockStream = new MockEventStream();
   const receivedPayloads: Payload[] = [];
-  const readerUnderTest = new PayloadReader(mockStream, {
+  const readerUnderTest = new PayloadStreamReader(mockStream, {
     mockKind: (it) => it, // obj processor that just returns the same obj
   });
   readerUnderTest.addPayloadListener((it) => {
@@ -57,10 +59,71 @@ it('it sets basis to false when intent code is xfer-changes', () => {
   expect(receivedPayloads[0].basis).toEqual(false);
 });
 
+it('it sets basis to false and emits empty payload when intent code is none', () => {
+  const mockStream = new MockEventStream();
+  const receivedPayloads: Payload[] = [];
+  const readerUnderTest = new PayloadStreamReader(mockStream, {
+    mockKind: (it) => it, // obj processor that just returns the same obj
+  });
+  readerUnderTest.addPayloadListener((it) => {
+    receivedPayloads.push(it);
+  });
+
+  mockStream.simulateEvent('server-intent', {
+    data: '{"payloads": [{"code": "none", "id": "mockId", "target": 42}]}',
+  });
+  expect(receivedPayloads.length).toEqual(1);
+  expect(receivedPayloads[0].id).toEqual('mockId');
+  expect(receivedPayloads[0].version).toEqual(42);
+  expect(receivedPayloads[0].basis).toEqual(false);
+});
+
+it('it handles xfer-full then xfer-changes', () => {
+  const mockStream = new MockEventStream();
+  const receivedPayloads: Payload[] = [];
+  const readerUnderTest = new PayloadStreamReader(mockStream, {
+    mockKind: (it) => it, // obj processor that just returns the same obj
+  });
+  readerUnderTest.addPayloadListener((it) => {
+    receivedPayloads.push(it);
+  });
+
+  mockStream.simulateEvent('server-intent', {
+    data: '{"payloads": [{"code": "xfer-full", "id": "mockId"}]}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"kind": "mockKind", "key": "flagA", "version": 123, "object": {"objectFieldA": "objectValueA"}}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"state": "mockState", "version": 1}',
+  });
+
+  mockStream.simulateEvent('put-object', {
+    data: '{"kind": "mockKind", "key": "flagA", "version": 456, "object": {"objectFieldA": "newValue"}}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"state": "mockState", "version": 1}',
+  });
+  expect(receivedPayloads.length).toEqual(2);
+  expect(receivedPayloads[0].id).toEqual('mockId');
+  expect(receivedPayloads[0].state).toEqual('mockState');
+  expect(receivedPayloads[0].basis).toEqual(true);
+  expect(receivedPayloads[0].updates.length).toEqual(1);
+  expect(receivedPayloads[0].updates[0].object).toEqual({ objectFieldA: 'objectValueA' });
+  expect(receivedPayloads[0].updates[0].deleted).toEqual(undefined);
+
+  expect(receivedPayloads[1].id).toEqual('mockId');
+  expect(receivedPayloads[1].state).toEqual('mockState');
+  expect(receivedPayloads[1].basis).toEqual(false);
+  expect(receivedPayloads[1].updates.length).toEqual(1);
+  expect(receivedPayloads[1].updates[0].object).toEqual({ objectFieldA: 'newValue' });
+  expect(receivedPayloads[1].updates[0].deleted).toEqual(undefined);
+});
+
 it('it includes multiple types of updates in payload', () => {
   const mockStream = new MockEventStream();
   const receivedPayloads: Payload[] = [];
-  const readerUnderTest = new PayloadReader(mockStream, {
+  const readerUnderTest = new PayloadStreamReader(mockStream, {
     mockKind: (it) => it, // obj processor that just returns the same obj
   });
   readerUnderTest.addPayloadListener((it) => {
@@ -98,7 +161,7 @@ it('it includes multiple types of updates in payload', () => {
 it('it does not include messages thats are not between server-intent and payloader-transferred', () => {
   const mockStream = new MockEventStream();
   const receivedPayloads: Payload[] = [];
-  const readerUnderTest = new PayloadReader(mockStream, {
+  const readerUnderTest = new PayloadStreamReader(mockStream, {
     mockKind: (it) => it, // obj processor that just returns the same obj
   });
   readerUnderTest.addPayloadListener((it) => {
@@ -131,7 +194,7 @@ it('logs prescribed message when goodbye event is encountered', () => {
   };
   const mockStream = new MockEventStream();
   const receivedPayloads: Payload[] = [];
-  const readerUnderTest = new PayloadReader(
+  const readerUnderTest = new PayloadStreamReader(
     mockStream,
     {
       mockKind: (it) => it, // obj processor that just returns the same obj
@@ -162,7 +225,7 @@ it('logs prescribed message when error event is encountered', () => {
   };
   const mockStream = new MockEventStream();
   const receivedPayloads: Payload[] = [];
-  const readerUnderTest = new PayloadReader(
+  const readerUnderTest = new PayloadStreamReader(
     mockStream,
     {
       mockKind: (it) => it, // obj processor that just returns the same obj
@@ -183,12 +246,15 @@ it('logs prescribed message when error event is encountered', () => {
   mockStream.simulateEvent('error', {
     data: '{"reason": "Womp womp"}',
   });
+  mockStream.simulateEvent('put-object', {
+    data: '{"kind": "mockKind", "key": "flagB", "version": 123, "object": {"objectFieldB": "objectValueB"}}',
+  });
   mockStream.simulateEvent('payload-transferred', {
     data: '{"state": "mockState", "version": 1}',
   });
-  expect(receivedPayloads.length).toEqual(0);
+  expect(receivedPayloads.length).toEqual(1);
   expect(mockLogger.info).toHaveBeenCalledWith(
-    'An issue was encountered receiving updates for payload mockId with reason: Womp womp. Automatic retry will occur.',
+    'An issue was encountered receiving updates for payload mockId with reason: Womp womp.',
   );
 });
 
@@ -201,7 +267,7 @@ it('discards partially transferred data when an error is encountered', () => {
   };
   const mockStream = new MockEventStream();
   const receivedPayloads: Payload[] = [];
-  const readerUnderTest = new PayloadReader(
+  const readerUnderTest = new PayloadStreamReader(
     mockStream,
     {
       mockKind: (it) => it, // obj processor that just returns the same obj
@@ -222,6 +288,9 @@ it('discards partially transferred data when an error is encountered', () => {
   mockStream.simulateEvent('error', {
     data: '{"reason": "Womp womp"}',
   });
+  mockStream.simulateEvent('put-object', {
+    data: '{"kind": "mockKind", "key": "flagB", "version": 123, "object": {"objectFieldB": "objectValueB"}}',
+  });
   mockStream.simulateEvent('payload-transferred', {
     data: '{"state": "mockState", "version": 1}',
   });
@@ -240,23 +309,29 @@ it('discards partially transferred data when an error is encountered', () => {
   mockStream.simulateEvent('payload-transferred', {
     data: '{"state": "mockState2", "version": 1}',
   });
-  expect(receivedPayloads.length).toEqual(1);
-  expect(receivedPayloads[0].id).toEqual('mockId2');
-  expect(receivedPayloads[0].state).toEqual('mockState2');
+  expect(receivedPayloads.length).toEqual(2);
+  expect(receivedPayloads[0].id).toEqual('mockId');
+  expect(receivedPayloads[0].state).toEqual('mockState');
   expect(receivedPayloads[0].basis).toEqual(true);
-  expect(receivedPayloads[0].updates.length).toEqual(3);
-  expect(receivedPayloads[0].updates[0].object).toEqual({ objectFieldX: 'objectValueX' });
+  expect(receivedPayloads[0].updates.length).toEqual(1);
+  expect(receivedPayloads[0].updates[0].object).toEqual({ objectFieldB: 'objectValueB' });
   expect(receivedPayloads[0].updates[0].deleted).toEqual(undefined);
-  expect(receivedPayloads[0].updates[1].object).toEqual(undefined);
-  expect(receivedPayloads[0].updates[1].deleted).toEqual(true);
-  expect(receivedPayloads[0].updates[2].object).toEqual({ objectFieldZ: 'objectValueZ' });
-  expect(receivedPayloads[0].updates[2].deleted).toEqual(undefined);
+  expect(receivedPayloads[1].id).toEqual('mockId2');
+  expect(receivedPayloads[1].state).toEqual('mockState2');
+  expect(receivedPayloads[1].basis).toEqual(true);
+  expect(receivedPayloads[1].updates.length).toEqual(3);
+  expect(receivedPayloads[1].updates[0].object).toEqual({ objectFieldX: 'objectValueX' });
+  expect(receivedPayloads[1].updates[0].deleted).toEqual(undefined);
+  expect(receivedPayloads[1].updates[1].object).toEqual(undefined);
+  expect(receivedPayloads[1].updates[1].deleted).toEqual(true);
+  expect(receivedPayloads[1].updates[2].object).toEqual({ objectFieldZ: 'objectValueZ' });
+  expect(receivedPayloads[1].updates[2].deleted).toEqual(undefined);
 });
 
 it('silently ignores unrecognized kinds', () => {
   const mockStream = new MockEventStream();
   const receivedPayloads: Payload[] = [];
-  const readerUnderTest = new PayloadReader(mockStream, {
+  const readerUnderTest = new PayloadStreamReader(mockStream, {
     mockKind: (it) => it, // obj processor that just returns the same obj
   });
   readerUnderTest.addPayloadListener((it) => {
@@ -286,7 +361,7 @@ it('silently ignores unrecognized kinds', () => {
 it('ignores additional payloads beyond the first payload in the server-intent message', () => {
   const mockStream = new MockEventStream();
   const receivedPayloads: Payload[] = [];
-  const readerUnderTest = new PayloadReader(mockStream, {
+  const readerUnderTest = new PayloadStreamReader(mockStream, {
     mockKind: (it) => it, // obj processor that just returns the same obj
   });
   readerUnderTest.addPayloadListener((it) => {

packages/shared/common/src/internal/fdv2/index.ts

@@ -1,3 +1,17 @@
-import { Payload, PayloadListener, PayloadReader, Update } from './payloadReader';
+import {
+  FDv2EventsCollection,
+  Payload,
+  PayloadListener,
+  PayloadProcessor,
+  Update,
+} from './payloadProcessor';
+import { PayloadStreamReader } from './payloadStreamReader';
 
-export { Payload, PayloadListener, PayloadReader, Update };
+export {
+  FDv2EventsCollection,
+  Payload,
+  PayloadListener,
+  PayloadProcessor,
+  PayloadStreamReader,
+  Update,
+};