feat: streaming poc #679

kwasniew · 2024-12-02T17:23:20Z

About the changes

This is a reference implementation for the streaming client using SSE (Server-Sent Events).

The changes include:

production code changes in the repository
example how to test the streaming API (we test the API, not the low-level details that are tested in the ES library itself)
example showing user facing property to enable streaming

Chosen library:

we choose a fork of https://github.com/EventSource/eventsource at https://github.com/launchdarkly/js-eventsource
because it provides extra options such as backoff with jitter, read timeout for inactive streams. Both libraries also support HTTP headers that are not part of the SSE spec itself but useful in practice to do auth

Main logic:

customers can specify mode (polling vs streaming) using experimentalMode property. When they opt into streaming we create EventSource and inject it to the repository
in the repository we always make the initial fetch with the HTTP fetch API same as before
however when streaming is enabled (= EventSource is injected to the repository) we never do setTimeout calls for the next fetch and we listen to event called 'unleash-updated' from the streaming API and update flags accordingly
so effectively our event stream is a flag stream and not just a notification stream that requires the client to fetch data on its own. This way we save a network roundtrip and piggyback on the broadcast message data

More comments with the important decisions inline

Important files

Discussion points

coveralls · 2024-12-02T17:25:20Z

coverage: 90.385% (-0.2%) from 90.625%
when pulling c8fda8a on streaming-client
into 4115c39 on main.

kwasniew · 2024-12-02T17:25:51Z

examples/variant.js

  },
+  streaming: true,


opt into streaming

kwasniew · 2024-12-02T17:26:11Z

src/repository/index.ts

@@ -13,6 +13,7 @@ import {
  Segment,
  StrategyTransportInterface,
 } from '../strategy/strategy';
+const EventSource = require('eventsource');


couldn't get it to work with import, only require worked in a quick PoC

kwasniew · 2024-12-02T17:26:36Z

src/repository/index.ts

+      this.eventSource = new EventSource(resolveUrl(this.url, './client/streaming'), {
+        headers: this.headers,
+      } as any);
+      this.eventSource?.addEventListener('message', (event) => {


required EventSource is seen as potentially undefined. Not sure why that's what ? is handling here

kwasniew · 2024-12-02T17:27:13Z

src/repository/index.ts

  }

  timedFetch(interval: number) {
-    if (interval > 0) {
+    if (interval > 0 && !this.streaming) {


this line is all it takes to stop intervals/timeouts and use ES client instead

kwasniew · 2024-12-05T09:51:12Z

examples/streaming.js

+  customHeaders: {
+    Authorization: '943ca9171e2c884c545c5d82417a655fb77cec970cc3b78a8ff87f4406b495d0',
+  },
+  experimentalMode: { type: 'streaming' },


this has to be provided explicitly to opt into streaming. The reason I model it as {type: 'polling'} | {type: 'streaming'} union is to add streaming config option later {type: 'streaming', config: {refresh, backoff}}

kwasniew · 2024-12-05T09:51:34Z

examples/streaming.js

+  experimentalMode: { type: 'streaming' },
+  skipInstanceCountWarning: true,
+});
+client.on('changed', () => {


changes are reflected instantly in the event emitter

kwasniew · 2024-12-05T09:52:00Z

src/repository/index.ts

@@ -13,6 +13,8 @@ import {
  Segment,
  StrategyTransportInterface,
 } from '../strategy/strategy';
+// @ts-expect-error


this library does not provide any types so we have to do this for now until we provide a type ourselves

kwasniew · 2024-12-05T09:52:30Z

src/repository/index.ts

@@ -105,6 +110,7 @@ export default class Repository extends EventEmitter implements EventEmitter {
    bootstrapProvider,
    bootstrapOverride = true,
    storageProvider,
+    eventSource,


we make EventSource injectable so that it's easy to swap in testing for a simple mock

kwasniew · 2024-12-05T09:53:38Z

src/repository/index.ts

+          this.emit(UnleashEvents.Error, err);
+        }
+      });
+      this.eventSource.addEventListener('error', (error: unknown) => {


for now I decided to translate event source errors to warnings in Unleash event API

kwasniew · 2024-12-05T09:54:02Z

src/repository/index.ts

  }

  timedFetch(interval: number) {
-    if (interval > 0) {
+    if (interval > 0 && !this.eventSource) {


do intervals only when eventSource is not provided

kwasniew · 2024-12-05T09:54:17Z

src/repository/index.ts

@@ -398,6 +424,9 @@ Message: ${err.message}`,
      clearTimeout(this.timer);
    }
    this.removeAllListeners();
+    if (this.eventSource) {


we have to remember about closing the event source when Unleash is closed

kwasniew · 2024-12-05T09:54:59Z

src/unleash.ts

@@ -123,6 +126,17 @@ export class Unleash extends EventEmitter {
        tags,
        bootstrapProvider,
        bootstrapOverride,
+        eventSource:
+          experimentalMode?.type === 'streaming'
+            ? new EventSource(resolveUrl(unleashUrl, './client/streaming'), {


we can expose the low level config options to the public API over time, but for now I set those defaults internally

kwasniew · 2024-12-05T10:07:02Z

src/test/repository.test.ts

+    ],
+  };
+  const storageProvider: StorageProvider<ClientFeaturesResponse> = new InMemStorageProvider();
+  const eventSource = {


in-memory mock is trivial to implement. It's just a thin layer on top of the EventEmitter to adjust the API semantic to SSE API

feat: streaming poc

f85d1f9

kwasniew commented Dec 2, 2024

View reviewed changes

examples/variant.js Outdated

},

streaming: true,

Copy link

Contributor Author

kwasniew Dec 2, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

opt into streaming

kwasniew commented Dec 2, 2024

View reviewed changes

kwasniew marked this pull request as draft December 2, 2024 17:33

nunogois and others added 6 commits December 3, 2024 18:04

update current state based on SSE event

acc0018

feat: dedicated event type and cleanup

6347977

feat: close stream

a5daa3d

test: streaming

43128b1

chore: streaming example

099093b

chore: streaming example

ca6cc15

kwasniew commented Dec 5, 2024

View reviewed changes

kwasniew marked this pull request as ready for review December 5, 2024 10:03

kwasniew commented Dec 5, 2024

View reviewed changes

GitHub Actions Bot and others added 3 commits December 5, 2024 11:45

v6.2.0-alpha.0

64e73a7

chore: sync with main

5a645e3

v6.3.0-alpha.0

c8fda8a

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: streaming poc #679

feat: streaming poc #679

kwasniew commented Dec 2, 2024 •

edited

Loading

coveralls commented Dec 2, 2024 •

edited

Loading

kwasniew Dec 2, 2024

kwasniew Dec 2, 2024

kwasniew Dec 2, 2024

kwasniew Dec 2, 2024

kwasniew Dec 5, 2024

kwasniew Dec 5, 2024

kwasniew Dec 5, 2024

kwasniew Dec 5, 2024

kwasniew Dec 5, 2024 •

edited

Loading

kwasniew Dec 5, 2024

kwasniew Dec 5, 2024

kwasniew Dec 5, 2024

kwasniew Dec 5, 2024

feat: streaming poc #679

Are you sure you want to change the base?

feat: streaming poc #679

Conversation

kwasniew commented Dec 2, 2024 • edited Loading

About the changes

Important files

Discussion points

coveralls commented Dec 2, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kwasniew Dec 5, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kwasniew commented Dec 2, 2024 •

edited

Loading

coveralls commented Dec 2, 2024 •

edited

Loading

kwasniew Dec 5, 2024 •

edited

Loading