fix: handle upgrade properly in signal router

Signed-off-by: Heikki Hellgren <heikki.hellgren@op.fi>

plugins/signals-node/api-report.md

@@ -3,13 +3,16 @@
 > Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/).
 
 ```ts
+/// <reference types="node" />
+
+import { Duplex } from 'stream';
 import { EventBroker } from '@backstage/plugin-events-node';
 import { EventParams } from '@backstage/plugin-events-node';
 import { EventSubscriber } from '@backstage/plugin-events-node';
 import { IdentityApi } from '@backstage/plugin-auth-node';
+import { IncomingMessage } from 'http';
 import { JsonObject } from '@backstage/types';
 import { Logger } from 'winston';
-import { Request as Request_2 } from 'express';
 
 // @public (undocumented)
 export type ServiceOptions = {
@@ -29,7 +32,11 @@ export type SignalEventBrokerPayload = {
 export class SignalService implements EventSubscriber {
   // (undocumented)
   static create(options: ServiceOptions): SignalService;
-  handleUpgrade: (req: Request_2) => Promise<void>;
+  handleUpgrade: (
+    req: IncomingMessage,
+    socket: Duplex,
+    head: Buffer,
+  ) => Promise<void>;
   hasSubscribers(topic: string): boolean;
   // (undocumented)
   onEvent(params: EventParams<SignalEventBrokerPayload>): Promise<void>;

plugins/signals-node/src/SignalService.ts

@@ -27,13 +27,13 @@ import {
 import { RawData, WebSocket, WebSocketServer } from 'ws';
 import { IncomingMessage } from 'http';
 import { v4 as uuid } from 'uuid';
-import { Request } from 'express';
 import { JsonObject } from '@backstage/types';
 import {
   BackstageIdentityResponse,
   IdentityApi,
   IdentityApiGetIdentityRequest,
 } from '@backstage/plugin-auth-node';
+import { Duplex } from 'stream';
 
 /** @public */
 export class SignalService implements EventSubscriber {
@@ -61,14 +61,7 @@ export class SignalService implements EventSubscriber {
     this.serverId = uuid();
     this.server = new WebSocketServer({
       noServer: true,
-    });
-
-    this.server.on('close', () => {
-      this.logger.info('Closing signals server');
-      this.connections.forEach(conn => {
-        conn.ws.close();
-      });
-      this.connections = new Map();
+      clientTracking: false,
     });
 
     this.eventBroker?.subscribe(this);
@@ -79,7 +72,11 @@ export class SignalService implements EventSubscriber {
    * list for publish/subscribe functionality
    * @param req - Request
    */
-  handleUpgrade = async (req: Request) => {
+  handleUpgrade = async (
+    req: IncomingMessage,
+    socket: Duplex,
+    head: Buffer,
+  ) => {
     let identity: BackstageIdentityResponse | undefined = undefined;
 
     // Authentication token is passed in Sec-WebSocket-Protocol header as there
@@ -95,8 +92,8 @@ export class SignalService implements EventSubscriber {
 
     this.server.handleUpgrade(
       req,
-      req.socket,
-      Buffer.from(''),
+      socket,
+      head,
       (ws: WebSocket, __: IncomingMessage) => {
         this.addConnection(ws, identity);
       },
@@ -131,17 +128,11 @@ export class SignalService implements EventSubscriber {
       this.connections.delete(id);
     });
 
-    ws.on('ping', () => {
-      this.logger.debug(`Ping from connection ${id}`);
-      ws.pong();
-    });
-
     ws.on('message', (data: RawData, isBinary: boolean) => {
       this.logger.debug(`Received message from connection ${id}: ${data}`);
       if (isBinary) {
         return;
       }
-
       try {
         const json = JSON.parse(data.toString()) as JsonObject;
         this.handleMessage(conn, json);