WIP: GraphQL SSE Single Connection mode

Author: ardatan

packages/graphql-yoga/src/plugins/types.ts

@@ -152,6 +152,8 @@ export interface OnResultProcessEventPayload {
     resultProcessor: ResultProcessor,
     acceptedMediaType: string,
   ): void
+  fetchAPI: FetchAPI
+  endResponse(response: Response): void
 }
 
 export type OnResponseHook<TServerContext> = (

packages/graphql-yoga/src/process-request.ts

@@ -27,6 +27,7 @@ export async function processResult({
   const acceptableMediaTypes: string[] = []
   let acceptedMediaType = '*/*'
 
+  let earlyResponse: Response | undefined
   for (const onResultProcessHook of onResultProcessHooks) {
     await onResultProcessHook({
       request,
@@ -37,7 +38,14 @@ export async function processResult({
         resultProcessor = newResultProcessor
         acceptedMediaType = newAcceptedMimeType
       },
+      fetchAPI,
+      endResponse(response) {
+        earlyResponse = response
+      },
     })
+    if (earlyResponse) {
+      return earlyResponse
+    }
   }
 
   // If no result processor found for this result, return an error

packages/plugins/graphql-sse/__tests__/graphql-sse.spec.ts

@@ -33,37 +33,46 @@ describe('graphql-sse', () => {
   const yoga = createYoga({
     schema,
     plugins: [useGraphQLSSE()],
+    sse: {
+      graphqlSSEDistinctConnections: true,
+    },
     maskedErrors: false,
   })
 
+  let client: ReturnType<typeof createClient>
+
+  afterEach(() => {
+    client?.dispose()
+  })
+
   it('should stream using distinct connection mode', async () => {
-    const client = createClient({
-      url: 'http://yoga/graphql/stream',
+    client = createClient({
+      url: 'http://yoga/graphql',
       fetchFn: yoga.fetch,
       abortControllerImpl: yoga.fetchAPI.AbortController,
       singleConnection: false, // distinct connection mode
       retryAttempts: 0,
     })
 
-    await expect(
-      new Promise((resolve, reject) => {
-        const msgs: unknown[] = []
-        client.subscribe(
-          {
-            query: /* GraphQL */ `
-              subscription {
-                greetings
-              }
-            `,
-          },
-          {
-            next: (msg) => msgs.push(msg),
-            error: reject,
-            complete: () => resolve(msgs),
-          },
-        )
-      }),
-    ).resolves.toMatchInlineSnapshot(`
+    const result = await new Promise((resolve, reject) => {
+      const msgs: unknown[] = []
+      client.subscribe(
+        {
+          query: /* GraphQL */ `
+            subscription {
+              greetings
+            }
+          `,
+        },
+        {
+          next: (msg) => msgs.push(msg),
+          error: reject,
+          complete: () => resolve(msgs),
+        },
+      )
+    })
+
+    expect(result).toMatchInlineSnapshot(`
       [
         {
           "data": {
@@ -92,39 +101,37 @@ describe('graphql-sse', () => {
         },
       ]
     `)
-
-    client.dispose()
   })
 
   it('should stream using single connection and lazy mode', async () => {
-    const client = createClient({
-      url: 'http://yoga/graphql/stream',
+    client = createClient({
+      url: 'http://yoga/graphql',
       fetchFn: yoga.fetch,
       abortControllerImpl: yoga.fetchAPI.AbortController,
       singleConnection: true, // single connection mode
       lazy: true,
       retryAttempts: 0,
     })
 
-    await expect(
-      new Promise((resolve, reject) => {
-        const msgs: unknown[] = []
-        client.subscribe(
-          {
-            query: /* GraphQL */ `
-              subscription {
-                greetings
-              }
-            `,
-          },
-          {
-            next: (msg) => msgs.push(msg),
-            error: reject,
-            complete: () => resolve(msgs),
-          },
-        )
-      }),
-    ).resolves.toMatchInlineSnapshot(`
+    const result = await new Promise((resolve, reject) => {
+      const msgs: unknown[] = []
+      client.subscribe(
+        {
+          query: /* GraphQL */ `
+            subscription {
+              greetings
+            }
+          `,
+        },
+        {
+          next: (msg) => msgs.push(msg),
+          error: reject,
+          complete: () => resolve(msgs),
+        },
+      )
+    })
+
+    expect(result).toMatchInlineSnapshot(`
       [
         {
           "data": {
@@ -153,39 +160,37 @@ describe('graphql-sse', () => {
         },
       ]
     `)
-
-    client.dispose()
   })
 
   it('should stream using single connection and non-lazy mode', async () => {
-    const client = createClient({
-      url: 'http://yoga/graphql/stream',
+    client = createClient({
+      url: 'http://yoga/graphql',
       fetchFn: yoga.fetch,
       abortControllerImpl: yoga.fetchAPI.AbortController,
       singleConnection: true, // single connection mode
       lazy: false,
       retryAttempts: 0,
     })
 
-    await expect(
-      new Promise((resolve, reject) => {
-        const msgs: unknown[] = []
-        client.subscribe(
-          {
-            query: /* GraphQL */ `
-              subscription {
-                greetings
-              }
-            `,
-          },
-          {
-            next: (msg) => msgs.push(msg),
-            error: reject,
-            complete: () => resolve(msgs),
-          },
-        )
-      }),
-    ).resolves.toMatchInlineSnapshot(`
+    const result = await new Promise((resolve, reject) => {
+      const msgs: unknown[] = []
+      client.subscribe(
+        {
+          query: /* GraphQL */ `
+            subscription {
+              greetings
+            }
+          `,
+        },
+        {
+          next: (msg) => msgs.push(msg),
+          error: reject,
+          complete: () => resolve(msgs),
+        },
+      )
+    })
+
+    expect(result).toMatchInlineSnapshot(`
       [
         {
           "data": {
@@ -214,8 +219,6 @@ describe('graphql-sse', () => {
         },
       ]
     `)
-
-    client.dispose()
   })
 
   it('should use CORS settings from the server', async () => {
@@ -231,7 +234,7 @@ describe('graphql-sse', () => {
       maskedErrors: false,
     })
 
-    const res = await yoga.fetch('http://yoga/graphql/stream', {
+    const res = await yoga.fetch('http://yoga/graphql', {
       method: 'OPTIONS',
       headers: {
         origin: 'http://yoga',