GraphQL SSE Distinct Connections support

.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/hapi/__integration-tests__/hapi.spec.ts

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

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/__tests__/graphql-sse.spec.ts

@@ -0,0 +1,188 @@
+import { createSchema, createYoga } from '../src/index.js'
+import { createClient } from 'graphql-sse'
+
+describe('GraphQL over SSE', () => {
+  const schema = createSchema({
+    typeDefs: /* GraphQL */ `
+      type Query {
+        hello: String!
+      }
+      type Subscription {
+        greetings: String!
+        waitForPings: String!
+      }
+    `,
+    resolvers: {
+      Query: {
+        async hello() {
+          return 'world'
+        },
+      },
+      Subscription: {
+        greetings: {
+          async *subscribe() {
+            for (const hi of ['Hi', 'Bonjour', 'Hola', 'Ciao', 'Zdravo']) {
+              yield { greetings: hi }
+            }
+          },
+        },
+        waitForPings: {
+          // eslint-disable-next-line require-yield
+          async *subscribe() {
+            // a ping is issued every 300ms, wait for a few and just return
+            await new Promise((resolve) => setTimeout(resolve, 300 * 3 + 100))
+            return
+          },
+        },
+      },
+    },
+  })
+
+  const yoga = createYoga({
+    schema,
+    legacySse: false,
+    maskedErrors: false,
+  })
+
+  describe('Distinct connections mode', () => {
+    test('should issue pings while connected', async () => {
+      const res = await yoga.fetch(
+        'http://yoga/graphql?query=subscription{waitForPings}',
+        {
+          headers: {
+            accept: 'text/event-stream',
+          },
+        },
+      )
+      expect(res.ok).toBeTruthy()
+      await expect(res.text()).resolves.toMatchInlineSnapshot(`
+        ":
+
+        :
+
+        :
+
+        event: complete
+
+        "
+      `)
+    })
+
+    it('should support single result operations', async () => {
+      const 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) => {
+          let result: unknown
+          client.subscribe(
+            {
+              query: /* GraphQL */ `
+                {
+                  hello
+                }
+              `,
+            },
+            {
+              next: (msg) => (result = msg),
+              error: reject,
+              complete: () => resolve(result),
+            },
+          )
+        }),
+      ).resolves.toMatchInlineSnapshot(`
+        {
+          "data": {
+            "hello": "world",
+          },
+        }
+      `)
+
+      client.dispose()
+    })
+
+    it('should support streaming operations', async () => {
+      const 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(`
+        [
+          {
+            "data": {
+              "greetings": "Hi",
+            },
+          },
+          {
+            "data": {
+              "greetings": "Bonjour",
+            },
+          },
+          {
+            "data": {
+              "greetings": "Hola",
+            },
+          },
+          {
+            "data": {
+              "greetings": "Ciao",
+            },
+          },
+          {
+            "data": {
+              "greetings": "Zdravo",
+            },
+          },
+        ]
+      `)
+
+      client.dispose()
+    })
+
+    it('should report errors through the stream', async () => {
+      const res = await yoga.fetch('http://yoga/graphql?query={nope}', {
+        headers: {
+          accept: 'text/event-stream',
+        },
+      })
+      expect(res.ok).toBeTruthy()
+      await expect(res.text()).resolves.toMatchInlineSnapshot(`
+        "event: next
+        data: {"errors":[{"message":"Cannot query field \\"nope\\" on type \\"Query\\".","locations":[{"line":1,"column":2}]}]}
+
+        event: complete
+
+        "
+      `)
+    })
+  })
+
+  it.todo('Single connections mode')
+})

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"