GraphQL SSE Distinct Connections support

Author: ardatan

.changeset/tricky-teachers-sin.md

@@ -0,0 +1,5 @@
+---
+'graphql-yoga': minor
+---
+
+GraphQL SSE Distinct Connections mode support with `sse.graphqlSSEDistinctConnections` flag

examples/fastify/__integration-tests__/fastify.spec.ts

@@ -161,6 +161,8 @@ describe('fastify example integration', () => {
 
       data: {"data":{"countdown":0}}
 
+      event: complete
+
       "
     `)
   })
@@ -202,6 +204,8 @@ describe('fastify example integration', () => {
 
       data: {"data":{"countdown":0}}
 
+      event: complete
+
       "
     `)
   })

examples/generic-auth/__integration-tests__/generic-auth.spec.ts

@@ -70,7 +70,7 @@ describe('graphql-auth example integration', () => {
     for await (const chunk of response.body!) {
       const chunkString = Buffer.from(chunk).toString('utf-8')
       if (chunkString.includes('data:')) {
-        expect(chunkString.trim()).toBe('data: {"data":{"public":"hi"}}')
+        expect(chunkString.trim()).toContain('data: {"data":{"public":"hi"}}')
         break
       }
     }
@@ -91,7 +91,7 @@ describe('graphql-auth example integration', () => {
     for await (const chunk of response.body!) {
       const chunkStr = Buffer.from(chunk).toString('utf-8')
       if (chunkStr.startsWith('data:')) {
-        expect(chunkStr.trim()).toBe(
+        expect(chunkStr.trim()).toContain(
           'data: {"data":{"requiresAuth":"hi foo@foo.com"}}',
         )
         break
@@ -112,7 +112,7 @@ describe('graphql-auth example integration', () => {
     for await (const chunk of response.body!) {
       const chunkStr = Buffer.from(chunk).toString('utf-8')
       if (chunkStr.startsWith('data:')) {
-        expect(chunkStr.trim()).toBe(
+        expect(chunkStr.trim()).toContain(
           'data: {"data":null,"errors":[{"message":"Accessing \'Subscription.requiresAuth\' requires authentication.","locations":[{"line":1,"column":14}]}]}',
         )
         break

examples/node-ts/__integration-tests__/node-ts.spec.ts

@@ -24,6 +24,8 @@ describe('node-ts example integration', () => {
     expect(await response.text()).toMatchInlineSnapshot(`
       "data: {"errors":[{"message":"Subscriptions have been disabled"}]}
 
+      event: complete
+
       "
     `)
   })

examples/pothos/__integration-tests__/pothos.spec.ts

@@ -32,6 +32,8 @@ describe('pothos example integration', () => {
 
       data: {"data":{"greetings":"Zdravo"}}
 
+      event: complete
+
       "
     `)
   })

packages/graphql-yoga/__integration-tests__/graphql-sse-client.spec.ts

@@ -0,0 +1,89 @@
+import { ExecutionResult } from 'graphql'
+import { createClient } from 'graphql-sse'
+import { createSchema, createYoga } from '../src'
+
+describe('GraphQL SSE Client compatibility', () => {
+  describe('Distinct Connections', () => {
+    const yoga = createYoga({
+      sse: {
+        graphqlSSEDistinctConnections: true,
+      },
+      schema: createSchema({
+        typeDefs: /* GraphQL */ `
+          type Query {
+            hello: String
+          }
+          type Subscription {
+            greetings: String
+          }
+        `,
+        resolvers: {
+          Query: {
+            hello: () => 'world',
+          },
+          Subscription: {
+            greetings: {
+              async *subscribe() {
+                yield { greetings: 'Hi' }
+                await new Promise((resolve) => setTimeout(resolve, 300))
+                yield { greetings: 'Bonjour' }
+                await new Promise((resolve) => setTimeout(resolve, 300))
+                yield { greetings: 'Hola' }
+                await new Promise((resolve) => setTimeout(resolve, 300))
+                yield { greetings: 'Ciao' }
+                await new Promise((resolve) => setTimeout(resolve, 300))
+                yield { greetings: 'Hallo' }
+              },
+            },
+          },
+        },
+      }),
+    })
+    const client = createClient({
+      url: 'http://localhost:4000/graphql',
+      fetchFn: yoga.fetch,
+      abortControllerImpl: yoga.fetchAPI.AbortController,
+      retryAttempts: 0,
+    })
+    let unsubscribe: () => void
+    afterAll(() => {
+      unsubscribe?.()
+      client.dispose()
+    })
+    it('handle queries', async () => {
+      const result = await new Promise((resolve, reject) => {
+        let result: ExecutionResult<Record<string, unknown>, unknown>
+        unsubscribe = client.subscribe(
+          {
+            query: '{ hello }',
+          },
+          {
+            next: (data) => (result = data),
+            error: reject,
+            complete: () => resolve(result),
+          },
+        )
+      })
+
+      expect(result).toEqual({ data: { hello: 'world' } })
+    })
+    it('handle subscriptions', async () => {
+      const onNext = jest.fn()
+
+      await new Promise<void>((resolve, reject) => {
+        unsubscribe = client.subscribe(
+          {
+            query: 'subscription { greetings }',
+          },
+          {
+            next: onNext,
+            error: reject,
+            complete: resolve,
+          },
+        )
+      })
+
+      expect(onNext).toBeCalledTimes(5) // we say "Hi" in 5 languages
+    })
+  })
+})

packages/graphql-yoga/__tests__/subscriptions.spec.ts

@@ -143,21 +143,24 @@ describe('Subscription', () => {
     }
 
     expect(results).toMatchInlineSnapshot(`
-        [
-          ":
+      [
+        ":
+
+      ",
+        ":
 
-        ",
-          ":
+      ",
+        ":
 
-        ",
-          ":
+      ",
+        "data: {"data":{"hi":"hi"}}
 
-        ",
-          "data: {"data":{"hi":"hi"}}
+      ",
+        "event: complete
 
-        ",
-        ]
-      `)
+      ",
+      ]
+    `)
   })
 
   test('should issue pings event if event source never publishes anything', async () => {

packages/graphql-yoga/package.json

@@ -73,6 +73,7 @@
     "graphql": "^16.0.1",
     "graphql-http": "^1.7.2",
     "graphql-scalars": "1.20.4",
+    "graphql-sse": "2.0.0",
     "html-minifier-terser": "7.1.0",
     "json-bigint-patch": "0.0.8",
     "puppeteer": "19.6.0"

packages/graphql-yoga/src/plugins/resultProcessor/push.ts

@@ -0,0 +1,86 @@
+import { ExecutionResult } from 'graphql'
+import { isAsyncIterable } from '@envelop/core'
+
+import { getResponseInitByRespectingErrors } from '../../error.js'
+import { FetchAPI, MaybeArray } from '../../types.js'
+import { ResultProcessor, ResultProcessorInput } from '../types.js'
+import { jsonStringifyResultWithoutInternals } from './stringify.js'
+
+export interface SSEProcessorOptions {
+  legacySSE: boolean
+}
+
+export function getSSEProcessor(opts: SSEProcessorOptions): ResultProcessor {
+  return function processSSEResult(
+    result: ResultProcessorInput,
+    fetchAPI: FetchAPI,
+  ): Response {
+    let pingIntervalMs = 12_000
+
+    // for testing the pings, reduce the timeout
+    if (globalThis.process?.env?.NODE_ENV === 'test') {
+      pingIntervalMs = 300
+    }
+
+    const headersInit = {
+      'Content-Type': 'text/event-stream',
+      Connection: 'keep-alive',
+      'Cache-Control': 'no-cache',
+      'Content-Encoding': 'none',
+    }
+
+    const responseInit = getResponseInitByRespectingErrors(result, headersInit)
+
+    let iterator: AsyncIterator<MaybeArray<ExecutionResult>>
+
+    let pingInterval: number
+    const textEncoder = new fetchAPI.TextEncoder()
+    const readableStream = new fetchAPI.ReadableStream({
+      start(controller) {
+        // ping client every 12 seconds to keep the connection alive
+        pingInterval = setInterval(() => {
+          if (!controller.desiredSize) {
+            clearInterval(pingInterval)
+            return
+          }
+          controller.enqueue(textEncoder.encode(':\n\n'))
+        }, pingIntervalMs) as unknown as number
+
+        if (isAsyncIterable(result)) {
+          iterator = result[Symbol.asyncIterator]()
+        } else {
+          let finished = false
+          iterator = {
+            next: () => {
+              if (finished) {
+                return Promise.resolve({ done: true, value: null })
+              }
+              finished = true
+              return Promise.resolve({ done: false, value: result })
+            },
+          }
+        }
+      },
+      async pull(controller) {
+        const { done, value } = await iterator.next()
+        if (value != null) {
+          if (!opts.legacySSE) {
+            controller.enqueue(textEncoder.encode(`event: next\n`))
+          }
+          const chunk = jsonStringifyResultWithoutInternals(value)
+          controller.enqueue(textEncoder.encode(`data: ${chunk}\n\n`))
+        }
+        if (done) {
+          controller.enqueue(textEncoder.encode(`event: complete\n\n`))
+          clearInterval(pingInterval)
+          controller.close()
+        }
+      },
+      async cancel(e) {
+        clearInterval(pingInterval)
+        await iterator.return?.(e)
+      },
+    })
+    return new fetchAPI.Response(readableStream, responseInit)
+  }
+}