wip

CryptoExchange.Net.UnitTests/RequestCoalescerTests.cs

@@ -0,0 +1,281 @@
+using CryptoExchange.Net.Objects;
+using Microsoft.Extensions.Logging.Abstractions;
+using NUnit.Framework;
+using System;
+using System.Collections;
+using System.Reflection;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace CryptoExchange.Net.UnitTests
+{
+    [TestFixture]
+    public class RequestCoalescerTests
+    {
+        private static readonly TimeSpan _testTimeout = TimeSpan.FromSeconds(5);
+
+        [Test]
+        public async Task ExecuteAsync_WithIdenticalRequests_ShouldExecuteRequestOnce()
+        {
+            var coalescer = CreateCoalescer();
+            var key = CreateKey();
+            var requestCompletion = CreateCompletionSource<string>();
+            var requestStarted = CreateCompletionSource<bool>();
+            var requestCount = 0;
+
+            async Task<string> ExecuteRequest(CancellationToken cancellationToken)
+            {
+                Interlocked.Increment(ref requestCount);
+                requestStarted.TrySetResult(true);
+                return await requestCompletion.Task;
+            }
+
+            var first = coalescer.ExecuteAsync(key, ExecuteRequest, () => "canceled", CancellationToken.None);
+            await requestStarted.Task.WaitAsync(_testTimeout);
+            var second = coalescer.ExecuteAsync(key, ExecuteRequest, () => "canceled", CancellationToken.None);
+
+            Assert.That(requestCount, Is.EqualTo(1));
+
+            requestCompletion.SetResult("result");
+
+            Assert.That(await first.WaitAsync(_testTimeout), Is.EqualTo("result"));
+            Assert.That(await second.WaitAsync(_testTimeout), Is.EqualTo("result"));
+            Assert.That(requestCount, Is.EqualTo(1));
+        }
+
+        [Test]
+        public async Task ExecuteAsync_WithDifferentKeys_ShouldExecuteEachRequest()
+        {
+            var coalescer = CreateCoalescer();
+            var firstCompletion = CreateCompletionSource<string>();
+            var secondCompletion = CreateCompletionSource<string>();
+            var firstStarted = CreateCompletionSource<bool>();
+            var secondStarted = CreateCompletionSource<bool>();
+            var requestCount = 0;
+
+            var first = coalescer.ExecuteAsync(
+                CreateKey(parameters: "symbol=ETH"),
+                async cancellationToken =>
+                {
+                    Interlocked.Increment(ref requestCount);
+                    firstStarted.TrySetResult(true);
+                    return await firstCompletion.Task;
+                },
+                () => "canceled",
+                CancellationToken.None);
+
+            var second = coalescer.ExecuteAsync(
+                CreateKey(parameters: "symbol=BTC"),
+                async cancellationToken =>
+                {
+                    Interlocked.Increment(ref requestCount);
+                    secondStarted.TrySetResult(true);
+                    return await secondCompletion.Task;
+                },
+                () => "canceled",
+                CancellationToken.None);
+
+            await Task.WhenAll(firstStarted.Task, secondStarted.Task).WaitAsync(_testTimeout);
+            Assert.That(requestCount, Is.EqualTo(2));
+
+            firstCompletion.SetResult("first");
+            secondCompletion.SetResult("second");
+
+            Assert.That(await first.WaitAsync(_testTimeout), Is.EqualTo("first"));
+            Assert.That(await second.WaitAsync(_testTimeout), Is.EqualTo("second"));
+        }
+
+        [Test]
+        public async Task ExecuteAsync_WhenOneListenerCancels_ShouldKeepSharedRequestRunning()
+        {
+            var coalescer = CreateCoalescer();
+            var requestCompletion = CreateCompletionSource<string>();
+            var requestStarted = CreateCompletionSource<CancellationToken>();
+            var requestCount = 0;
+            using var firstCancellationSource = new CancellationTokenSource();
+
+            async Task<string> ExecuteRequest(CancellationToken cancellationToken)
+            {
+                Interlocked.Increment(ref requestCount);
+                requestStarted.TrySetResult(cancellationToken);
+                return await requestCompletion.Task;
+            }
+
+            var first = coalescer.ExecuteAsync(CreateKey(), ExecuteRequest, () => "canceled", firstCancellationSource.Token);
+            var requestCancellationToken = await requestStarted.Task.WaitAsync(_testTimeout);
+            var second = coalescer.ExecuteAsync(CreateKey(), ExecuteRequest, () => "canceled", CancellationToken.None);
+
+            firstCancellationSource.Cancel();
+
+            Assert.That(await first.WaitAsync(_testTimeout), Is.EqualTo("canceled"));
+            Assert.That(requestCancellationToken.IsCancellationRequested, Is.False);
+            Assert.That(requestCount, Is.EqualTo(1));
+
+            requestCompletion.SetResult("result");
+
+            Assert.That(await second.WaitAsync(_testTimeout), Is.EqualTo("result"));
+        }
+
+        [Test]
+        public async Task ExecuteAsync_WhenAllListenersCancel_ShouldCancelSharedRequest()
+        {
+            var coalescer = CreateCoalescer();
+            var requestStarted = CreateCompletionSource<bool>();
+            var requestCanceled = CreateCompletionSource<bool>();
+            using var firstCancellationSource = new CancellationTokenSource();
+            using var secondCancellationSource = new CancellationTokenSource();
+
+            async Task<string> ExecuteRequest(CancellationToken cancellationToken)
+            {
+                using (cancellationToken.Register(() => requestCanceled.TrySetResult(true)))
+                {
+                    requestStarted.TrySetResult(true);
+                    await requestCanceled.Task;
+                    return "request canceled";
+                }
+            }
+
+            var first = coalescer.ExecuteAsync(CreateKey(), ExecuteRequest, () => "listener canceled", firstCancellationSource.Token);
+            await requestStarted.Task.WaitAsync(_testTimeout);
+            var second = coalescer.ExecuteAsync(CreateKey(), ExecuteRequest, () => "listener canceled", secondCancellationSource.Token);
+
+            firstCancellationSource.Cancel();
+            Assert.That(await first.WaitAsync(_testTimeout), Is.EqualTo("listener canceled"));
+            Assert.That(requestCanceled.Task.IsCompleted, Is.False);
+
+            secondCancellationSource.Cancel();
+
+            Assert.That(await second.WaitAsync(_testTimeout), Is.EqualTo("listener canceled"));
+            Assert.That(await requestCanceled.Task.WaitAsync(_testTimeout), Is.True);
+        }
+
+        [Test]
+        public async Task ExecuteAsync_WithAlreadyCanceledListener_ShouldNotStartRequest()
+        {
+            var coalescer = CreateCoalescer();
+            var requestCount = 0;
+            using var cancellationSource = new CancellationTokenSource();
+            cancellationSource.Cancel();
+
+            var result = await coalescer.ExecuteAsync(
+                CreateKey(),
+                cancellationToken =>
+                {
+                    Interlocked.Increment(ref requestCount);
+                    return Task.FromResult("result");
+                },
+                () => "canceled",
+                cancellationSource.Token);
+
+            Assert.That(result, Is.EqualTo("canceled"));
+            Assert.That(requestCount, Is.Zero);
+        }
+
+        [Test]
+        public async Task ExecuteAsync_AfterRequestCompletes_ShouldStartNewRequest()
+        {
+            var coalescer = CreateCoalescer();
+            var requestCount = 0;
+
+            Task<int> ExecuteRequest(CancellationToken cancellationToken)
+                => Task.FromResult(Interlocked.Increment(ref requestCount));
+
+            var first = await coalescer.ExecuteAsync(CreateKey(typeof(int)), ExecuteRequest, () => -1, CancellationToken.None);
+            var second = await coalescer.ExecuteAsync(CreateKey(typeof(int)), ExecuteRequest, () => -1, CancellationToken.None);
+
+            Assert.That(first, Is.EqualTo(1));
+            Assert.That(second, Is.EqualTo(2));
+            Assert.That(requestCount, Is.EqualTo(2));
+        }
+
+        [Test]
+        public void ExecuteAsync_AfterRequestFails_ShouldStartNewRequest()
+        {
+            var coalescer = CreateCoalescer();
+            var requestCount = 0;
+
+            Task<string> ExecuteRequest(CancellationToken cancellationToken)
+            {
+                if (Interlocked.Increment(ref requestCount) == 1)
+                    return Task.FromException<string>(new InvalidOperationException("Request failed"));
+
+                return Task.FromResult("result");
+            }
+
+            Assert.ThrowsAsync<InvalidOperationException>(async () =>
+                await coalescer.ExecuteAsync(CreateKey(), ExecuteRequest, () => "canceled", CancellationToken.None));
+
+            Assert.That(
+                coalescer.ExecuteAsync(CreateKey(), ExecuteRequest, () => "canceled", CancellationToken.None).GetAwaiter().GetResult(),
+                Is.EqualTo("result"));
+            Assert.That(requestCount, Is.EqualTo(2));
+        }
+
+        [Test]
+        public async Task ExecuteAsync_WhenAbandonedRequestCompletes_ShouldNotRemoveReplacementRequest()
+        {
+            var coalescer = CreateCoalescer();
+            var key = CreateKey();
+            var oldRequestCompletion = CreateCompletionSource<string>();
+            var oldRequestStarted = CreateCompletionSource<bool>();
+            var replacementCompletion = CreateCompletionSource<string>();
+            var replacementStarted = CreateCompletionSource<bool>();
+            var requestCount = 0;
+            using var cancellationSource = new CancellationTokenSource();
+
+            async Task<string> ExecuteOldRequest(CancellationToken cancellationToken)
+            {
+                Interlocked.Increment(ref requestCount);
+                oldRequestStarted.TrySetResult(true);
+                return await oldRequestCompletion.Task;
+            }
+
+            async Task<string> ExecuteReplacementRequest(CancellationToken cancellationToken)
+            {
+                Interlocked.Increment(ref requestCount);
+                replacementStarted.TrySetResult(true);
+                return await replacementCompletion.Task;
+            }
+
+            var abandonedListener = coalescer.ExecuteAsync(key, ExecuteOldRequest, () => "canceled", cancellationSource.Token);
+            await oldRequestStarted.Task.WaitAsync(_testTimeout);
+            var oldPhysicalRequest = GetPhysicalRequest(coalescer, key);
+
+            cancellationSource.Cancel();
+            Assert.That(await abandonedListener.WaitAsync(_testTimeout), Is.EqualTo("canceled"));
+
+            var replacementListener = coalescer.ExecuteAsync(key, ExecuteReplacementRequest, () => "canceled", CancellationToken.None);
+            await replacementStarted.Task.WaitAsync(_testTimeout);
+
+            oldRequestCompletion.SetResult("old result");
+            await oldPhysicalRequest.WaitAsync(_testTimeout);
+
+            var joinedReplacementListener = coalescer.ExecuteAsync(key, ExecuteReplacementRequest, () => "canceled", CancellationToken.None);
+            Assert.That(requestCount, Is.EqualTo(2));
+
+            replacementCompletion.SetResult("replacement result");
+
+            Assert.That(await replacementListener.WaitAsync(_testTimeout), Is.EqualTo("replacement result"));
+            Assert.That(await joinedReplacementListener.WaitAsync(_testTimeout), Is.EqualTo("replacement result"));
+            Assert.That(requestCount, Is.EqualTo(2));
+        }
+
+        private static RequestCoalescer CreateCoalescer()
+            => new RequestCoalescer(NullLogger.Instance);
+
+        private static RequestCoalescingKey CreateKey(Type? responseType = null, string? parameters = null)
+            => new RequestCoalescingKey("GET", "https://localhost/test", parameters, responseType ?? typeof(string));
+
+        private static TaskCompletionSource<T> CreateCompletionSource<T>()
+            => new TaskCompletionSource<T>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        private static Task GetPhysicalRequest(RequestCoalescer coalescer, RequestCoalescingKey key)
+        {
+            var requestsField = typeof(RequestCoalescer).GetField("_requests", BindingFlags.Instance | BindingFlags.NonPublic)!;
+            var requests = (IDictionary)requestsField.GetValue(coalescer)!;
+            var entry = requests[key]!;
+            var requestProperty = entry.GetType().GetProperty("Request")!;
+            return ((Lazy<Task<object>>)requestProperty.GetValue(entry)!).Value;
+        }
+    }
+}

CryptoExchange.Net/Clients/RestApiClient.cs

@@ -98,6 +98,10 @@ namespace CryptoExchange.Net.Clients
         /// Configured environment name
         /// </summary>
         public abstract string EnvironmentName { get; }
+        /// <summary>
+        /// Request coalescer
+        /// </summary>
+        protected RequestCoalescer RequestCoalescer { get; }
 
         /// <summary>
         /// ctor
@@ -123,6 +127,7 @@ namespace CryptoExchange.Net.Clients
         {
             TimeOffsetManager.RegisterRestApi(ClientName);
 
+            RequestCoalescer = new RequestCoalescer(_logger);
             RequestFactory.Configure(options, httpClient);
         }
 
@@ -177,8 +182,7 @@ namespace CryptoExchange.Net.Clients
         /// <param name="weight">Override the request weight for this request definition, for example when the weight depends on the parameters</param>
         /// <param name="weightSingleLimiter">Specify the weight to apply to the individual rate limit guard for this request</param>
         /// <param name="rateLimitKeySuffix">An additional optional suffix for the key selector. Can be used to make rate limiting work based on parameters.</param>
-        /// <returns></returns>
-        protected virtual async Task<HttpResult<T>> SendAsync<T>(
+        protected virtual Task<HttpResult<T>> SendAsync<T>(
             RequestDefinition definition,
             Parameters? uriParameters,
             Parameters? bodyParameters,
@@ -187,6 +191,61 @@ namespace CryptoExchange.Net.Clients
             int? weight = null,
             int? weightSingleLimiter = null,
             string? rateLimitKeySuffix = null)
+        {
+            if (!ShouldCoalesce(definition, additionalHeaders))
+            {
+                return SendCoreAsync<T>(
+                    definition,
+                    uriParameters,
+                    bodyParameters,
+                    cancellationToken,
+                    additionalHeaders,
+                    weight,
+                    weightSingleLimiter,
+                    rateLimitKeySuffix);
+            }
+
+            var key = GetRequestCoalescingKey<T>(
+                definition,
+                uriParameters);
+
+            return RequestCoalescer.ExecuteAsync(
+                key,
+                requestCancellationToken => SendCoreAsync<T>(
+                    definition,
+                    uriParameters,
+                    bodyParameters,
+                    requestCancellationToken,
+                    additionalHeaders,
+                    weight,
+                    weightSingleLimiter,
+                    rateLimitKeySuffix),
+                () => HttpResult.Fail<T>(
+                    Exchange,
+                    new CancellationRequestedError()),
+                cancellationToken);
+        }
+
+        /// <summary>
+        /// Send a request to the base address based on the request definition
+        /// </summary>
+        /// <typeparam name="T">Response type</typeparam>
+        /// <param name="definition">Request definition</param>
+        /// <param name="uriParameters">Request query parameters</param>
+        /// <param name="bodyParameters">Request body parameters</param>
+        /// <param name="cancellationToken">Cancellation token</param>
+        /// <param name="additionalHeaders">Additional headers for this request</param>
+        /// <param name="weight">Override the request weight for this request definition, for example when the weight depends on the parameters</param>
+        /// <param name="weightSingleLimiter">Specify the weight to apply to the individual rate limit guard for this request</param>
+        /// <param name="rateLimitKeySuffix">An additional optional suffix for the key selector. Can be used to make rate limiting work based on parameters.</param>
+        protected virtual async Task<HttpResult<T>> SendCoreAsync<T>(RequestDefinition definition,
+            Parameters? uriParameters,
+            Parameters? bodyParameters,
+            CancellationToken cancellationToken,
+            Dictionary<string, string>? additionalHeaders = null,
+            int? weight = null,
+            int? weightSingleLimiter = null,
+            string? rateLimitKeySuffix = null)
         {
             var requestId = ExchangeHelpers.NextId();
             if (definition.Authenticated && GetAuthenticationProvider() == null)
@@ -688,7 +747,10 @@ namespace CryptoExchange.Net.Clients
                         stringData = stringSerializer.Serialize(parameters.BodyValue);
                 }
                 else
+                {
                     stringData = stringSerializer.Serialize(parameters);
+                }
+
                 request.SetContent(stringData, RequestBodyContentEncoding, contentType);
             }
             else if (contentType == Constants.FormContentHeader)
@@ -808,6 +870,34 @@ namespace CryptoExchange.Net.Clients
             return definition.FullUrl + definition + parameters?.ToFormData();
         }
 
+        /// <summary>
+        /// Whether a request can be coalesced or not
+        /// </summary>
+        protected virtual bool ShouldCoalesce(
+            RequestDefinition definition,
+            Dictionary<string, string>? additionalHeaders)
+        {
+            return ClientOptions.RequestCoalescingEnabled                       // Enabled in client options
+                && !definition.PreventRequestCoalescing                         // Enabled in request definition
+                && definition.Method == HttpMethod.Get                          // Is a GET request
+                && !definition.Authenticated                                    // Not an authenticated request
+                && (additionalHeaders == null || additionalHeaders.Count == 0); // No additional headers that might change the response
+        }
+
+        /// <summary>
+        /// Get a unique key for coalescing requests based on the request definition, parameters, and the type of the expected response.
+        /// </summary>
+        protected virtual RequestCoalescingKey GetRequestCoalescingKey<T>(
+            RequestDefinition definition,
+            Parameters? uriParameters)
+        {
+            return new RequestCoalescingKey(
+                definition.Method.Method,
+                definition.FullUrl,
+                uriParameters?.ToFormData(),
+                typeof(T));
+        }
+
         /// <inheritdoc />
         public virtual void SetOptions(UpdateOptions options)
         {

CryptoExchange.Net/Logging/Extensions/RestApiClientLoggingExtensions.cs

@@ -22,6 +22,7 @@ namespace CryptoExchange.Net.Logging.Extensions
         private static readonly Action<ILogger, string, Exception?> _restApiCacheHit;
         private static readonly Action<ILogger, string, Exception?> _restApiCacheNotHit;
         private static readonly Action<ILogger, int?, Exception?> _restApiCancellationRequested;
+        private static readonly Action<ILogger, string, string, Exception?> _restApiRequestsJoined;
 
         static RestApiClientLoggingExtensions()
         {
@@ -89,7 +90,11 @@ namespace CryptoExchange.Net.Logging.Extensions
                 LogLevel.Debug,
                 new EventId(4012, "RestApiCancellationRequested"),
                 "[Req {RequestId}] request cancelled by user");
-            
+
+            _restApiRequestsJoined = LoggerMessage.Define<string, string>(
+                LogLevel.Debug,
+                new EventId(4013, "RestApiRequestsJoined"),
+                "Request for {Method} {Path} joined with in flight request");
         }
 
         public static void RestApiErrorReceived(this ILogger logger, int? requestId, HttpStatusCode? responseStatusCode, long responseTime, string? error, string? originalData, Exception? exception)
@@ -151,9 +156,15 @@ namespace CryptoExchange.Net.Logging.Extensions
         {
             _restApiCacheNotHit(logger, key, null);
         }
+
         public static void RestApiCancellationRequested(this ILogger logger, int? requestId)
         {
             _restApiCancellationRequested(logger, requestId, null);
         }
+
+        public static void RestApiRequestsJoined(this ILogger logger, string method, string path)
+        {
+            _restApiRequestsJoined(logger, method, path, null);
+        }
     }
 }

CryptoExchange.Net/Objects/Options/RestExchangeOptions.cs

@@ -34,6 +34,11 @@ namespace CryptoExchange.Net.Objects.Options
             = new Version(1, 1);
 #endif
 
+        /// <summary>
+        /// Whether concurrent identical REST requests may share the same in-flight request when allowed by the request definition.
+        /// </summary>
+        public bool RequestCoalescingEnabled { get; set; } = true;
+
         /// <summary>
         /// Http client keep alive interval for keeping connections open. Only applied when using dotnet8.0 or higher and dependency injection
         /// </summary>
@@ -73,6 +78,7 @@ namespace CryptoExchange.Net.Objects.Options
             item.CachingEnabled = CachingEnabled;
             item.CachingMaxAge = CachingMaxAge;
             item.HttpVersion = HttpVersion;
+            item.RequestCoalescingEnabled = RequestCoalescingEnabled;
             item.HttpKeepAliveInterval = HttpKeepAliveInterval;
 #if NET5_0_OR_GREATER
             item.HttpMaxConnectionsPerServer = HttpMaxConnectionsPerServer;

CryptoExchange.Net/Objects/RequestCoalescer.cs

@@ -0,0 +1,277 @@
+using CryptoExchange.Net.Logging.Extensions;
+using Microsoft.Extensions.Logging;
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace CryptoExchange.Net.Objects
+{
+    /// <summary>
+    /// Request coalescer, used to detect multiple identical requests and wait for and return only a single actual request result
+    /// </summary>
+    public class RequestCoalescer
+    {
+        private sealed class RequestEntry
+        {
+            public RequestCoalescingKey Key { get; }
+
+            public CancellationTokenSource CancellationSource { get; } = new();
+
+            public Lazy<Task<object>> Request { get; set; } = null!;
+
+            public int ListenerCount { get; set; }
+
+            public bool Completed { get; set; }
+
+            public bool CancellationInProgress { get; set; }
+
+            public bool Disposed { get; set; }
+
+            public RequestEntry(RequestCoalescingKey key)
+            {
+                Key = key;
+            }
+        }
+
+        private readonly object _sync = new();
+        private readonly ILogger _logger;
+
+        private readonly Dictionary<RequestCoalescingKey, RequestEntry> _requests = new();
+
+        /// <summary>
+        /// ctor
+        /// </summary>
+        public RequestCoalescer(ILogger logger)
+        {
+            _logger = logger;
+        }
+
+        /// <summary>
+        /// Execute a request, coalescing identical requests into a single request
+        /// </summary>
+        public Task<TResult> ExecuteAsync<TResult>(
+            RequestCoalescingKey key,
+            Func<CancellationToken, Task<TResult>> requestFactory,
+            Func<TResult> cancellationResultFactory,
+            CancellationToken cancellationToken)
+        {
+            if (cancellationToken.IsCancellationRequested)
+                return Task.FromResult(cancellationResultFactory());
+
+            RequestEntry entry;
+            bool joined;
+
+            lock (_sync)
+            {
+                joined = _requests.TryGetValue(key, out entry!);
+
+                if (!joined)
+                {
+                    entry = CreateEntry(key, requestFactory);
+                    _requests.Add(key, entry);
+                }
+
+                entry.ListenerCount++;
+            }
+
+            if (joined)
+                _logger.RestApiRequestsJoined(key.Method, key.Url);
+
+            return WaitAsync(
+                entry,
+                cancellationResultFactory,
+                cancellationToken);
+        }
+
+        private RequestEntry CreateEntry<TResult>(
+            RequestCoalescingKey key,
+            Func<CancellationToken, Task<TResult>> requestFactory)
+        {
+            var entry = new RequestEntry(key);
+
+            entry.Request = new Lazy<Task<object>>(
+                () => ExecuteCoreAsync(
+                    entry,
+                    async requestCancellationToken =>
+                        (await requestFactory(requestCancellationToken).ConfigureAwait(false))!),
+                LazyThreadSafetyMode.ExecutionAndPublication);
+
+            return entry;
+        }
+
+        private async Task<object> ExecuteCoreAsync(
+            RequestEntry entry,
+            Func<CancellationToken, Task<object>> requestFactory)
+        {
+            try
+            {
+                return await requestFactory(entry.CancellationSource.Token).ConfigureAwait(false);
+            }
+            finally
+            {
+                CompleteRequest(entry);
+            }
+        }
+
+        private async Task<TResult> WaitAsync<TResult>(
+            RequestEntry entry,
+            Func<TResult> cancellationResultFactory,
+            CancellationToken cancellationToken)
+        {
+            try
+            {
+                var requestTask = entry.Request.Value;
+
+                if (!cancellationToken.CanBeCanceled)
+                    return (TResult)await requestTask.ConfigureAwait(false);
+
+                var cancellationCompletion = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+                using (cancellationToken.Register(
+                    state => ((TaskCompletionSource<bool>)state!).TrySetResult(true),
+                    cancellationCompletion))
+                {
+                    var completedTask = await Task.WhenAny(requestTask, cancellationCompletion.Task).ConfigureAwait(false);
+                    if (completedTask != requestTask)
+                        return cancellationResultFactory();
+                }
+
+                return (TResult)await requestTask.ConfigureAwait(false);
+            }
+            finally
+            {
+                ReleaseListener(entry);
+            }
+        }
+
+        private void ReleaseListener(RequestEntry entry)
+        {
+            var cancelRequest = false;
+            var disposeCancellationSource = false;
+
+            lock (_sync)
+            {
+                entry.ListenerCount--;
+
+                if (entry.ListenerCount == 0)
+                {
+                    if (!entry.Completed)
+                    {
+                        // Remove the entry before canceling it. A caller arriving
+                        // after this point must create a new physical request
+                        // instead of joining one that is being canceled.
+                        RemoveEntry(entry);
+
+                        entry.CancellationInProgress = true;
+                        cancelRequest = true;
+                    }
+                    else if (!entry.Disposed)
+                    {
+                        entry.Disposed = true;
+                        disposeCancellationSource = true;
+                    }
+                }
+            }
+
+            if (cancelRequest)
+            {
+                try
+                {
+                    entry.CancellationSource.Cancel();
+                }
+                catch (Exception exception)
+                {
+                    // Cancellation callbacks are external to the coalescer and
+                    // must not cause an exception to escape the library.
+                    _logger.LogWarning(
+                        exception,
+                        "Error canceling in-flight request {Method} {Url}",
+                        entry.Key.Method,
+                        entry.Key.Url);
+                }
+                finally
+                {
+                    CompleteCancellation(entry);
+                }
+            }
+            else if (disposeCancellationSource)
+            {
+                entry.CancellationSource.Dispose();
+            }
+        }
+
+        private void CompleteCancellation(RequestEntry entry)
+        {
+            var disposeCancellationSource = false;
+
+            lock (_sync)
+            {
+                entry.CancellationInProgress = false;
+
+                if (entry.Completed
+                    && entry.ListenerCount == 0
+                    && !entry.Disposed)
+                {
+                    entry.Disposed = true;
+                    disposeCancellationSource = true;
+                }
+            }
+
+            if (disposeCancellationSource)
+                entry.CancellationSource.Dispose();
+        }
+
+        private void CompleteRequest(RequestEntry entry)
+        {
+            var disposeCancellationSource = false;
+
+            lock (_sync)
+            {
+                entry.Completed = true;
+
+                // Only remove this specific entry. If all listeners canceled,
+                // a replacement request may already have been created for the
+                // same key.
+                RemoveEntry(entry);
+
+                if (entry.ListenerCount == 0
+                    && !entry.CancellationInProgress
+                    && !entry.Disposed)
+                {
+                    entry.Disposed = true;
+                    disposeCancellationSource = true;
+                }
+            }
+
+            if (disposeCancellationSource)
+                entry.CancellationSource.Dispose();
+        }
+
+        /// <summary>
+        /// Remove an entry when it is still the current entry for its key.
+        /// Must only be called while holding <see cref="_sync"/>.
+        /// </summary>
+        private void RemoveEntry(RequestEntry entry)
+        {
+            if (_requests.TryGetValue(entry.Key, out var current)
+                && ReferenceEquals(current, entry))
+            {
+                _requests.Remove(entry.Key);
+            }
+        }
+    }
+
+    /// <summary>
+    /// Key used to identify identical requests for coalescing
+    /// </summary>
+    /// <param name="Method">Request method</param>
+    /// <param name="Url">Request URL</param>
+    /// <param name="Parameters">Request parameter string</param>
+    /// <param name="ResponseType">Response type</param>
+    public sealed record RequestCoalescingKey(
+        string Method,
+        string Url,
+        string? Parameters,
+        Type ResponseType);
+}

CryptoExchange.Net/Objects/RequestDefinition.cs

@@ -101,6 +101,11 @@ namespace CryptoExchange.Net.Objects
             }   
         }
 
+        /// <summary>
+        /// Whether concurrent identical requests should be prevented from sharing the same request/response
+        /// </summary>
+        public bool PreventRequestCoalescing { get; set; }
+
         /// <summary>
         /// ctor
         /// </summary>

CryptoExchange.Net/Objects/RequestDefinitionCache.cs

@@ -53,6 +53,7 @@ namespace CryptoExchange.Net.Objects
         /// <param name="tryParseOnNonSuccess">Try parse the response even when status is not success</param>
         /// <param name="forcePathEndWithSlash">Force trailing `/`</param>
         /// <param name="identifier">Optional request identifier override</param>
+        /// <param name="preventRequestCoalescing">Whether concurrent identical requests should be prevented from sharing the same request/response</param>
         /// <returns></returns>
         public RequestDefinition GetOrCreate(
             HttpMethod method,
@@ -68,7 +69,8 @@ namespace CryptoExchange.Net.Objects
             bool? preventCaching = null,
             bool? tryParseOnNonSuccess = null,
             bool? forcePathEndWithSlash = null,
-            string? identifier = null)
+            string? identifier = null,
+            bool? preventRequestCoalescing = null)
         {
             var identifierToUse = identifier ?? $"{path}{method.Method}{baseAddress}";
             if (!_definitions.TryGetValue(identifierToUse, out var def))
@@ -85,6 +87,7 @@ namespace CryptoExchange.Net.Objects
                     PreventCaching = preventCaching ?? false,
                     TryParseOnNonSuccess = tryParseOnNonSuccess ?? false,
                     ForcePathEndWithSlash = forcePathEndWithSlash ?? false,
+                    PreventRequestCoalescing = preventRequestCoalescing ?? false,
                 };
                 _definitions.TryAdd(identifierToUse, def);
             }

CryptoExchange.Net/Objects/Sockets/DataEvent.cs

@@ -165,7 +165,10 @@ namespace CryptoExchange.Net.Objects.Sockets
             {
                 StreamId = StreamId,
                 UpdateType = UpdateType,
-                Symbol = Symbol                
+                Symbol = Symbol,
+                DataTime = DataTime,
+                DataTimeLocal = DataTimeLocal,
+                SequenceNumber = SequenceNumber
             };
         }
 

CryptoExchange.Net/SharedApis/Enums/SharedOrderBookSubscriptionType.cs

@@ -0,0 +1,21 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace CryptoExchange.Net.SharedApis
+{
+    /// <summary>
+    /// The type of updates the incremental order book subscription produces
+    /// </summary>
+    public enum SharedOrderBookSubscriptionType
+    {
+        /// <summary>
+        /// Subscription produces an initial snapshot update, followed by incremental change updates
+        /// </summary>
+        SnapshotThenIncremental,
+        /// <summary>
+        /// Subscription produces only incremental change updates
+        /// </summary>
+        OnlyIncremental
+    }
+}

CryptoExchange.Net/SharedApis/Interfaces/V2/Endpoints/Futures/UserTrades/IGetFuturesUserTradeHistoryEndpoint.cs

@@ -25,6 +25,6 @@ namespace CryptoExchange.Net.SharedApis
         /// <param name="request">Request info</param>
         /// <param name="nextPageToken">The pagination request from the previous request result <see cref="HttpResult{T}.NextPageRequest"/> property to continue pagination</param>
         /// <param name="ct">Cancellation token</param>
-        Task<HttpResult<SharedUserTrade[]>> GetFuturesUserTradesAsync(GetUserTradesRequest request, PageRequest? nextPageToken = null, CancellationToken ct = default);
+        Task<HttpResult<SharedUserTrade[]>> GetFuturesUserTradeHistoryAsync(GetUserTradesRequest request, PageRequest? nextPageToken = null, CancellationToken ct = default);
     }
 }

CryptoExchange.Net/SharedApis/Interfaces/V2/Operations/Futures/ISubscribeFuturesOrdersOperation.cs

@@ -23,6 +23,6 @@ namespace CryptoExchange.Net.SharedApis
         /// <param name="handler">Update handler</param>
         /// <param name="ct">Cancellation token, can be used to stop the updates</param>
         /// <returns></returns>
-        Task<WebSocketResult<UpdateSubscription>> SubscribeToFuturesOrderUpdatesAsync(SubscribeFuturesOrderRequest request, Action<DataEvent<SharedFuturesOrder[]>> handler, CancellationToken ct = default);
+        Task<WebSocketResult<UpdateSubscription>> SubscribeToFuturesOrderUpdatesAsync(SubscribeFuturesOrderRequest request, Action<DataEvent<SharedFuturesOrderUpdate[]>> handler, CancellationToken ct = default);
     }
 }

CryptoExchange.Net/SharedApis/Interfaces/V2/Operations/ISubscribeIncrementalOrderBookOperation.cs

@@ -0,0 +1,28 @@
+using CryptoExchange.Net.Objects;
+using CryptoExchange.Net.Objects.Sockets;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace CryptoExchange.Net.SharedApis
+{
+    /// <summary>
+    /// Client for subscribing to incremental order book updates for a symbol
+    /// </summary>
+    public interface ISubscribeIncrementalOrderBookOperation : ISharedSubscription
+    {
+        /// <summary>
+        /// Order book subscription options
+        /// </summary>
+        SubscribeOrderBookOptions SubscribeOrderBookOptions { get; }
+
+        /// <summary>
+        /// Subscribe to incremental order book updates for a symbol
+        /// </summary>
+        /// <param name="request">Request info</param>
+        /// <param name="handler">Update handler</param>
+        /// <param name="ct">Cancellation token, can be used to stop the updates</param>
+        /// <returns></returns>
+        Task<WebSocketResult<UpdateSubscription>> SubscribeToOrderBookUpdatesAsync(SubscribeOrderBookRequest request, Action<DataEvent<SharedOrderBook>> handler, CancellationToken ct = default);
+    }
+}

CryptoExchange.Net/SharedApis/Interfaces/V2/Operations/Spot/ISubscribeSpotOrdersOperation.cs

@@ -23,6 +23,6 @@ namespace CryptoExchange.Net.SharedApis
         /// <param name="handler">Update handler</param>
         /// <param name="ct">Cancellation token, can be used to stop the updates</param>
         /// <returns></returns>
-        Task<WebSocketResult<UpdateSubscription>> SubscribeToSpotOrderUpdatesAsync(SubscribeSpotOrderRequest request, Action<DataEvent<SharedSpotOrder[]>> handler, CancellationToken ct = default);
+        Task<WebSocketResult<UpdateSubscription>> SubscribeToSpotOrderUpdatesAsync(SubscribeSpotOrderRequest request, Action<DataEvent<SharedSpotOrderUpdate[]>> handler, CancellationToken ct = default);
     }
 }

CryptoExchange.Net/SharedApis/Models/Options/Endpoints/GetFuturesUserTradeHistoryOptions.cs

@@ -16,7 +16,7 @@ namespace CryptoExchange.Net.SharedApis
         /// ctor
         /// </summary>
         public GetFuturesUserTradeHistoryOptions(string exchange, bool supportsAscending, bool supportsDescending, bool timeFilterSupported, int maxLimit) 
-            : base(exchange, supportsAscending, supportsDescending, timeFilterSupported, maxLimit, true, nameof(IGetFuturesUserTradeHistoryEndpoint.GetFuturesUserTradesAsync))
+            : base(exchange, supportsAscending, supportsDescending, timeFilterSupported, maxLimit, true, nameof(IGetFuturesUserTradeHistoryEndpoint.GetFuturesUserTradeHistoryAsync))
         {
         }
 

CryptoExchange.Net/SharedApis/Models/Options/Operations/SubscribeIncrementalOrderBookOptions.cs

@@ -0,0 +1,46 @@
+using CryptoExchange.Net.Objects;
+using System;
+using System.Linq;
+
+namespace CryptoExchange.Net.SharedApis
+{
+    /// <summary>
+    /// Options for subscribing to order book snapshot updates
+    /// </summary>
+    public class SubscribeIncrementalOrderBookOptions : CapabilityOptions<SubscribeOrderBookRequest, ISubscribeIncrementalOrderBookOperation>
+    {
+        /// <inheritdoc />
+        public override string Description => "Subscribe to incremental order book updates for a symbol";
+
+        /// <summary>
+        /// Order book depths supported for updates
+        /// </summary>
+        public int[] SupportedLimits { get; }
+
+        /// <summary>
+        /// The type of updates the subscription produces
+        /// </summary>
+        public SharedOrderBookSubscriptionType UpdateType { get; }
+
+        /// <summary>
+        /// ctor
+        /// </summary>
+        public SubscribeIncrementalOrderBookOptions(string exchange, bool needsAuthentication, int[] limits, SharedOrderBookSubscriptionType updateType)
+            : base(exchange, needsAuthentication, nameof(ISubscribeIncrementalOrderBookOperation.SubscribeToOrderBookUpdatesAsync))
+        {
+            SupportedLimits = limits;
+            UpdateType = updateType;
+        }
+
+        /// <summary>
+        /// Validate a request
+        /// </summary>
+        public override Error? ValidateRequest(SubscribeOrderBookRequest request, ISubscribeIncrementalOrderBookOperation client)
+        {
+            if (request.Limit != null && !SupportedLimits.Contains(request.Limit.Value))
+                return ArgumentError.Invalid(nameof(SubscribeOrderBookRequest.Limit), "Limit not supported");
+
+            return base.ValidateRequest(request, client);
+        }
+    }
+}

CryptoExchange.Net/SharedApis/ResponseModels/SharedFuturesOrder.cs

@@ -73,10 +73,12 @@ namespace CryptoExchange.Net.SharedApis
         /// <summary>
         /// Asset the fee is in
         /// </summary>
+        [Obsolete("FeeAsset on order level is deprecated and will be removed in a futures version, use FeeAsset on trade level instead")]
         public string? FeeAsset { get; set; }
         /// <summary>
         /// Fee paid
         /// </summary>
+        [Obsolete("Fee on order level is deprecated and will be removed in a futures version, use Fee on trade level instead")]
         public decimal? Fee { get; set; }
         /// <summary>
         /// Leverage
@@ -94,6 +96,7 @@ namespace CryptoExchange.Net.SharedApis
         /// <summary>
         /// Last trade info, only available for websocket order updates if the API provides this data in the update
         /// </summary>
+        [Obsolete("Use SharedFuturesOrderUpdate.LastTrade instead, LastTrade is never filled on non websocket updates")]
         public SharedUserTrade? LastTrade { get; set; }
 
         /// <summary>

CryptoExchange.Net/SharedApis/ResponseModels/SharedFuturesOrderUpdate.cs

@@ -0,0 +1,38 @@
+using System;
+using System.Diagnostics;
+
+namespace CryptoExchange.Net.SharedApis
+{
+    /// <summary>
+    /// Futures order info
+    /// </summary>
+    [DebuggerDisplay("{DebugView,nq}")]
+    public record SharedFuturesOrderUpdate : SharedFuturesOrder
+    {
+        /// <summary>
+        /// The info on the executed trade for this update
+        /// </summary>
+        public new SharedUserTrade? LastTrade
+        {
+#pragma warning disable CS0618 // Type or member is obsolete
+            get => base.LastTrade;
+            set => base.LastTrade = value;
+#pragma warning restore CS0618 // Type or member is obsolete
+        }
+
+        /// <summary>
+        /// ctor
+        /// </summary>
+        public SharedFuturesOrderUpdate(
+            SharedSymbol? sharedSymbol, 
+            string symbol,
+            string orderId,
+            SharedOrderType orderType,
+            SharedOrderSide orderSide,
+            SharedOrderStatus orderStatus,
+            DateTime? createTime)
+            : base(sharedSymbol, symbol, orderId, orderType, orderSide, orderStatus, createTime)
+        {
+        }
+    }
+}

CryptoExchange.Net/SharedApis/ResponseModels/SharedOrderBook.cs

@@ -12,6 +12,10 @@ namespace CryptoExchange.Net.SharedApis
         /// </summary>
         public SharedQuantityType QuantityType { get; set; }
         /// <summary>
+        /// The sequence number this snapshot or update is at
+        /// </summary>
+        public long? SequenceNumber { get; set; }
+        /// <summary>
         /// Asks list
         /// </summary>
         public ISymbolOrderBookEntry[] Asks { get; set; }
@@ -23,9 +27,10 @@ namespace CryptoExchange.Net.SharedApis
         /// <summary>
         /// ctor
         /// </summary>
-        public SharedOrderBook(SharedQuantityType quantityType, ISymbolOrderBookEntry[] asks, ISymbolOrderBookEntry[] bids)
+        public SharedOrderBook(SharedQuantityType quantityType, long? sequenceNumber, ISymbolOrderBookEntry[] asks, ISymbolOrderBookEntry[] bids)
         {
             QuantityType = quantityType;
+            SequenceNumber = sequenceNumber;
             Asks = asks;
             Bids = bids;
         }

CryptoExchange.Net/SharedApis/ResponseModels/SharedSpotOrder.cs

@@ -65,10 +65,12 @@ namespace CryptoExchange.Net.SharedApis
         /// <summary>
         /// Asset the fee is in
         /// </summary>
+        [Obsolete("FeeAsset on order level is deprecated and will be removed in a futures version, use FeeAsset on trade level instead")]
         public string? FeeAsset { get; set; }
         /// <summary>
         /// Fee paid for the order
         /// </summary>
+        [Obsolete("Fee on order level is deprecated and will be removed in a futures version, use Fee on trade level instead")]
         public decimal? Fee { get; set; }
         /// <summary>
         /// Timestamp the order was created
@@ -79,8 +81,9 @@ namespace CryptoExchange.Net.SharedApis
         /// </summary>
         public DateTime? UpdateTime { get; set; }
         /// <summary>
-        /// Last trade info, only available for websocket order updates if the API provides this data in the update
+        /// Only available for order updates, contains the info on the executed trade
         /// </summary>
+        [Obsolete("Use SharedSpotOrderUpdate.LastTrade instead, LastTrade is never filled on non websocket updates")]
         public SharedUserTrade? LastTrade { get; set; }
 
         /// <summary>

CryptoExchange.Net/SharedApis/ResponseModels/SharedSpotOrderUpdate.cs

@@ -0,0 +1,38 @@
+using System;
+using System.Diagnostics;
+
+namespace CryptoExchange.Net.SharedApis
+{
+    /// <summary>
+    /// Spot order info
+    /// </summary>
+    [DebuggerDisplay("{DebugView,nq}")]
+    public record SharedSpotOrderUpdate : SharedSpotOrder
+    {
+        /// <summary>
+        /// The info on the executed trade for this update
+        /// </summary>
+        public new SharedUserTrade? LastTrade
+        {
+#pragma warning disable CS0618 // Type or member is obsolete
+            get => base.LastTrade;
+            set => base.LastTrade = value;
+#pragma warning restore CS0618 // Type or member is obsolete
+        }
+
+        /// <summary>
+        /// ctor
+        /// </summary>
+        public SharedSpotOrderUpdate(
+            SharedSymbol? sharedSymbol,
+            string symbol,
+            string orderId,
+            SharedOrderType orderType,
+            SharedOrderSide orderSide,
+            SharedOrderStatus orderStatus,
+            DateTime? createTime)
+            : base(sharedSymbol, symbol, orderId, orderType, orderSide, orderStatus, createTime)
+        {
+        }
+    }
+}

CryptoExchange.Net/Trackers/UserData/ItemTrackers/FuturesOrderTracker.cs

@@ -83,18 +83,6 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
                 changed = true;
             }
 
-            if (updateItem.Fee != null && updateItem.Fee != existingItem.Fee)
-            {
-                existingItem.Fee = updateItem.Fee;
-                changed = true;
-            }
-
-            if (updateItem.FeeAsset != null && updateItem.FeeAsset != existingItem.FeeAsset)
-            {
-                existingItem.FeeAsset = updateItem.FeeAsset;
-                changed = true;
-            }
-
             if (updateItem.OrderQuantity != null && updateItem.OrderQuantity != existingItem.OrderQuantity)
             {
                 existingItem.OrderQuantity = updateItem.OrderQuantity;
@@ -196,16 +184,6 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
                 }
             }
 
-            if (existingItem.Fee != null && updateItem.Fee != null)
-            {
-                // Higher fee means later processing
-                if (existingItem.Fee < updateItem.Fee)
-                    return true;
-
-                if (existingItem.Fee > updateItem.Fee)
-                    return false;
-            }
-
             return null;
         }
 
@@ -214,7 +192,7 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
         {
             await base.HandleUpdateAsync(source, @event).ConfigureAwait(false);
 
-            var trades = @event.Where(x => x.LastTrade != null).Select(x => x.LastTrade!).ToArray();
+            var trades = @event.OfType<SharedFuturesOrderUpdate>().Where(x => x.LastTrade != null).Select(x => x.LastTrade!).ToArray();
             if (trades.Length != 0 && OnTradeUpdate != null)
                 await OnTradeUpdate.Invoke(source, trades).ConfigureAwait(false);
         }
@@ -225,7 +203,7 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
             if (_socketClient == null)
                 return Task.FromResult(new WebSocketResult<UpdateSubscription?>(_exchange, default!, default));
 
-            return ExchangeHelpers.ProcessQueuedAsync<SharedFuturesOrder[]>(
+            return ExchangeHelpers.ProcessQueuedAsync<SharedFuturesOrderUpdate[]>(
                 async handler => await _socketClient.SubscribeToFuturesOrderUpdatesAsync(new SubscribeFuturesOrderRequest(exchangeParameters: _exchangeParameters), handler, ct: _cts!.Token).ConfigureAwait(false),
                 x => HandleUpdateAsync(UpdateSource.Push, x.Data))!;
         }

CryptoExchange.Net/Trackers/UserData/ItemTrackers/FuturesUserTradeTracker.cs

@@ -75,7 +75,7 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
             var updatedPollTime = DateTime.UtcNow;
             foreach (var symbol in _symbolTracker.GetTrackedSymbols())
             {
-                var tradesResult = await _restClient.GetFuturesUserTradesAsync(new GetUserTradesRequest(symbol, startTime: fromTimeTrades, exchangeParameters: _exchangeParameters)).ConfigureAwait(false);
+                var tradesResult = await _restClient.GetFuturesUserTradeHistoryAsync(new GetUserTradesRequest(symbol, startTime: fromTimeTrades, exchangeParameters: _exchangeParameters)).ConfigureAwait(false);
                 if (!tradesResult.Success)
                 {
                     anyError = true;

CryptoExchange.Net/Trackers/UserData/ItemTrackers/SpotOrderTracker.cs

@@ -82,18 +82,6 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
                 changed = true;
             }
 
-            if (updateItem.Fee != null && updateItem.Fee != existingItem.Fee)
-            {
-                existingItem.Fee = updateItem.Fee;
-                changed = true;
-            }
-
-            if (updateItem.FeeAsset != null && updateItem.FeeAsset != existingItem.FeeAsset)
-            {
-                existingItem.FeeAsset = updateItem.FeeAsset;
-                changed = true;
-            }
-
             if (updateItem.OrderQuantity != null && updateItem.OrderQuantity != existingItem.OrderQuantity)
             {
                 existingItem.OrderQuantity ??= new SharedOrderQuantity();
@@ -206,16 +194,6 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
                 }
             }
 
-            if (existingItem.Fee != null && updateItem.Fee != null)
-            {
-                // Higher fee means later processing
-                if (existingItem.Fee < updateItem.Fee)
-                    return true;
-
-                if (existingItem.Fee > updateItem.Fee)
-                    return false;
-            }
-
             return null;
         }
 
@@ -224,7 +202,7 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
         {
             await base.HandleUpdateAsync(source, @event).ConfigureAwait(false);
 
-            var trades = @event.Where(x => x.LastTrade != null).Select(x => x.LastTrade!).ToArray();
+            var trades = @event.OfType<SharedSpotOrderUpdate>().Where(x => x.LastTrade != null).Select(x => x.LastTrade!).ToArray();
             if (trades.Length != 0 && OnTradeUpdate != null)
                 await OnTradeUpdate(source, trades).ConfigureAwait(false);
         }
@@ -235,7 +213,7 @@ namespace CryptoExchange.Net.Trackers.UserData.ItemTrackers
             if (_socketClient == null)
                 return Task.FromResult(new WebSocketResult<UpdateSubscription?>(_exchange!, default!, default));
 
-            return ExchangeHelpers.ProcessQueuedAsync<SharedSpotOrder[]>(
+            return ExchangeHelpers.ProcessQueuedAsync<SharedSpotOrderUpdate[]>(
                 async handler => await _socketClient.SubscribeToSpotOrderUpdatesAsync(new SubscribeSpotOrderRequest(exchangeParameters: _exchangeParameters), handler, ct: _cts!.Token).ConfigureAwait(false),
                 x => HandleUpdateAsync(UpdateSource.Push, x.Data))!;
         }