[@graphql-tools/url-loader] Fix SSE ping event (#3821)

* Fix ping event

* Create plenty-nails-shave.md

* Add test for readable stream

* sorted packages in package.json (done by npm)

* rename typo in filename

Author: dan-lee

.changeset/plenty-nails-shave.md

@@ -0,0 +1,5 @@
+---
+"@graphql-tools/url-loader": patch
+---
+
+[@graphql-tools/url-loader] Fix SSE ping event

packages/loaders/url/package.json

@@ -36,13 +36,14 @@
   },
   "devDependencies": {
     "@types/express": "4.17.13",
-    "@types/valid-url": "1.0.3",
     "@types/extract-files": "8.1.1",
+    "@types/valid-url": "1.0.3",
+    "babel-loader": "8.2.3",
     "express": "4.17.1",
     "express-graphql": "0.12.0",
     "graphql-upload": "12.0.0",
     "puppeteer": "11.0.0",
-    "babel-loader": "8.2.3",
+    "web-streams-polyfill": "^3.1.1",
     "webpack": "5.61.0"
   },
   "dependencies": {
@@ -57,14 +58,14 @@
     "extract-files": "11.0.0",
     "graphql-sse": "^1.0.1",
     "graphql-ws": "^5.4.1",
-    "meros": "1.1.4",
     "isomorphic-ws": "4.0.1",
+    "meros": "1.1.4",
     "subscriptions-transport-ws": "^0.11.0",
     "sync-fetch": "0.3.1",
     "tslib": "~2.3.0",
     "valid-url": "1.0.9",
-    "ws": "8.2.3",
-    "value-or-promise": "1.0.11"
+    "value-or-promise": "1.0.11",
+    "ws": "8.2.3"
   },
   "publishConfig": {
     "access": "public",

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

@@ -95,13 +95,16 @@ export async function* handleReadableStream(stream: ReadableStream<Uint8Array>)
         const line = buffer.subarray(lineStart, lineEnd);
         if (line.length === 0) {
           // empty line denotes end of message. Trigger the callback and start a new message:
-          yield JSON.parse(message.data);
-          message = {
-            data: '',
-            event: '',
-            id: '',
-            retry: undefined,
-          };
+          if (message.event || message.data) {
+            // NOT a server ping (":\n\n")
+            yield JSON.parse(message.data);
+            message = {
+              data: '',
+              event: '',
+              id: '',
+              retry: undefined,
+            };
+          }
         } else if (fieldLength > 0) {
           // exclude comments and lines with no values
           // line is of format "<field>:<value>" or "<field>: <value>"

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

@@ -0,0 +1,49 @@
+import { handleReadableStream } from '../src/event-stream/handleReadableStream';
+import { TransformStream } from 'web-streams-polyfill/ponyfill';
+import { TextEncoder } from 'util';
+
+describe('handleReadableStream', () => {
+  it('should handle an event with data', async () => {
+    const { readable, writable } = new TransformStream();
+    const encoder = new TextEncoder();
+    const stream = writable.getWriter();
+
+    const generator = handleReadableStream(readable as any);
+
+    // 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 = () => {
+      const generator = handleReadableStream(readable as any);
+
+      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(readStream()).resolves.toMatchInlineSnapshot(`
+      Object {
+        "done": false,
+        "value": Object {
+          "foo": "bar",
+        },
+      }
+    `);
+  });
+});