Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
25 changes: 25 additions & 0 deletions dotnet/Devolutions.Cadeau/DucStreamer/DucStreamer.cs
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,31 @@ public Task Disconnect(CancellationToken token)
}
}

/// <summary>
/// Closes the stream and tells the gateway the recording is over, so it ends the recording at once.
/// </summary>
/// <remarks>
/// <see cref="Disconnect"/> leaves the recording open for a reconnect, for example after a resize.
/// Over a WebSocket (V3) this sends close code <c>4011</c>; older protocols just disconnect.
/// </remarks>
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;
Expand Down
10 changes: 10 additions & 0 deletions dotnet/Devolutions.Cadeau/DucStreamer/DucStreamerBackendV3.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
11 changes: 11 additions & 0 deletions dotnet/Devolutions.Cadeau/GatewayPushCloseCodes.cs
Original file line number Diff line number Diff line change
@@ -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;
}
}
26 changes: 24 additions & 2 deletions dotnet/Devolutions.Cadeau/Push/DucPushClient.cs
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,29 @@ await socket.SendAsync(payload, WebSocketMessageType.Binary, endOfMessage: true,
/// <paramref name="cancellationToken"/> was cancelled before the close handshake completed.
/// The receive loop is still stopped before this propagates.
/// </exception>
public async Task CloseAsync(CancellationToken cancellationToken = default)
/// <remarks>
/// The gateway treats a normal closure as a possible reconnect (for example after a resize)
/// and keeps the recording open. Use <see cref="EndRecordingAsync"/> when the recording is over.
/// </remarks>
public Task CloseAsync(CancellationToken cancellationToken = default)
{
return this.CloseAsync(WebSocketCloseStatus.NormalClosure, cancellationToken);
}

/// <summary>
/// Sends a close frame with close code <c>4011</c> so the gateway ends the recording at once,
/// then waits for the receive loop to terminate.
/// </summary>
/// <exception cref="OperationCanceledException">
/// <paramref name="cancellationToken"/> was cancelled before the close handshake completed.
/// The receive loop is still stopped before this propagates.
/// </exception>
public Task EndRecordingAsync(CancellationToken cancellationToken = default)
{
return this.CloseAsync((WebSocketCloseStatus)GatewayPushCloseCodes.RecordingEnded, cancellationToken);
}

private async Task CloseAsync(WebSocketCloseStatus closeStatus, CancellationToken cancellationToken)
{
ClientWebSocket? socket = this.ws;
if (socket == null)
Expand All @@ -141,7 +163,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)
Expand Down
5 changes: 5 additions & 0 deletions dotnet/Devolutions.Cadeau/XmfWsStream.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}
}