feat: Adds StreamingProcessor for FDv2 to sdk-server package.

packages/shared/common/__tests__/internal/fdv2/PayloadReader.test.ts

@@ -0,0 +1,316 @@
+import { EventListener, EventName, LDLogger } from '../../../src/api';
+import { EventStream, Payload, PayloadReader } from '../../../src/internal/fdv2/payloadReader';
+
+class MockEventStreamm implements EventStream {
+  private _listeners: {
+    [event: 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 MockEventStreamm();
+  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: '{"data": {"payloads": [{"intentCode": "xfer-full", "id": "mockId"}]}}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"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 MockEventStreamm();
+  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: '{"data": {"payloads": [{"intentCode": "xfer-changes", "id": "mockId"}]}}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"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 MockEventStreamm();
+  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: '{"data": {"payloads": [{"intentCode": "xfer-full", "id": "mockId"}]}}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"data": {"kind": "mockKind", "key": "flagA", "version": 123, "object": {"objectFieldA": "objectValueA"}}}',
+  });
+  mockStream.simulateEvent('delete-object', {
+    data: '{"data": {"kind": "mockKind", "key": "flagB", "version": 123}}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"data": {"kind": "mockKind", "key": "flagC", "version": 123, "object": {"objectFieldC": "objectValueC"}}}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"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 MockEventStreamm();
+  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: '{"data": {"kind": "mockKind", "key": "flagShouldIgnore", "version": 123, "object": {"objectFieldShouldIgnore": "objectValueShouldIgnore"}}}',
+  });
+  mockStream.simulateEvent('server-intent', {
+    data: '{"data": {"payloads": [{"intentCode": "xfer-full", "id": "mockId"}]}}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"data": {"kind": "mockKind", "key": "flagA", "version": 123, "object": {"objectFieldA": "objectValueA"}}}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"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 MockEventStreamm();
+  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: '{"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 MockEventStreamm();
+  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: '{"data": {"payloads": [{"intentCode": "xfer-full", "id": "mockId"}]}}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"data": {"kind": "mockKind", "key": "flagA", "version": 123, "object": {"objectFieldA": "objectValueA"}}}',
+  });
+  mockStream.simulateEvent('error', {
+    data: '{"data": {"reason": "Womp womp"}}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"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 MockEventStreamm();
+  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: '{"data": {"payloads": [{"intentCode": "xfer-full", "id": "mockId"}]}}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"data": {"kind": "mockKind", "key": "flagA", "version": 123, "object": {"objectFieldA": "objectValueA"}}}',
+  });
+  mockStream.simulateEvent('error', {
+    data: '{"data": {"reason": "Womp womp"}}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"data": {"state": "mockState", "version": 1}}',
+  });
+  mockStream.simulateEvent('server-intent', {
+    data: '{"data": {"payloads": [{"intentCode": "xfer-full", "id": "mockId2"}]}}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"data": {"kind": "mockKind", "key": "flagX", "version": 123, "object": {"objectFieldX": "objectValueX"}}}',
+  });
+  mockStream.simulateEvent('delete-object', {
+    data: '{"data": {"kind": "mockKind", "key": "flagY", "version": 123}}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"data": {"kind": "mockKind", "key": "flagZ", "version": 123, "object": {"objectFieldZ": "objectValueZ"}}}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"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 MockEventStreamm();
+  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: '{"data": {"payloads": [{"intentCode": "xfer-full", "id": "mockId"}]}}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"data": {"kind": "mockKind", "key": "flagA", "version": 123, "object": {"objectFieldA": "objectValueA"}}}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"data": {"kind": "ItsMeYourBrotherUnrecognizedKind", "key": "unrecognized", "version": 123, "object": {"unrecognized": "unrecognized"}}}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"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 MockEventStreamm();
+  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: '{"data": {"payloads": [{"intentCode": "xfer-full", "id": "mockId"},{"intentCode": "IShouldBeIgnored", "id": "IShouldBeIgnored"}]}}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"data": {"kind": "mockKind", "key": "flagA", "version": 123, "object": {"objectFieldA": "objectValueA"}}}',
+  });
+  mockStream.simulateEvent('put-object', {
+    data: '{"data": {"kind": "ItsMeYourBrotherUnrecognizedKind", "key": "unrecognized", "version": 123, "object": {"unrecognized": "unrecognized"}}}',
+  });
+  mockStream.simulateEvent('payload-transferred', {
+    data: '{"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' });
+});

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;

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

@@ -0,0 +1,3 @@
+import { Payload, PayloadListener, PayloadReader, Update } from './payloadReader';
+
+export { Payload, PayloadListener, PayloadReader, Update };