[UE5.8] fix(signalling): leave the previous streamer when an SFU subscribes again (#1156)

.changeset/sfu-resubscribe.md

@@ -0,0 +1,5 @@
+---
+'@epicgames-ps/lib-pixelstreamingsignalling-ue5.8': patch
+---
+
+Leave the previous streamer when an SFU subscribes to another. `SFUConnection.subscribe` overwrote `subscribedStreamer` without unsubscribing from the old streamer, so the old streamer kept forwarding to the SFU and kept the SFU's listeners. When that old streamer later disconnected, it detached the SFU from the streamer it was actually on and dropped the SFU's players. `PlayerConnection` already unsubscribes first; the SFU now does the same.

Signalling/src/SFUConnection.test.ts

@@ -0,0 +1,45 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+import type WebSocket from 'ws';
+import { Messages } from '@epicgames-ps/lib-pixelstreamingcommon-ue5.8';
+import { SFUConnection } from './SFUConnection';
+import { StreamerConnection } from './StreamerConnection';
+import { StreamerRegistry } from './StreamerRegistry';
+import { PlayerRegistry } from './PlayerRegistry';
+import { SignallingServer } from './SignallingServer';
+
+// A streamer the SFU can subscribe to, with no socket behind it.
+function unconnectedStreamer(streamerId: string): StreamerConnection {
+    const streamer = new StreamerConnection({} as unknown as SignallingServer, {} as unknown as WebSocket);
+    streamer.streamerId = streamerId;
+    return streamer;
+}
+
+describe('SFUConnection subscribe', () => {
+    it('leaves the previous streamer when it subscribes to another', () => {
+        const a = unconnectedStreamer('A');
+        const b = unconnectedStreamer('B');
+        const streamerRegistry = new StreamerRegistry();
+        streamerRegistry.streamers.push(a, b);
+        const server = {
+            streamerRegistry,
+            playerRegistry: new PlayerRegistry()
+        } as unknown as SignallingServer;
+        const sfu = new SFUConnection(server, {} as unknown as WebSocket);
+        sfu.playerId = 'SFU1';
+
+        sfu.protocol.emit(Messages.subscribe.typeName, {
+            type: Messages.subscribe.typeName,
+            streamerId: 'A'
+        });
+        sfu.protocol.emit(Messages.subscribe.typeName, {
+            type: Messages.subscribe.typeName,
+            streamerId: 'B'
+        });
+
+        expect(a.subscribers.has('SFU1')).toBe(false);
+        expect(b.subscribers.has('SFU1')).toBe(true);
+        // A disconnect from the old streamer must not detach the SFU from the one it is on now.
+        a.emit('disconnect');
+        expect(sfu.subscribedStreamer).toBe(b);
+    });
+});

Signalling/src/SFUConnection.ts

@@ -184,6 +184,11 @@ export class SFUConnection extends EventEmitter implements IPlayer, IStreamer, L
             return;
         }
 
+        // Leave the previous streamer first, as PlayerConnection does, so it stops forwarding to this SFU
+        // and a later disconnect of it cannot detach the SFU from the streamer it is now on.
+        if (this.subscribedStreamer) {
+            this.unsubscribe();
+        }
         this.subscribedStreamer = streamer;
         // Register as a subscriber of the upstream streamer so it will forward signalling messages
         // back to this SFU (the streamer's forwardMessage only sends to subscribed players).