fix: apply client.registerAPIRequestMiddleware to SDK streaming requests (#163)

Author: EmilianoSanchez

README.md

@@ -81,21 +81,21 @@ const client = initialize('00000000-1111-2222-3333-444444444444', {
 By default, Harness Feature Flags SDK has streaming enabled and polling enabled. Both modes can be toggled according to your preference using the SDK's configuration.
 
 ### Streaming Mode
-Streaming mode establishes a continuous connection between your application and the Feature Flags service. 
-This allows for real-time updates on feature flags without requiring periodic checks. 
+Streaming mode establishes a continuous connection between your application and the Feature Flags service.
+This allows for real-time updates on feature flags without requiring periodic checks.
 If an error occurs while streaming and `pollingEnabled` is set to `true`,
-the SDK will automatically fall back to polling mode until streaming can be reestablished. 
+the SDK will automatically fall back to polling mode until streaming can be reestablished.
 If `pollingEnabled` is `false`, streaming will attempt to reconnect without falling back to polling.
 
 ### Polling Mode
 In polling mode, the SDK will periodically check with the Feature Flags service to retrieve updates for feature flags. The frequency of these checks can be adjusted using the SDK's configurations.
 
 ### No Streaming or Polling
-If both streaming and polling modes are disabled (`streamEnabled: false` and `pollingEnabled: false`), 
-the SDK will not automatically fetch feature flag updates after the initial fetch. 
+If both streaming and polling modes are disabled (`streamEnabled: false` and `pollingEnabled: false`),
+the SDK will not automatically fetch feature flag updates after the initial fetch.
 This means that after the initial load, any changes made to the feature flags on the Harness server will not be reflected in the application until the SDK is re-initialized or one of the modes is re-enabled.
 
-This configuration might be useful in specific scenarios where you want to ensure a consistent set of feature flags 
+This configuration might be useful in specific scenarios where you want to ensure a consistent set of feature flags
 for a session or when the application operates in an environment where regular updates are not necessary. However, it's essential to be aware that this configuration can lead to outdated flag evaluations if the flags change on the server.
 
 To configure the modes:
@@ -127,9 +127,9 @@ You can configure the maximum number of streaming retries before the SDK stops a
 ```typescript
 const options = {
   maxRetries: 5, // Set the maximum number of retries for streaming. Default is Infinity.
-  streamEnabled: true, 
-  pollingEnabled: true, 
-  pollingInterval: 60000, 
+  streamEnabled: true,
+  pollingEnabled: true,
+  pollingInterval: 60000,
 }
 
 const client = initialize(
@@ -173,7 +173,7 @@ client.on(Event.DISCONNECTED, () => {
 })
 
 client.on(Event.CONNECTED, () => {
-  // Event happens when connection has been lost and reestablished 
+  // Event happens when connection has been lost and reestablished
 })
 
 client.on(Event.POLLING, () => {
@@ -238,7 +238,7 @@ For the example above, if the flag identifier 'Dark_Theme' is not found, result
 }
 ```
 
-If you do not need to know the default variation was returned: 
+If you do not need to know the default variation was returned:
 
 ```typescript
 const variationValue = client.variation('Dark_Theme', false) // second argument is default value when variation does not exist
@@ -257,7 +257,7 @@ For the example above:
   3. Wrong project API key being used
 
 #### Listening for the `ERROR_DEFAULT_VARIATION_RETURNED` event
-You can also listen for the `ERROR_DEFAULT_VARIATION_RETURNED` event, which is emitted whenever a default variation is returned because the flag has not been found in the cache. This is useful for logging or taking other action when a flag is not found. 
+You can also listen for the `ERROR_DEFAULT_VARIATION_RETURNED` event, which is emitted whenever a default variation is returned because the flag has not been found in the cache. This is useful for logging or taking other action when a flag is not found.
 
 Example of listening for the event:
 
@@ -386,7 +386,7 @@ const client = initialize(
 ```
 
 ## API Middleware
-The `registerAPIRequestMiddleware` function allows you to register a middleware function to manipulate the payload (URL, body and headers) of API requests after the AUTH call has successfully completed
+The `registerAPIRequestMiddleware` function allows you to register a middleware function to manipulate the payload (URL, body and headers) of API requests.
 
 ```typescript
 function abortControllerMiddleware([url, options]) {

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@harnessio/ff-javascript-client-sdk",
-  "version": "1.31.1",
+  "version": "1.31.2",
   "author": "Harness",
   "license": "Apache-2.0",
   "main": "dist/sdk.cjs.js",

package.json

@@ -246,4 +246,29 @@ describe('Streamer', () => {
     expect(mockEventBus.emit).toHaveBeenCalledWith(Event.CONNECTED)
     expect(mockXHR.send).toHaveBeenCalledTimes(4) // Should attempt to reconnect 3 times before succeeding
   })
+
+  it('should apply middleware to requests', () => {
+    const streamer = getStreamer()
+
+    const newHeader: string = 'header-value'
+
+    streamer.registerAPIRequestMiddleware(args => {
+      args[0] = 'http://test/stream2'
+      args[1].headers = { ...args[1].headers, newHeader }
+      return args
+    })
+
+    streamer.start()
+    expect(mockXHR.open).toHaveBeenCalledWith('GET', 'http://test/stream2')
+    expect(mockXHR.setRequestHeader).toHaveBeenCalledTimes(5)
+    expect(mockXHR.setRequestHeader).toHaveBeenCalledWith('Test-Header', 'value')
+    expect(mockXHR.setRequestHeader).toHaveBeenCalledWith('Cache-Control', 'no-cache')
+    expect(mockXHR.setRequestHeader).toHaveBeenCalledWith('Accept', 'text/event-stream')
+    expect(mockXHR.setRequestHeader).toHaveBeenCalledWith('API-Key', 'test-api-key')
+    expect(mockXHR.setRequestHeader).toHaveBeenCalledWith('newHeader', 'header-value')
+    expect(mockXHR.send).toHaveBeenCalled()
+
+    mockXHR.onprogress({} as ProgressEvent)
+    expect(mockEventBus.emit).toHaveBeenCalledWith(Event.CONNECTED)
+  })
 })

src/__tests__/stream.test.ts

@@ -23,10 +23,9 @@ import { getVariation } from './variation'
 import Poller from './poller'
 import { createCacheIdSeed, getCache } from './cache'
 
-const SDK_VERSION = '1.31.1'
+const SDK_VERSION = '1.31.2'
 const SDK_INFO = `Javascript ${SDK_VERSION} Client`
 const METRICS_VALID_COUNT_INTERVAL = 500
-const fetch = globalThis.fetch
 
 // Flag to detect is Proxy is supported (not under IE 11)
 const hasProxy = !!globalThis.Proxy
@@ -35,12 +34,13 @@ const initialize = (apiKey: string, target: Target, options?: Options): Result =
   let closed = false
   let environment: string
   let clusterIdentifier: string
-  let eventSource: any
+  let eventSource: Streamer | undefined
   let poller: Poller
   let jwtToken: string
   let metricsSchedulerId: number
   let standardHeaders: Record<string, string> = {}
-  let fetchWithMiddleware = addMiddlewareToFetch(args => args)
+  let defaultMiddleware: APIRequestMiddleware = args => args
+  let fetchWithMiddleware = addMiddlewareToFetch(defaultMiddleware)
   let lastCacheRefreshTime = 0
   let initialised = false
   // We need to pause metrics in certain situations, such as when we are doing the initial evaluation load, and when
@@ -594,7 +594,8 @@ const initialize = (apiKey: string, target: Target, options?: Options): Result =
           handleSegmentEvent(event)
         }
       },
-      configurations.maxStreamRetries
+      configurations.maxStreamRetries,
+      defaultMiddleware
     )
     eventSource.start()
   }
@@ -678,7 +679,9 @@ const initialize = (apiKey: string, target: Target, options?: Options): Result =
   }
 
   const registerAPIRequestMiddleware = (middleware: APIRequestMiddleware): void => {
+    defaultMiddleware = middleware
     fetchWithMiddleware = addMiddlewareToFetch(middleware)
+    if (eventSource) eventSource.registerAPIRequestMiddleware(middleware)
   }
 
   const refreshEvaluations = () => {

src/index.ts

@@ -1,4 +1,4 @@
-import { Event, type Options, StreamEvent } from './types'
+import { APIRequestMiddleware, Event, type Options, StreamEvent } from './types'
 import { getRandom } from './utils'
 import type Poller from './poller'
 import type { Emitter } from 'mitt'
@@ -24,9 +24,14 @@ export class Streamer {
     private logDebug: (...data: any[]) => void,
     private logError: (...data: any[]) => void,
     private eventCallback: (e: StreamEvent) => void,
-    private maxRetries: number
+    private maxRetries: number,
+    private middleware?: APIRequestMiddleware
   ) {}
 
+  registerAPIRequestMiddleware(middleware: APIRequestMiddleware): void {
+    this.middleware = middleware
+  };
+
   start() {
     const processData = (data: any): void => {
       data.toString().split(/\r?\n/).forEach(processLine)
@@ -94,13 +99,19 @@ export class Streamer {
       onDisconnect()
     }
 
-    const sseHeaders: Record<string, string> = {
+    let sseHeaders: Record<string, string> = {
       'Cache-Control': 'no-cache',
       Accept: 'text/event-stream',
       'API-Key': this.apiKey,
       ...this.standardHeaders
     }
 
+    if (this.middleware) {
+      const [url, options] = this.middleware([this.url, { headers: sseHeaders }])
+      this.url = url as string
+      sseHeaders = options?.headers as Record<string, string> || {}
+    }
+
     this.logDebugMessage('SSE HTTP start request', this.url)
 
     this.xhr = new XMLHttpRequest()