From 907399b878b263f938ad22e949a7c5dcec624a4a Mon Sep 17 00:00:00 2001 From: JKorf Date: Wed, 5 Aug 2026 21:31:59 +0200 Subject: [PATCH] Added ManualUpdateSubscription and UpdateSubscription additional constructor to allow producing websocket events without actual connection --- .../ManualUpdateSubscriptionTests.cs | 136 ++++++++++++ .../Sockets/ManualUpdateSubscription.cs | 209 ++++++++++++++++++ .../Objects/Sockets/UpdateSubscription.cs | 74 +++++-- 3 files changed, 398 insertions(+), 21 deletions(-) create mode 100644 CryptoExchange.Net.UnitTests/ManualUpdateSubscriptionTests.cs create mode 100644 CryptoExchange.Net/Objects/Sockets/ManualUpdateSubscription.cs diff --git a/CryptoExchange.Net.UnitTests/ManualUpdateSubscriptionTests.cs b/CryptoExchange.Net.UnitTests/ManualUpdateSubscriptionTests.cs new file mode 100644 index 00000000..fea2947a --- /dev/null +++ b/CryptoExchange.Net.UnitTests/ManualUpdateSubscriptionTests.cs @@ -0,0 +1,136 @@ +using CryptoExchange.Net.Objects; +using CryptoExchange.Net.Objects.Errors; +using CryptoExchange.Net.Objects.Sockets; +using CryptoExchange.Net.Sockets.Default; +using NUnit.Framework; +using System; +using System.Collections.Generic; +using System.Threading.Tasks; + +namespace CryptoExchange.Net.UnitTests +{ + [TestFixture] + public class ManualUpdateSubscriptionTests + { + [Test] + public void Constructor_Should_CreateSubscribedVirtualSubscription() + { + var controller = new ManualUpdateSubscription(socketId: 12); + + Assert.That(controller.Subscription.SocketId, Is.EqualTo(12)); + Assert.That(controller.Subscription.Id, Is.GreaterThan(0)); + Assert.That(controller.Subscription.SocketStatus, Is.EqualTo(SocketStatus.Connected)); + Assert.That(controller.Subscription.SubscriptionStatus, Is.EqualTo(SubscriptionStatus.Subscribed)); + Assert.That(controller.Subscription.LastReceiveTime, Is.Null); + } + + [Test] + public void StateChanges_Should_BeVisibleOnSubscription() + { + var controller = new ManualUpdateSubscription(); + var timestamp = new DateTime(2026, 8, 5, 12, 0, 0, DateTimeKind.Utc); + var statuses = new List(); + controller.Subscription.SubscriptionStatusChanged += statuses.Add; + + controller.SetLastReceiveTime(timestamp); + controller.SetSocketStatus(SocketStatus.Reconnecting); + controller.SetSubscriptionStatus(SubscriptionStatus.Subscribing); + controller.SetSubscriptionStatus(SubscriptionStatus.Subscribed); + + Assert.That(controller.Subscription.LastReceiveTime, Is.EqualTo(timestamp)); + Assert.That(controller.Subscription.SocketStatus, Is.EqualTo(SocketStatus.Reconnecting)); + Assert.That(controller.Subscription.SubscriptionStatus, Is.EqualTo(SubscriptionStatus.Subscribed)); + Assert.That(statuses, Is.EqualTo(new[] + { + SubscriptionStatus.Subscribing, + SubscriptionStatus.Subscribed + })); + } + + [Test] + public void LifecycleMethods_Should_InvokeSubscriptionEvents() + { + var controller = new ManualUpdateSubscription(); + var error = new ServerError("Test error", ErrorInfo.Unknown); + var exception = new InvalidOperationException("Test exception"); + var disconnectedPeriod = TimeSpan.FromMinutes(2); + var lost = 0; + var restored = TimeSpan.Zero; + Error? resubscribeError = null; + var paused = 0; + var unpaused = 0; + Exception? receivedException = null; + + controller.Subscription.ConnectionLost += () => lost++; + controller.Subscription.ConnectionRestored += x => restored = x; + controller.Subscription.ResubscribingFailed += x => resubscribeError = x; + controller.Subscription.ActivityPaused += () => paused++; + controller.Subscription.ActivityUnpaused += () => unpaused++; + controller.Subscription.Exception += x => receivedException = x; + + controller.InvokeConnectionLost(); + controller.InvokeConnectionRestored(disconnectedPeriod); + controller.InvokeResubscribingFailed(error); + controller.InvokeActivityPaused(); + controller.InvokeActivityUnpaused(); + controller.InvokeException(exception); + + Assert.That(lost, Is.EqualTo(1)); + Assert.That(restored, Is.EqualTo(disconnectedPeriod)); + Assert.That(resubscribeError, Is.SameAs(error)); + Assert.That(paused, Is.EqualTo(1)); + Assert.That(unpaused, Is.EqualTo(1)); + Assert.That(receivedException, Is.SameAs(exception)); + } + + [Test] + public void InvokeConnectionClosed_Should_CloseAndOnlyInvokeOnce() + { + var controller = new ManualUpdateSubscription(); + var closed = 0; + controller.Subscription.ConnectionClosed += () => closed++; + + controller.InvokeConnectionClosed(); + controller.InvokeConnectionClosed(); + + Assert.That(closed, Is.EqualTo(1)); + Assert.That(controller.Subscription.SocketStatus, Is.EqualTo(SocketStatus.Closed)); + Assert.That(controller.Subscription.SubscriptionStatus, Is.EqualTo(SubscriptionStatus.Closed)); + } + + [Test] + public async Task SubscriptionOperations_Should_InvokeCallbacks() + { + var closes = 0; + var reconnects = 0; + var resubscribes = 0; + var controller = new ManualUpdateSubscription( + closeAsync: () => + { + closes++; + return Task.CompletedTask; + }, + reconnectAsync: () => + { + reconnects++; + return Task.CompletedTask; + }, + resubscribeAsync: () => + { + resubscribes++; + return Task.FromResult(CallResult.Ok()); + }); + + await controller.Subscription.ReconnectAsync(); + var resubscribeResult = await controller.Subscription.ResubscribeAsync(); + await controller.Subscription.CloseAsync(); + await controller.Subscription.CloseAsync(); + + Assert.That(reconnects, Is.EqualTo(1)); + Assert.That(resubscribes, Is.EqualTo(1)); + Assert.That(resubscribeResult.Success, Is.True); + Assert.That(closes, Is.EqualTo(1)); + Assert.That(controller.Subscription.SubscriptionStatus, Is.EqualTo(SubscriptionStatus.Closed)); + } + } +} diff --git a/CryptoExchange.Net/Objects/Sockets/ManualUpdateSubscription.cs b/CryptoExchange.Net/Objects/Sockets/ManualUpdateSubscription.cs new file mode 100644 index 00000000..58252b9a --- /dev/null +++ b/CryptoExchange.Net/Objects/Sockets/ManualUpdateSubscription.cs @@ -0,0 +1,209 @@ +using CryptoExchange.Net.Sockets; +using CryptoExchange.Net.Sockets.Default; +using CryptoExchange.Net.Sockets.Default.Routing; +using Microsoft.Extensions.Logging.Abstractions; +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace CryptoExchange.Net.Objects.Sockets +{ + /// + /// Controller for an update subscription which isn't backed by a websocket connection. Can be used for testing. + /// + public class ManualUpdateSubscription + { + private readonly Func _closeAsync; + private readonly Func _reconnectAsync; + private readonly Func> _resubscribeAsync; + private readonly ManualSubscription _manualSubscription; + private int _closedEventInvoked; + + /// + /// The update subscription + /// + public UpdateSubscription Subscription { get; } + + /// + /// The virtual socket id + /// + public int SocketId { get; } + + /// + /// The last timestamp anything was received by the subscription + /// + public DateTime? LastReceiveTime { get; private set; } + + /// + /// The current virtual websocket status + /// + public SocketStatus SocketStatus { get; private set; } + + /// + /// Create a manually controlled update subscription + /// + /// The virtual socket id + /// Callback when the subscription is closed + /// Callback when a reconnect is requested + /// Callback when a resubscribe is requested + public ManualUpdateSubscription( + int socketId = 0, + Func? closeAsync = null, + Func? reconnectAsync = null, + Func>? resubscribeAsync = null) + { + SocketId = socketId; + SocketStatus = SocketStatus.Connected; + _closeAsync = closeAsync ?? (() => Task.CompletedTask); + _reconnectAsync = reconnectAsync ?? (() => Task.CompletedTask); + _resubscribeAsync = resubscribeAsync ?? (() => Task.FromResult(CallResult.Ok())); + + _manualSubscription = new ManualSubscription(); + _manualSubscription.Status = SubscriptionStatus.Subscribed; + Subscription = new UpdateSubscription(this, _manualSubscription); + } + + /// + /// Set the last timestamp anything was received by the subscription + /// + /// The receive timestamp + public void SetLastReceiveTime(DateTime? timestamp) + { + LastReceiveTime = timestamp; + } + + /// + /// Set the virtual websocket status + /// + /// The status + public void SetSocketStatus(SocketStatus status) + { + SocketStatus = status; + } + + /// + /// Set the subscription status + /// + /// The status + public void SetSubscriptionStatus(SubscriptionStatus status) + { + _manualSubscription.Status = status; + } + + /// + /// Invoke the connection lost event + /// + public void InvokeConnectionLost() + { + Subscription.HandleConnectionLostEvent(); + } + + /// + /// Invoke the connection restored event + /// + /// The period the connection was disconnected + public void InvokeConnectionRestored(TimeSpan disconnectedPeriod) + { + Subscription.HandleConnectionRestoredEvent(disconnectedPeriod); + } + + /// + /// Invoke the connection closed event + /// + public void InvokeConnectionClosed() + { + if (Interlocked.Exchange(ref _closedEventInvoked, 1) != 0) + return; + + SocketStatus = SocketStatus.Closed; + _manualSubscription.Status = SubscriptionStatus.Closed; + Subscription.HandleConnectionClosedEvent(); + } + + /// + /// Invoke the resubscribing failed event + /// + /// The resubscribe error + public void InvokeResubscribingFailed(Error error) + { + if (error == null) + throw new ArgumentNullException(nameof(error)); + + Subscription.HandleResubscribeFailedEvent(error); + } + + /// + /// Invoke the activity paused event + /// + public void InvokeActivityPaused() + { + Subscription.HandlePausedEvent(); + } + + /// + /// Invoke the activity unpaused event + /// + public void InvokeActivityUnpaused() + { + Subscription.HandleUnpausedEvent(); + } + + /// + /// Invoke the exception event + /// + /// The exception + public void InvokeException(Exception exception) + { + if (exception == null) + throw new ArgumentNullException(nameof(exception)); + + _manualSubscription.InvokeExceptionHandler(exception); + } + + internal async Task CloseAsync() + { + if (_manualSubscription.Status == SubscriptionStatus.Closed + || _manualSubscription.Status == SubscriptionStatus.Closing) + return; + + _manualSubscription.Status = SubscriptionStatus.Closing; + try + { + await _closeAsync().ConfigureAwait(false); + } + finally + { + _manualSubscription.Status = SubscriptionStatus.Closed; + } + } + + internal Task ReconnectAsync() + { + return _reconnectAsync(); + } + + internal Task ResubscribeAsync() + { + return _resubscribeAsync(); + } + + private class ManualSubscription : Subscription + { + public ManualSubscription() + : base(NullLogger.Instance, false) + { + MessageRouter = MessageRouter.Create(); + } + + protected override Query? GetSubQuery(SocketConnection connection) + { + return null; + } + + protected override Query? GetUnsubQuery(SocketConnection connection) + { + return null; + } + } + } +} diff --git a/CryptoExchange.Net/Objects/Sockets/UpdateSubscription.cs b/CryptoExchange.Net/Objects/Sockets/UpdateSubscription.cs index 4f07ebb3..6e1bbcc8 100644 --- a/CryptoExchange.Net/Objects/Sockets/UpdateSubscription.cs +++ b/CryptoExchange.Net/Objects/Sockets/UpdateSubscription.cs @@ -12,7 +12,8 @@ namespace CryptoExchange.Net.Objects.Sockets /// public class UpdateSubscription { - private readonly SocketConnection _connection; + private readonly SocketConnection? _connection; + private readonly ManualUpdateSubscription? _manualSubscription; internal readonly Subscription _subscription; #if NET9_0_OR_GREATER @@ -102,7 +103,7 @@ namespace CryptoExchange.Net.Objects.Sockets /// /// The id of the socket /// - public int SocketId => _connection.SocketId; + public int SocketId => _connection?.SocketId ?? _manualSubscription!.SocketId; /// /// The id of the subscription @@ -112,12 +113,12 @@ namespace CryptoExchange.Net.Objects.Sockets /// /// The last timestamp anything was received from the server /// - public DateTime? LastReceiveTime => _connection.LastReceiveTime; + public DateTime? LastReceiveTime => _connection?.LastReceiveTime ?? _manualSubscription!.LastReceiveTime; /// /// The current websocket status /// - public SocketStatus SocketStatus => _connection.Status; + public SocketStatus SocketStatus => _connection?.Status ?? _manualSubscription!.SocketStatus; /// /// The current subscription status @@ -143,6 +144,18 @@ namespace CryptoExchange.Net.Objects.Sockets _subscription.StatusChanged += (x) => SubscriptionStatusChanged?.Invoke(x); } + /// + /// ctor + /// + /// The manual subscription for controlling events and data + /// The subscription + internal UpdateSubscription(ManualUpdateSubscription manualSubscription, Subscription subscription) + { + _manualSubscription = manualSubscription; + _subscription = subscription; + _subscription.StatusChanged += (x) => SubscriptionStatusChanged?.Invoke(x); + } + private void UnsubscribeConnectionEvents() { lock (_eventLock) @@ -150,22 +163,26 @@ namespace CryptoExchange.Net.Objects.Sockets if (!_connectionEventsSubscribed) return; - _connection.ConnectionClosed -= HandleConnectionClosedEvent; - _connection.ConnectionLost -= HandleConnectionLostEvent; - _connection.ConnectionRestored -= HandleConnectionRestoredEvent; - _connection.ResubscribingFailed -= HandleResubscribeFailedEvent; - _connection.ActivityPaused -= HandlePausedEvent; - _connection.ActivityUnpaused -= HandleUnpausedEvent; + if (_connection != null) + { + _connection.ConnectionClosed -= HandleConnectionClosedEvent; + _connection.ConnectionLost -= HandleConnectionLostEvent; + _connection.ConnectionRestored -= HandleConnectionRestoredEvent; + _connection.ResubscribingFailed -= HandleResubscribeFailedEvent; + _connection.ActivityPaused -= HandlePausedEvent; + _connection.ActivityUnpaused -= HandleUnpausedEvent; + } + _connectionEventsSubscribed = false; } } - private void HandleConnectionClosedEvent() + internal void HandleConnectionClosedEvent() { UnsubscribeConnectionEvents(); // If we're not the subscription closing this connection don't bother emitting - if (!_subscription.IsClosingConnection) + if (_connection != null && !_subscription.IsClosingConnection) return; List handlers; @@ -176,7 +193,7 @@ namespace CryptoExchange.Net.Objects.Sockets callback(); } - private void HandleConnectionLostEvent() + internal void HandleConnectionLostEvent() { if (!_subscription.Active) { @@ -192,7 +209,7 @@ namespace CryptoExchange.Net.Objects.Sockets callback(); } - private void HandleConnectionRestoredEvent(TimeSpan period) + internal void HandleConnectionRestoredEvent(TimeSpan period) { if (!_subscription.Active) { @@ -208,7 +225,7 @@ namespace CryptoExchange.Net.Objects.Sockets callback(period); } - private void HandleResubscribeFailedEvent(Error error) + internal void HandleResubscribeFailedEvent(Error error) { if (!_subscription.Active) { @@ -224,7 +241,7 @@ namespace CryptoExchange.Net.Objects.Sockets callback(error); } - private void HandlePausedEvent() + internal void HandlePausedEvent() { if (!_subscription.Active) { @@ -240,7 +257,7 @@ namespace CryptoExchange.Net.Objects.Sockets callback(); } - private void HandleUnpausedEvent() + internal void HandleUnpausedEvent() { if (!_subscription.Active) { @@ -262,7 +279,10 @@ namespace CryptoExchange.Net.Objects.Sockets /// public Task CloseAsync() { - return _connection.CloseAsync(_subscription); + if (_connection != null) + return _connection.CloseAsync(_subscription); + + return _manualSubscription!.CloseAsync(); } /// @@ -271,7 +291,10 @@ namespace CryptoExchange.Net.Objects.Sockets /// public Task ReconnectAsync() { - return _connection.TriggerReconnectAsync(); + if (_connection != null) + return _connection.TriggerReconnectAsync(); + + return _manualSubscription!.ReconnectAsync(); } /// @@ -280,7 +303,13 @@ namespace CryptoExchange.Net.Objects.Sockets /// internal async Task UnsubscribeAsync() { - await _connection.UnsubscribeAsync(_subscription).ConfigureAwait(false); + if (_connection != null) + { + await _connection.UnsubscribeAsync(_subscription).ConfigureAwait(false); + return; + } + + await _manualSubscription!.CloseAsync().ConfigureAwait(false); } /// @@ -289,7 +318,10 @@ namespace CryptoExchange.Net.Objects.Sockets /// internal async Task ResubscribeAsync() { - return await _connection.ResubscribeAsync(_subscription).ConfigureAwait(false); + if (_connection != null) + return await _connection.ResubscribeAsync(_subscription).ConfigureAwait(false); + + return await _manualSubscription!.ResubscribeAsync().ConfigureAwait(false); } } }