gateway: handle sse events

Signed-off-by: Vincenzo Scamporlino <vincenzos@spotify.com>

plugins/gateway-backend/package.json

@@ -45,6 +45,8 @@
     "@backstage/backend-defaults": "workspace:^",
     "@backstage/backend-test-utils": "workspace:^",
     "@backstage/cli": "workspace:^",
-    "@types/express": "^4.17.6"
+    "@types/express": "^4.17.6",
+    "eventsource": "^3.0.6",
+    "wait-for-expect": "^3.0.2"
   }
 }

plugins/gateway-backend/src/plugin.test.ts

@@ -22,6 +22,8 @@ import {
   createServiceFactory,
 } from '@backstage/backend-plugin-api';
 import { Router } from 'express';
+import { EventSource } from 'eventsource';
+import waitForExpect from 'wait-for-expect';
 
 describe('gateway', () => {
   let backend: Backend;
@@ -47,6 +49,12 @@ describe('gateway', () => {
   });
 
   const discovery = mockServices.discovery.mock();
+  discovery.getBaseUrl.mockImplementation(async (pluginId: string) => {
+    if (pluginId === 'external-plugin') {
+      return 'http://localhost:7778/api/external-plugin';
+    }
+    return `http://localhost:7777/api/${pluginId}`;
+  });
 
   beforeAll(async () => {
     backend = createBackend();
@@ -97,6 +105,30 @@ describe('gateway', () => {
                 res.json({ bar: true });
               });
 
+              router.get('/endpoint-sse', async (_req, res) => {
+                res.setHeader('Content-Type', 'text/event-stream');
+                res.setHeader('Cache-Control', 'no-cache');
+                res.setHeader('Connection', 'keep-alive');
+
+                // Send periodic updates
+                let data = { timestamp: new Date().toISOString() };
+                res.write(`data: ${JSON.stringify(data)}\n\n`);
+                res.flush();
+                await pause(50);
+
+                data = { timestamp: new Date().toISOString() };
+                res.write(`data: ${JSON.stringify(data)}\n\n`);
+                res.flush();
+                await pause(50);
+
+                data = { timestamp: new Date().toISOString() };
+                res.write(`data: ${JSON.stringify(data)}\n\n`);
+                res.flush();
+                await pause(50);
+
+                res.destroy();
+              });
+
               httpRouter.use(router);
             },
           });
@@ -120,12 +152,6 @@ describe('gateway', () => {
   });
 
   it('should proxy requests for unknown plugins', async () => {
-    discovery.getBaseUrl.mockImplementation(async (pluginId: string) => {
-      if (pluginId === 'external-plugin') {
-        return 'http://localhost:7778/api/external-plugin';
-      }
-      return `http://localhost:7777/api/${pluginId}`;
-    });
     const response = await fetch(
       'http://localhost:7777/api/external-plugin/foo',
     );
@@ -134,4 +160,24 @@ describe('gateway', () => {
     const data = await response.json();
     expect(data).toEqual({ bar: true });
   });
+
+  it('should close the response for sse connections', async () => {
+    const eventSource = new EventSource(
+      'http://localhost:7777/api/external-plugin/endpoint-sse',
+    );
+
+    const mockOnMessage = jest.fn();
+    const mockOnError = jest.fn();
+    eventSource.addEventListener('message', mockOnMessage);
+    eventSource.addEventListener('error', mockOnError);
+
+    await waitForExpect(() => {
+      expect(mockOnMessage).toHaveBeenCalledTimes(3);
+      expect(mockOnError).toHaveBeenCalled();
+    });
+  });
 });
+
+function pause(ms: number) {
+  return new Promise(resolve => setTimeout(resolve, ms));
+}

plugins/gateway-backend/src/router.ts

@@ -41,6 +41,16 @@ export function createRouter({
       const pluginId = req.params.pluginId;
       return discovery.getBaseUrl(pluginId);
     },
+    on: {
+      proxyRes(proxyRes, _req, res) {
+        // https://github.com/chimurai/http-proxy-middleware/discussions/765
+        proxyRes.on('close', () => {
+          if (!res.writableEnded) {
+            res.end();
+          }
+        });
+      },
+    },
   });
 
   return function proxyMiddleware(

yarn.lock

@@ -6745,8 +6745,10 @@ __metadata:
     "@opentelemetry/api": "npm:^1.9.0"
     "@opentelemetry/core": "npm:^1.29.0"
     "@types/express": "npm:^4.17.6"
+    eventsource: "npm:^3.0.6"
     express: "npm:^4.17.1"
     http-proxy-middleware: "npm:^3.0.3"
+    wait-for-expect: "npm:^3.0.2"
   languageName: unknown
   linkType: soft
 
@@ -29241,6 +29243,22 @@ __metadata:
   languageName: node
   linkType: hard
 
+"eventsource-parser@npm:^3.0.1":
+  version: 3.0.1
+  resolution: "eventsource-parser@npm:3.0.1"
+  checksum: 10/2730c54c3cb47d55d2967f2ece843f9fc95d8a11c2fef6fece8d17d9080193cbe3cd9ac7b04a325977f63cbf8c1664fdd0512dec1aec601666a5c5bd8564b61f
+  languageName: node
+  linkType: hard
+
+"eventsource@npm:^3.0.6":
+  version: 3.0.6
+  resolution: "eventsource@npm:3.0.6"
+  dependencies:
+    eventsource-parser: "npm:^3.0.1"
+  checksum: 10/ac08c7d1b21e454c7685693fe4ace53fc0b84f3cf752699a556876f2a7f33b7a12972ae33d1c407fb920d6d4aed10de52fdf0dd01902ccdf45cd5da8d55e7f88
+  languageName: node
+  linkType: hard
+
 "evp_bytestokey@npm:^1.0.0, evp_bytestokey@npm:^1.0.3":
   version: 1.0.3
   resolution: "evp_bytestokey@npm:1.0.3"