fix(url-loader): handle SSE ping event for Readable

Author: ardatan

.changeset/plenty-nails-shave.md

@@ -2,4 +2,4 @@
 "@graphql-tools/url-loader": patch
 ---
 
-[@graphql-tools/url-loader] Fix SSE ping event
+fix(url-loader): handle SSE ping event correctly

packages/loaders/url/package.json

@@ -43,7 +43,6 @@
     "express-graphql": "0.12.0",
     "graphql-upload": "12.0.0",
     "puppeteer": "11.0.0",
-    "web-streams-polyfill": "^3.1.1",
     "webpack": "5.61.0"
   },
   "dependencies": {

packages/loaders/url/src/event-stream/handleReadable.ts

@@ -8,11 +8,13 @@ export async function* handleReadable(readable: Readable) {
       if (part) {
         const eventStr = part.split('event: ')[1];
         const dataStr = part.split('data: ')[1];
-        const data = JSON.parse(dataStr);
         if (eventStr === 'complete') {
           break outer;
         }
-        yield data.payload || data;
+        if (dataStr) {
+          const data = JSON.parse(dataStr);
+          yield data.payload || data;
+        }
       }
     }
   }

packages/loaders/url/tests/handleEventStreamResponse.ts

@@ -0,0 +1,55 @@
+import { handleEventStreamResponse } from '../src/event-stream/handleEventStreamResponse';
+import { TextEncoder } from 'util';
+
+describe('handleEventStreamResponse', () => {
+  describe('ReadableStream', () => {
+    if (parseInt(process.versions.node.split('.')[0]) < 16) {
+      it('dummy', () => { });
+    }
+    const { TransformStream } = require('stream/web');
+    it('should handle an event with data', async () => {
+      const { readable, writable } = new TransformStream();
+      const encoder = new TextEncoder();
+      const stream = writable.getWriter();
+
+      const generator = await handleEventStreamResponse(readable);
+
+      // stream.write(encoder.encode(':\n\n'));
+      stream.write(encoder.encode('event: complete\n'));
+      stream.write(encoder.encode('data: { "foo": "bar" }\n'));
+      stream.write(encoder.encode('\n'));
+
+      expect((await generator.next()).value).toMatchInlineSnapshot(`
+          Object {
+            "foo": "bar",
+          }
+      `);
+    });
+
+    it('should ignore server pings', async () => {
+      const { readable, writable } = new TransformStream();
+      const encoder = new TextEncoder();
+      const stream = writable.getWriter();
+
+      const readStream = async () => {
+        const generator = await handleEventStreamResponse(readable);
+
+        stream.write(encoder.encode(':\n\n'));
+        stream.write(encoder.encode('event: next\n'));
+        stream.write(encoder.encode('data: { "foo": "bar" }\n\n'));
+
+        return generator.next();
+      };
+
+      await expect(await readStream()).resolves.toMatchInlineSnapshot(`
+      Object {
+        "done": false,
+        "value": Object {
+          "foo": "bar",
+        },
+      }
+    `);
+    });
+  });
+
+})

packages/loaders/url/tests/handleReadableStream.test.ts

@@ -290,12 +290,19 @@ describe('[url-loader] webpack bundle compat', () => {
 
         responseClosed$ = new Promise(resolve => res.once('close', () => resolve(true)));
 
+        const ping = setInterval(() => {
+          // Ping
+          res.write(':\n\n');
+        }, 100);
+
         for (const data of sentDatas) {
           await new Promise(resolve => setTimeout(resolve, 300));
           res.write(`data: ${JSON.stringify(data)}\n\n`);
           await new Promise(resolve => setTimeout(resolve, 300));
         }
 
+        clearInterval(ping);
+
       };
 
       const document = parse(/* GraphQL */ `

packages/loaders/url/tests/url-loader-browser.spec.ts

@@ -700,8 +700,15 @@ input TestInput {
             "Cache-Control": "no-cache",
           });
 
+          const ping = setInterval(() => {
+            // Ping
+            res.write(':\n\n');
+          }, 50);
           sentDatas.forEach(result => sleep(300).then(() => res.write(`data: ${JSON.stringify(result)}\n\n`)));
-          serverResponseEnded$ = new Promise(resolve => res.once('close', () => resolve(true)));
+          serverResponseEnded$ = new Promise(resolve => res.once('close', () => {
+            resolve(true);
+            clearInterval(ping);
+          }));
         });
 
         await new Promise<void>((resolve) => httpServer.listen(serverPort, () => resolve()));