feat(client): specify headers for individual subscribe calls

src/client.ts

@@ -233,11 +233,13 @@ export interface Client<SingleConnection extends boolean = false> {
    * function used for dropping the subscription and cleaning up.
    *
    * @param on - The event listener for "distinct connections mode". Note that **no events will be emitted** in "single connection mode"; for that, consider using the event listener in {@link ClientOptions}.
+   * @param headers - Additional request headers to send for this subscribe request on. If a header is specified on the client and here, this one will take precedence.
    */
   subscribe<Data = Record<string, unknown>, Extensions = unknown>(
     request: RequestParams,
     sink: Sink<ExecutionResult<Data, Extensions>>,
     on?: SingleConnection extends true ? never : EventListeners<false>,
+    headers?: SingleConnection extends true ? never : Record<string, string>,
   ): () => void;
   /**
    * Subscribes and iterates over emitted results from an SSE connection
@@ -471,6 +473,7 @@ export function createClient<SingleConnection extends boolean = false>(
     request: RequestParams,
     sink: Sink,
     on?: EventListeners<false>,
+    requestHeaders?: Record<string, string>,
   ) {
     if (!singleConnection) {
       // distinct connections mode
@@ -518,6 +521,7 @@ export function createClient<SingleConnection extends boolean = false>(
               signal: control.signal,
               headers: {
                 ...headers,
+                ...requestHeaders,
                 'content-type': 'application/json; charset=utf-8',
               },
               credentials,

tests/client.test.ts

@@ -69,6 +69,48 @@ it('should use the provided headers', async () => {
   expect(headers.get('x-distinct')).toBe('header');
 });
 
+it('should use the request headers on a subscribe call', async () => {
+  let headers!: Headers;
+  // distinct connections mode
+  const { fetch } = createTFetch({
+    authenticate: (req) => {
+      headers = req.raw.headers;
+      return '';
+    },
+  });
+
+  const distinctConnClient = createClient({
+    singleConnection: false,
+    url: 'http://localhost',
+    fetchFn: fetch,
+    retryAttempts: 0,
+    headers: () => {
+      return {
+        'x-distinct': 'header',
+        'x-overridden': 'client',
+      };
+    },
+  });
+
+  const client = tsubscribe(
+    distinctConnClient,
+    {
+      query: '{ getValue }',
+    },
+    {
+      'x-subscribe': 'header',
+      'x-overridden': 'request',
+    },
+  );
+  await Promise.race([client.throwOnError(), client.waitForComplete()]);
+  client.dispose();
+
+  expect(headers.get('x-distinct')).toBe('header');
+  expect(headers.get('x-subscribe')).toBe('header');
+  // subscribe header should override the client header
+  expect(headers.get('x-overridden')).toBe('request');
+});
+
 it('should supply all valid messages received to onMessage', async () => {
   const { fetch } = createTFetch();
 

tests/utils/tsubscribe.ts

@@ -14,27 +14,36 @@ interface TSubscribe<T> {
 export function tsubscribe<
   SingleConnection extends boolean = false,
   T = unknown,
->(client: Client<SingleConnection>, payload: RequestParams): TSubscribe<T> {
+>(
+  client: Client<SingleConnection>,
+  payload: RequestParams,
+  headers?: SingleConnection extends true ? never : Record<string, string>,
+): TSubscribe<T> {
   const emitter = new EventEmitter();
   const results: ExecutionResult<T, unknown>[] = [];
   let error: unknown,
     completed = false;
-  const dispose = client.subscribe<T>(payload, {
-    next: (value) => {
-      results.push(value);
-      emitter.emit('next');
+  const dispose = client.subscribe<T>(
+    payload,
+    {
+      next: (value) => {
+        results.push(value);
+        emitter.emit('next');
+      },
+      error: (err) => {
+        error = err;
+        emitter.emit('err');
+        emitter.removeAllListeners();
+      },
+      complete: () => {
+        completed = true;
+        emitter.emit('complete');
+        emitter.removeAllListeners();
+      },
     },
-    error: (err) => {
-      error = err;
-      emitter.emit('err');
-      emitter.removeAllListeners();
-    },
-    complete: () => {
-      completed = true;
-      emitter.emit('complete');
-      emitter.removeAllListeners();
-    },
-  });
+    undefined,
+    headers,
+  );
   function waitForError() {
     return new Promise((resolve) => {
       function done() {