feat: Adds StreamingProcessor for FDv2 to sdk-server package. (#707)

**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) - [x] I have validated my changes against all supported platform versions **Related issues** SDK-849 **Describe the solution you've provided** Inserted PayloadReader between EventSource and DataSourceUpdates. Contract test glue code can be found on `ta/sdk-849/fdv2-streaming-datasource-contract-test-glue` --------- Co-authored-by: Ryan Lamb <[email protected]>
launchdarkly · Dec 23, 2024 · 7f5c275 · 7f5c275
1 parent 220b6d6
commit 7f5c275
Show file tree

Hide file tree

Showing 11 changed files with 1,264 additions and 6 deletions.
diff --git a/packages/shared/common/__tests__/internal/fdv2/PayloadReader.test.ts b/packages/shared/common/__tests__/internal/fdv2/PayloadReader.test.ts
@@ -0,0 +1,314 @@
+import { EventListener, EventName, LDLogger } from '../../../src/api';
+import { EventStream, Payload, PayloadReader } from '../../../src/internal/fdv2/payloadReader';
+
+class MockEventStream implements EventStream {
+  private _listeners: Record<EventName, EventListener> = {};
+
+  addEventListener(eventName: EventName, listener: EventListener): void {
+    this._listeners[eventName] = listener;
+  }
+
+  simulateEvent(eventName: EventName, event: { data?: string }) {
+    this._listeners[eventName](event);
+  }
+}
+
+it('it sets basis to true when intent code is xfer-full', () => {
+  const mockStream = new MockEventStream();
+  const receivedPayloads: Payload[] = [];
+  const readerUnderTest = new PayloadReader(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('payload-transferred', {
+    data: '{"state": "mockState", "version": 1}',
+  });
+  expect(receivedPayloads.length).toEqual(1);
+  expect(receivedPayloads[0].id).toEqual('mockId');
+  expect(receivedPayloads[0].state).toEqual('mockState');
+  expect(receivedPayloads[0].basis).toEqual(true);
+});
+
+it('it sets basis to false when intent code is xfer-changes', () => {
+  const mockStream = new MockEventStream();
+  const receivedPayloads: Payload[] = [];
+  const readerUnderTest = new PayloadReader(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-changes", "id": "mockId"}]}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"state": "mockState", "version": 1}',
+  });
+  expect(receivedPayloads.length).toEqual(1);
+  expect(receivedPayloads[0].id).toEqual('mockId');
+  expect(receivedPayloads[0].state).toEqual('mockState');
+  expect(receivedPayloads[0].basis).toEqual(false);
+});
+
+it('it includes multiple types of updates in payload', () => {
+  const mockStream = new MockEventStream();
+  const receivedPayloads: Payload[] = [];
+  const readerUnderTest = new PayloadReader(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('delete-object', {
+    data: '{"kind": "mockKind", "key": "flagB", "version": 123}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"kind": "mockKind", "key": "flagC", "version": 123, "object": {"objectFieldC": "objectValueC"}}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"state": "mockState", "version": 1}',
+  });
+  expect(receivedPayloads.length).toEqual(1);
+  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({ objectFieldA: 'objectValueA' });
+  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({ objectFieldC: 'objectValueC' });
+  expect(receivedPayloads[0].updates[2].deleted).toEqual(undefined);
+});
+
+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, {
+    mockKind: (it) => it, // obj processor that just returns the same obj
+  });
+  readerUnderTest.addPayloadListener((it) => {
+    receivedPayloads.push(it);
+  });
+
+  mockStream.simulateEvent('put-object', {
+    data: '{"kind": "mockKind", "key": "flagShouldIgnore", "version": 123, "object": {"objectFieldShouldIgnore": "objectValueShouldIgnore"}}',
+  });
+  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}',
+  });
+  expect(receivedPayloads.length).toEqual(1);
+  expect(receivedPayloads[0].updates.length).toEqual(1);
+  expect(receivedPayloads[0].updates[0].object).toEqual({ objectFieldA: 'objectValueA' });
+});
+
+it('logs prescribed message when goodbye event is encountered', () => {
+  const mockLogger: LDLogger = {
+    error: jest.fn(),
+    warn: jest.fn(),
+    info: jest.fn(),
+    debug: jest.fn(),
+  };
+  const mockStream = new MockEventStream();
+  const receivedPayloads: Payload[] = [];
+  const readerUnderTest = new PayloadReader(
+    mockStream,
+    {
+      mockKind: (it) => it, // obj processor that just returns the same obj
+    },
+    undefined,
+    mockLogger,
+  );
+  readerUnderTest.addPayloadListener((it) => {
+    receivedPayloads.push(it);
+  });
+
+  mockStream.simulateEvent('goodbye', {
+    data: '{"reason": "Bye"}',
+  });
+
+  expect(receivedPayloads.length).toEqual(0);
+  expect(mockLogger.info).toHaveBeenCalledWith(
+    'Goodbye was received from the LaunchDarkly connection with reason: Bye.',
+  );
+});
+
+it('logs prescribed message when error event is encountered', () => {
+  const mockLogger: LDLogger = {
+    error: jest.fn(),
+    warn: jest.fn(),
+    info: jest.fn(),
+    debug: jest.fn(),
+  };
+  const mockStream = new MockEventStream();
+  const receivedPayloads: Payload[] = [];
+  const readerUnderTest = new PayloadReader(
+    mockStream,
+    {
+      mockKind: (it) => it, // obj processor that just returns the same obj
+    },
+    undefined,
+    mockLogger,
+  );
+  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('error', {
+    data: '{"reason": "Womp womp"}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"state": "mockState", "version": 1}',
+  });
+  expect(receivedPayloads.length).toEqual(0);
+  expect(mockLogger.info).toHaveBeenCalledWith(
+    'An issue was encountered receiving updates for payload mockId with reason: Womp womp. Automatic retry will occur.',
+  );
+});
+
+it('discards partially transferred data when an error is encountered', () => {
+  const mockLogger: LDLogger = {
+    error: jest.fn(),
+    warn: jest.fn(),
+    info: jest.fn(),
+    debug: jest.fn(),
+  };
+  const mockStream = new MockEventStream();
+  const receivedPayloads: Payload[] = [];
+  const readerUnderTest = new PayloadReader(
+    mockStream,
+    {
+      mockKind: (it) => it, // obj processor that just returns the same obj
+    },
+    undefined,
+    mockLogger,
+  );
+  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('error', {
+    data: '{"reason": "Womp womp"}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"state": "mockState", "version": 1}',
+  });
+  mockStream.simulateEvent('server-intent', {
+    data: '{"payloads": [{"code": "xfer-full", "id": "mockId2"}]}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"kind": "mockKind", "key": "flagX", "version": 123, "object": {"objectFieldX": "objectValueX"}}',
+  });
+  mockStream.simulateEvent('delete-object', {
+    data: '{"kind": "mockKind", "key": "flagY", "version": 123}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"kind": "mockKind", "key": "flagZ", "version": 123, "object": {"objectFieldZ": "objectValueZ"}}',
+  });
+  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[0].basis).toEqual(true);
+  expect(receivedPayloads[0].updates.length).toEqual(3);
+  expect(receivedPayloads[0].updates[0].object).toEqual({ objectFieldX: 'objectValueX' });
+  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);
+});
+
+it('silently ignores unrecognized kinds', () => {
+  const mockStream = new MockEventStream();
+  const receivedPayloads: Payload[] = [];
+  const readerUnderTest = new PayloadReader(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('put-object', {
+    data: '{"kind": "ItsMeYourBrotherUnrecognizedKind", "key": "unrecognized", "version": 123, "object": {"unrecognized": "unrecognized"}}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"state": "mockState", "version": 1}',
+  });
+  expect(receivedPayloads.length).toEqual(1);
+  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' });
+});
+
+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, {
+    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"},{"code": "IShouldBeIgnored", "id": "IShouldBeIgnored"}]}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"kind": "mockKind", "key": "flagA", "version": 123, "object": {"objectFieldA": "objectValueA"}}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"kind": "ItsMeYourBrotherUnrecognizedKind", "key": "unrecognized", "version": 123, "object": {"unrecognized": "unrecognized"}}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"state": "mockState", "version": 1}',
+  });
+  expect(receivedPayloads.length).toEqual(1);
+  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' });
+});
diff --git a/packages/shared/common/src/api/platform/EventSource.ts b/packages/shared/common/src/api/platform/EventSource.ts
@@ -1,6 +1,6 @@
 import type { HttpErrorResponse } from './Requests';
 
-export type EventName = 'delete' | 'patch' | 'ping' | 'put';
+export type EventName = string;
 export type EventListener = (event?: { data?: any }) => void;
 export type ProcessStreamResponse = {
   deserializeData: (data: string) => any;

diff --git a/packages/shared/common/src/internal/fdv2/index.ts b/packages/shared/common/src/internal/fdv2/index.ts
@@ -0,0 +1,3 @@
+import { Payload, PayloadListener, PayloadReader, Update } from './payloadReader';
+
+export { Payload, PayloadListener, PayloadReader, Update };
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,3 @@
		import { Payload, PayloadListener, PayloadReader, Update } from './payloadReader';

		export { Payload, PayloadListener, PayloadReader, Update };