GraphQL SSE Single Connection mode

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

@@ -57,7 +57,7 @@ export type Plugin<
      * Use this hook with your own risk. It is still experimental and may change in the future.
      * @internal
      */
-    onResultProcess?: OnResultProcess
+    onResultProcess?: OnResultProcess<TServerContext>
   }
 
 export type OnYogaInitHook<TServerContext extends Record<string, any>> = (
@@ -122,8 +122,8 @@ export interface OnParamsEventPayload {
   fetchAPI: FetchAPI
 }
 
-export type OnResultProcess = (
-  payload: OnResultProcessEventPayload,
+export type OnResultProcess<TServerContext> = (
+  payload: OnResultProcessEventPayload<TServerContext>,
 ) => PromiseOrValue<void>
 
 export type ResultProcessorInput =
@@ -136,7 +136,7 @@ export type ResultProcessor = (
   acceptedMediaType: string,
 ) => PromiseOrValue<Response>
 
-export interface OnResultProcessEventPayload {
+export interface OnResultProcessEventPayload<TServerContext> {
   request: Request
   result: ResultProcessorInput
   setResult(result: ResultProcessorInput): void
@@ -146,6 +146,9 @@ export interface OnResultProcessEventPayload {
     resultProcessor: ResultProcessor,
     acceptedMediaType: string,
   ): void
+  fetchAPI: FetchAPI
+  serverContext: TServerContext
+  endResponse(response: Response): void
 }
 
 export type OnResponseHook<TServerContext> = (

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

@@ -8,25 +8,28 @@ import {
 } from './plugins/types.js'
 import { FetchAPI, GraphQLParams } from './types.js'
 
-export async function processResult({
+export async function processResult<TServerContext>({
   request,
   result,
   fetchAPI,
+  serverContext,
   onResultProcessHooks,
 }: {
   request: Request
   result: ResultProcessorInput
   fetchAPI: FetchAPI
+  serverContext: TServerContext
   /**
    * Response Hooks
    */
-  onResultProcessHooks: OnResultProcess[]
+  onResultProcessHooks: OnResultProcess<TServerContext>[]
 }) {
   let resultProcessor: ResultProcessor | undefined
 
   const acceptableMediaTypes: string[] = []
   let acceptedMediaType = '*/*'
 
+  let earlyResponse: Response | undefined
   for (const onResultProcessHook of onResultProcessHooks) {
     await onResultProcessHook({
       request,
@@ -40,7 +43,15 @@ export async function processResult({
         resultProcessor = newResultProcessor
         acceptedMediaType = newAcceptedMimeType
       },
+      fetchAPI,
+      serverContext,
+      endResponse(response) {
+        earlyResponse = response
+      },
     })
+    if (earlyResponse) {
+      return earlyResponse
+    }
   }
 
   // If no result processor found for this result, return an error

packages/graphql-yoga/src/server.ts

@@ -216,7 +216,7 @@ export class YogaServer<
   >
   private onRequestParseHooks: OnRequestParseHook<TServerContext>[]
   private onParamsHooks: OnParamsHook[]
-  private onResultProcessHooks: OnResultProcess[]
+  private onResultProcessHooks: OnResultProcess<TServerContext>[]
   private maskedErrorsOpts: YogaMaskedErrorOpts | null
   private id: string
 
@@ -356,7 +356,7 @@ export class YogaServer<
       useResultProcessors({
         legacySSE: options?.legacySse !== false,
       }),
-      useErrorHandling((error, request) => {
+      useErrorHandling((error, request, serverContext) => {
         const errors = handleError(error, this.maskedErrorsOpts, this.logger)
 
         const result = {
@@ -368,6 +368,7 @@ export class YogaServer<
           result,
           fetchAPI: this.fetchAPI,
           onResultProcessHooks: this.onResultProcessHooks,
+          serverContext,
         })
       }),
       ...(options?.plugins ?? []),
@@ -603,6 +604,7 @@ export class YogaServer<
       result,
       fetchAPI: this.fetchAPI,
       onResultProcessHooks: this.onResultProcessHooks,
+      serverContext,
     })
   }
 }

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

@@ -33,37 +33,44 @@ describe('graphql-sse', () => {
   const yoga = createYoga({
     schema,
     plugins: [useGraphQLSSE()],
+    legacySse: false,
     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 +99,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 +158,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 +217,6 @@ describe('graphql-sse', () => {
         },
       ]
     `)
-
-    client.dispose()
   })
 
   it('should use CORS settings from the server', async () => {
@@ -231,7 +232,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',