From 431467db9731fbc937650886dd2cf6d1e79810fd Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 19 Oct 2026 14:58:55 -0400 Subject: [PATCH 1/2] feat(dotnet): let DucPushClient tell Gateway the recording ended Adds `DucPushClient.EndRecordingAsync`, which closes the push stream with close code 4011. Gateway then ends the recording right away instead of waiting out the reconnection window. `CloseAsync` and `DisposeAsync` still send 1000, since RDM drops and reopens the stream on resize and Gateway has to keep the recording open for that. Gateway side: Devolutions/devolutions-gateway#2116 (DGW-477). --- .../Devolutions.Cadeau/Push/DucPushClient.cs | 32 +++++++++++++++++-- 1 file changed, 30 insertions(+), 2 deletions(-) diff --git a/dotnet/Devolutions.Cadeau/Push/DucPushClient.cs b/dotnet/Devolutions.Cadeau/Push/DucPushClient.cs index a81306f..749323b 100644 --- a/dotnet/Devolutions.Cadeau/Push/DucPushClient.cs +++ b/dotnet/Devolutions.Cadeau/Push/DucPushClient.cs @@ -26,6 +26,12 @@ public sealed class DucPushClient : IAsyncDisposable // STORAGE_FULL_CLOSE_CODE constant. private const ushort StorageFullCloseCode = 4010; + // Private WebSocket close code (4000-4999 range) sent by this client to tell the gateway + // the recording is over. The gateway then ends the recording at once instead of keeping + // it open for a reconnect (any other close, such as 1000, keeps it open for the token's + // reconnection window). Must match the gateway's RECORDING_ENDED_CLOSE_CODE constant. + private const ushort RecordingEndedCloseCode = 4011; + private ClientWebSocket? ws; private Task? receiveLoop; @@ -127,7 +133,29 @@ await socket.SendAsync(payload, WebSocketMessageType.Binary, endOfMessage: true, /// was cancelled before the close handshake completed. /// The receive loop is still stopped before this propagates. /// - public async Task CloseAsync(CancellationToken cancellationToken = default) + /// + /// The gateway treats a normal closure as a possible reconnect (for example after a resize) + /// and keeps the recording open. Use when the recording is over. + /// + public Task CloseAsync(CancellationToken cancellationToken = default) + { + return this.CloseAsync(WebSocketCloseStatus.NormalClosure, cancellationToken); + } + + /// + /// Sends a close frame with close code 4011 so the gateway ends the recording at once, + /// then waits for the receive loop to terminate. + /// + /// + /// was cancelled before the close handshake completed. + /// The receive loop is still stopped before this propagates. + /// + public Task EndRecordingAsync(CancellationToken cancellationToken = default) + { + return this.CloseAsync((WebSocketCloseStatus)RecordingEndedCloseCode, cancellationToken); + } + + private async Task CloseAsync(WebSocketCloseStatus closeStatus, CancellationToken cancellationToken) { ClientWebSocket? socket = this.ws; if (socket == null) @@ -141,7 +169,7 @@ public async Task CloseAsync(CancellationToken cancellationToken = default) { try { - await socket.CloseAsync(WebSocketCloseStatus.NormalClosure, statusDescription: string.Empty, cancellationToken) + await socket.CloseAsync(closeStatus, statusDescription: string.Empty, cancellationToken) .ConfigureAwait(false); } catch (OperationCanceledException) From bdcc282211e86b0ffaf73401c067fa7825a75405 Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 19 Oct 2026 16:17:55 -0400 Subject: [PATCH 2/2] feat(streamer): let DucStreamer tell Gateway the recording ended Mac and Linux RDM push recordings through DucStreamer, whose WebSocket (V3) backend always closed with 1000. Gateway keeps such a recording open for its reconnection window, so live viewers only saw the end after that delay. DucStreamer.EndRecording closes the WebSocket with code 4011, which tells Gateway the recording is over. Disconnect still closes with 1000 because callers also use it before reconnecting. Older protocols have no close code and just disconnect. The 4011 constant is now shared with DucPushClient. Refs: DGW-477, Devolutions/devolutions-gateway#2116 --- .../DucStreamer/BaseDucStreamerBackend.cs | 3 +++ .../DucStreamer/DucStreamer.cs | 25 +++++++++++++++++++ .../DucStreamer/DucStreamerBackendV3.cs | 10 ++++++++ .../GatewayPushCloseCodes.cs | 11 ++++++++ .../Devolutions.Cadeau/Push/DucPushClient.cs | 8 +----- dotnet/Devolutions.Cadeau/XmfWsStream.cs | 5 ++++ 6 files changed, 55 insertions(+), 7 deletions(-) create mode 100644 dotnet/Devolutions.Cadeau/GatewayPushCloseCodes.cs diff --git a/dotnet/Devolutions.Cadeau/DucStreamer/BaseDucStreamerBackend.cs b/dotnet/Devolutions.Cadeau/DucStreamer/BaseDucStreamerBackend.cs index 4ef2cc4..043015a 100644 --- a/dotnet/Devolutions.Cadeau/DucStreamer/BaseDucStreamerBackend.cs +++ b/dotnet/Devolutions.Cadeau/DucStreamer/BaseDucStreamerBackend.cs @@ -23,6 +23,9 @@ internal abstract class BaseDucStreamerBackend : IDisposable public virtual Task Disconnect(CancellationToken token) => null; + // Only the WebSocket protocol can say the recording is over; older protocols just disconnect. + public virtual Task EndRecording(CancellationToken token) => this.Disconnect(token); + public virtual bool SendKeepAlive() => true; public void Dispose() diff --git a/dotnet/Devolutions.Cadeau/DucStreamer/DucStreamer.cs b/dotnet/Devolutions.Cadeau/DucStreamer/DucStreamer.cs index 30e8076..e88fffb 100644 --- a/dotnet/Devolutions.Cadeau/DucStreamer/DucStreamer.cs +++ b/dotnet/Devolutions.Cadeau/DucStreamer/DucStreamer.cs @@ -143,6 +143,31 @@ public Task Disconnect(CancellationToken token) } } + /// + /// Closes the stream and tells the gateway the recording is over, so it ends the recording at once. + /// + /// + /// leaves the recording open for a reconnect, for example after a resize. + /// Over a WebSocket (V3) this sends close code 4011; older protocols just disconnect. + /// + public Task EndRecording(CancellationToken token) + { + try + { + if (!this.Streamer?.Connected ?? false) + { + return null; + } + + return this.Streamer?.EndRecording(token); + } + finally + { + this.Streamer?.Dispose(); + this.Streamer = null; + } + } + public bool SendKeepAlive() => this.Streamer?.SendKeepAlive() ?? true; public bool SendPayload(DateTime timestamp, IntPtr payload, int length) => this.Streamer?.SendPayload(timestamp, payload, length) ?? false; diff --git a/dotnet/Devolutions.Cadeau/DucStreamer/DucStreamerBackendV3.cs b/dotnet/Devolutions.Cadeau/DucStreamer/DucStreamerBackendV3.cs index 23a82ce..96809c9 100644 --- a/dotnet/Devolutions.Cadeau/DucStreamer/DucStreamerBackendV3.cs +++ b/dotnet/Devolutions.Cadeau/DucStreamer/DucStreamerBackendV3.cs @@ -65,6 +65,16 @@ public override Task Disconnect(CancellationToken token) return base.Disconnect(token); } + public override Task EndRecording(CancellationToken token) + { + if (this.Stream is XmfWsStream wsStream) + { + return wsStream.EndRecording(token); + } + + return base.EndRecording(token); + } + public override bool SendPayload(DateTime timestamp, IntPtr payload, int length) { if (payload == IntPtr.Zero || length <= 0) diff --git a/dotnet/Devolutions.Cadeau/GatewayPushCloseCodes.cs b/dotnet/Devolutions.Cadeau/GatewayPushCloseCodes.cs new file mode 100644 index 0000000..a5b9f94 --- /dev/null +++ b/dotnet/Devolutions.Cadeau/GatewayPushCloseCodes.cs @@ -0,0 +1,11 @@ +namespace Devolutions.Cadeau +{ + internal static class GatewayPushCloseCodes + { + // Private WebSocket close code (4000-4999 range) sent by a producer to tell the gateway + // the recording is over. The gateway then ends the recording at once instead of keeping + // it open for a reconnect (any other close, such as 1000, keeps it open for the token's + // reconnection window). Must match the gateway's RECORDING_ENDED_CLOSE_CODE constant. + internal const ushort RecordingEnded = 4011; + } +} diff --git a/dotnet/Devolutions.Cadeau/Push/DucPushClient.cs b/dotnet/Devolutions.Cadeau/Push/DucPushClient.cs index 749323b..51ea435 100644 --- a/dotnet/Devolutions.Cadeau/Push/DucPushClient.cs +++ b/dotnet/Devolutions.Cadeau/Push/DucPushClient.cs @@ -26,12 +26,6 @@ public sealed class DucPushClient : IAsyncDisposable // STORAGE_FULL_CLOSE_CODE constant. private const ushort StorageFullCloseCode = 4010; - // Private WebSocket close code (4000-4999 range) sent by this client to tell the gateway - // the recording is over. The gateway then ends the recording at once instead of keeping - // it open for a reconnect (any other close, such as 1000, keeps it open for the token's - // reconnection window). Must match the gateway's RECORDING_ENDED_CLOSE_CODE constant. - private const ushort RecordingEndedCloseCode = 4011; - private ClientWebSocket? ws; private Task? receiveLoop; @@ -152,7 +146,7 @@ public Task CloseAsync(CancellationToken cancellationToken = default) /// public Task EndRecordingAsync(CancellationToken cancellationToken = default) { - return this.CloseAsync((WebSocketCloseStatus)RecordingEndedCloseCode, cancellationToken); + return this.CloseAsync((WebSocketCloseStatus)GatewayPushCloseCodes.RecordingEnded, cancellationToken); } private async Task CloseAsync(WebSocketCloseStatus closeStatus, CancellationToken cancellationToken) diff --git a/dotnet/Devolutions.Cadeau/XmfWsStream.cs b/dotnet/Devolutions.Cadeau/XmfWsStream.cs index fbec60f..8d099e0 100644 --- a/dotnet/Devolutions.Cadeau/XmfWsStream.cs +++ b/dotnet/Devolutions.Cadeau/XmfWsStream.cs @@ -69,5 +69,10 @@ public Task Close(CancellationToken cancellationToken) { return this.ws.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken); } + + internal Task EndRecording(CancellationToken cancellationToken) + { + return this.ws.CloseAsync((WebSocketCloseStatus)GatewayPushCloseCodes.RecordingEnded, string.Empty, cancellationToken); + } } }