Merge pull request #416 from EpicGamesExt/backport/UE5.5/pr-409

[UE5.5] Merge pull request #409 from mcottontensor/heartbeat

Author: mcottontensor

Common/src/Protocol/KeepaliveMonitor.ts

@@ -0,0 +1,77 @@
+import * as Messages from '../Messages/signalling_messages';
+import * as MessageHelpers from '../Messages/message_helpers';
+import { SignallingProtocol } from './SignallingProtocol';
+
+/**
+ * Used to regularly ping a protocol connection to make sure the connection is still good and open.
+ * When the pong doesn't come in response to a ping in time a callback is fired that can be handed
+ * by the owner.
+ */
+export class KeepaliveMonitor {
+    private protocol: SignallingProtocol;
+    private timeout: number;
+    private keepalive?: ReturnType<typeof setInterval>;
+    private alive: boolean = false;
+    private rtt: number = 0;
+
+    // naming a bound function so we can add and then remove it with on and off.
+    private onResponse: (pongMsg: Messages.pong) => void;
+
+    /**
+     * Called when a pong does not come back from a ping.
+     */
+    onTimeout?: () => void;
+
+    /**
+     * Gets the Round Trip Time of the current connection in milliseconds.
+     */
+    get RTT(): number {
+        return this.rtt;
+    }
+
+    /**
+     * Creates a new monitor and starts the ping timer. If a pong does not come back by the time we want
+     * to send a second ping then the connection is considered dead and the onTimeout callback is fired.
+     * @param protocol The connection that we want to monitor.
+     * @param timeout The time in milliseconds between ping messages.
+     */
+    constructor(protocol: SignallingProtocol, timeout: number) {
+        this.protocol = protocol;
+        this.timeout = timeout;
+        this.onResponse = this.onHeartbeatResponse.bind(this);
+        this.protocol.transport.on('close', this.stop.bind(this));
+        this.start();
+    }
+
+    private start(): void {
+        this.alive = true;
+        this.protocol.on('pong', this.onResponse);
+        this.keepalive = setInterval(this.sendHeartbeat.bind(this), this.timeout);
+    }
+
+    private stop(): void {
+        clearInterval(this.keepalive);
+        this.protocol.off('pong', this.onResponse);
+    }
+
+    private sendHeartbeat(): void {
+        // if we never got a response from the last heartbeat, assume the connection is dead and timeout
+        if (this.alive === false) {
+            this.onTimeout?.();
+            return;
+        }
+
+        // mark the connection as temporarily dead until we get a response from the ping
+        this.alive = false;
+        this.protocol.sendMessage(
+            MessageHelpers.createMessage(Messages.ping, { time: new Date().getTime() })
+        );
+    }
+
+    private onHeartbeatResponse(pongMsg: Messages.pong): void {
+        // we got a pong response from the other side, the connection is good.
+        // we also store the round trip time if anyone is curious
+        this.rtt = new Date().getTime() - pongMsg.time;
+        this.alive = true;
+    }
+}

Common/src/Protocol/SignallingProtocol.ts

@@ -3,8 +3,6 @@
 import { ITransport } from '../Transport/ITransport';
 import { EventEmitter } from '../Event/EventEmitter';
 import { BaseMessage } from '../Messages/base_message';
-import * as Messages from '../Messages/signalling_messages';
-import * as MessageHelpers from '../Messages/message_helpers';
 
 /**
  * Signalling protocol for handling messages from the signalling server.
@@ -38,14 +36,6 @@ export class SignallingProtocol extends EventEmitter {
         this.transport = transport;
 
         transport.onMessage = (msg: BaseMessage) => {
-            // auto handle ping messages
-            if (msg.type == Messages.ping.typeName) {
-                const pongMessage = MessageHelpers.createMessage(Messages.pong, {
-                    time: new Date().getTime()
-                });
-                transport.sendMessage(pongMessage);
-            }
-
             // call the handlers
             transport.emit('message', msg); // emit this for listeners listening to any message
             if (!this.emit(msg.type, msg)) {

Common/src/Transport/WebSocketTransport.ts

@@ -114,11 +114,12 @@ export class WebSocketTransport extends EventEmitter implements ITransport {
             return;
         }
 
-        Logger.Info('received => \n' + JSON.stringify(JSON.parse(event.data as string), undefined, 4));
-
         let parsedMessage: BaseMessage;
         try {
-            parsedMessage = JSON.parse(event.data as string) as BaseMessage;
+            // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
+            const parsedData = JSON.parse(event.data as string);
+            Logger.Info('received => \n' + JSON.stringify(parsedData, undefined, 4));
+            parsedMessage = parsedData as BaseMessage;
         } catch (e: unknown) {
             if (e instanceof Error) {
                 Logger.Error(`Error parsing message string ${event.data}.\n${e.message}`);

Common/src/pixelstreamingcommon.ts

@@ -9,3 +9,4 @@ export { EventEmitter } from './Event/EventEmitter';
 export { MessageRegistry } from './Messages/message_registry';
 export * as Messages from './Messages/signalling_messages';
 export * as MessageHelpers from './Messages/message_helpers';
+export { KeepaliveMonitor } from './Protocol/KeepaliveMonitor';

Extras/JSStreamer/src/streamer.ts

@@ -121,6 +121,8 @@ export class Streamer extends EventEmitter {
         this.protocol.addListener(Messages.iceCandidate.typeName, (msg: BaseMessage) =>
             this.handleIceMessage(msg as Messages.iceCandidate)
         );
+
+        this.transport.on('timeout', () => console.log('Streamer connection timeout'));
     }
 
     startStreaming(signallingURL: string, stream: MediaStream) {

Frontend/library/src/Config/Config.ts

@@ -59,6 +59,7 @@ export class NumericParameters {
     static WebRTCMaxBitrate = 'WebRTCMaxBitrate' as const;
     static MaxReconnectAttempts = 'MaxReconnectAttempts' as const;
     static StreamerAutoJoinInterval = 'StreamerAutoJoinInterval' as const;
+    static KeepaliveDelay = 'KeepaliveDelay' as const;
 }
 
 export type NumericParametersKeys = Exclude<keyof typeof NumericParameters, 'prototype'>;
@@ -761,6 +762,21 @@ export class Config {
                 useUrlParams
             )
         );
+
+        this.numericParameters.set(
+            NumericParameters.KeepaliveDelay,
+            new SettingNumber(
+                NumericParameters.KeepaliveDelay,
+                'Connection Keepalive delay',
+                'Delay between keepalive pings to the signalling server.',
+                0 /*min*/,
+                900000 /*max*/,
+                settings && Object.prototype.hasOwnProperty.call(settings, NumericParameters.KeepaliveDelay)
+                    ? settings[NumericParameters.KeepaliveDelay]
+                    : 30000 /*value*/,
+                useUrlParams
+            )
+        );
     }
 
     /**

Frontend/library/src/WebRtcPlayer/WebRtcPlayerController.ts

@@ -7,7 +7,8 @@ import {
     ITransport,
     Messages,
     MessageHelpers,
-    BaseMessage
+    BaseMessage,
+    KeepaliveMonitor
 } from '@epicgames-ps/lib-pixelstreamingcommon-ue5.5';
 import { StreamController } from '../VideoPlayer/StreamController';
 import { FreezeFrameController } from '../FreezeFrame/FreezeFrameController';
@@ -104,6 +105,7 @@ export class WebRtcPlayerController {
     subscribedStream: string;
     signallingUrlBuilder: () => string;
     autoJoinTimer: ReturnType<typeof setTimeout> = undefined;
+    keepalive: KeepaliveMonitor;
 
     /**
      *
@@ -171,6 +173,9 @@ export class WebRtcPlayerController {
         this.protocol.addListener(Messages.config.typeName, (msg: BaseMessage) =>
             this.handleOnConfigMessage(msg as Messages.config)
         );
+        this.protocol.addListener(Messages.ping.typeName, (msg: BaseMessage) =>
+            this.handlePingMessage(msg as Messages.ping)
+        );
         this.protocol.addListener(Messages.streamerList.typeName, (msg: BaseMessage) =>
             this.handleStreamerListMessage(msg as Messages.streamerList)
         );
@@ -846,7 +851,7 @@ export class WebRtcPlayerController {
         // if the connection is open, first close it and force a reconnect.
         if (this.protocol.isConnected()) {
             if (!this.forceReconnect) {
-                message = `${message} Reconnecting.`;
+                this.disconnectMessage = `${message} Reconnecting.`;
             }
             this.closeSignalingServer(message, true);
         } else {
@@ -1023,6 +1028,15 @@ export class WebRtcPlayerController {
         this.disconnectMessage = null;
         const signallingUrl = this.signallingUrlBuilder();
         this.protocol.connect(signallingUrl);
+        const keepaliveDelay = this.config.getNumericSettingValue(NumericParameters.KeepaliveDelay);
+        if (keepaliveDelay > 0) {
+            this.keepalive = new KeepaliveMonitor(this.protocol, keepaliveDelay);
+            this.keepalive.onTimeout = () => {
+                // if the ping fails just disconnect
+                Logger.Error(`Protocol timeout`);
+                this.protocol.disconnect();
+            };
+        }
     }
 
     /**
@@ -1138,6 +1152,10 @@ export class WebRtcPlayerController {
         this.startSession(messageConfig.peerConnectionOptions);
     }
 
+    handlePingMessage(pingMessage: Messages.ping) {
+        this.protocol.sendMessage(MessageHelpers.createMessage(Messages.pong, { time: pingMessage.time }));
+    }
+
     /**
      * Handles when the signalling server gives us the list of streamer ids.
      */

Frontend/ui-library/src/Config/ConfigUI.ts

@@ -188,6 +188,11 @@ export class ConfigUI {
                     psSettingsSection,
                     this.numericParametersUi.get(NumericParameters.StreamerAutoJoinInterval)
                 );
+            if (isSettingEnabled(settingsConfig, NumericParameters.KeepaliveDelay))
+                this.addSettingNumeric(
+                    psSettingsSection,
+                    this.numericParametersUi.get(NumericParameters.KeepaliveDelay)
+                );
         }
 
         if (isSectionEnabled(settingsConfig, SettingsSections.UI)) {

Signalling/src/PlayerConnection.ts

@@ -108,6 +108,7 @@ export class PlayerConnection implements IPlayer, LogUtils.IMessageLogger {
             Messages.listStreamers.typeName,
             LogUtils.createHandlerListener(this, this.onListStreamers)
         );
+        this.protocol.on(Messages.ping.typeName, LogUtils.createHandlerListener(this, this.onPingMessage));
         /* eslint-enable @typescript-eslint/unbound-method */
 
         this.protocol.on(Messages.offer.typeName, this.sendToStreamer.bind(this));
@@ -118,7 +119,7 @@ export class PlayerConnection implements IPlayer, LogUtils.IMessageLogger {
         this.protocol.on(Messages.layerPreference.typeName, this.sendToStreamer.bind(this));
 
         this.protocol.on('unhandled', (message: BaseMessage) => {
-            Logger.warn(`Unhandled protocol message: ${JSON.stringify(message)}`);
+            Logger.warn(`Unhandled player protocol message: ${JSON.stringify(message)}`);
         });
     }
 
@@ -226,7 +227,7 @@ export class PlayerConnection implements IPlayer, LogUtils.IMessageLogger {
         this.sendMessage(renameMessage);
     }
 
-    private onStreamerRemoved() {
-        this.disconnect();
+    private onPingMessage(message: Messages.ping): void {
+        this.sendMessage(MessageHelpers.createMessage(Messages.pong, { time: message.time }));
     }
 }

Signalling/src/SFUConnection.ts

@@ -155,6 +155,7 @@ export class SFUConnection extends EventEmitter implements IPlayer, IStreamer, L
             Messages.stopStreaming.typeName,
             LogUtils.createHandlerListener(this, this.onStopStreaming)
         );
+        this.protocol.on(Messages.ping.typeName, LogUtils.createHandlerListener(this, this.onPingMessage));
         /* eslint-enable @typescript-eslint/unbound-method */
 
         this.protocol.on(Messages.offer.typeName, this.sendToPlayer.bind(this));
@@ -297,4 +298,8 @@ export class SFUConnection extends EventEmitter implements IPlayer, IStreamer, L
         this.streaming = false;
         this.emit('disconnect');
     }
+
+    private onPingMessage(message: Messages.ping): void {
+        this.sendMessage(MessageHelpers.createMessage(Messages.pong, { time: message.time }));
+    }
 }