diff --git a/CryptoExchange.Net.UnitTests/RequestCoalescerTests.cs b/CryptoExchange.Net.UnitTests/RequestCoalescerTests.cs new file mode 100644 index 00000000..6d6cfc7e --- /dev/null +++ b/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(); + var requestStarted = CreateCompletionSource(); + var requestCount = 0; + + async Task 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(); + var secondCompletion = CreateCompletionSource(); + var firstStarted = CreateCompletionSource(); + var secondStarted = CreateCompletionSource(); + 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(); + var requestStarted = CreateCompletionSource(); + var requestCount = 0; + using var firstCancellationSource = new CancellationTokenSource(); + + async Task 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(); + var requestCanceled = CreateCompletionSource(); + using var firstCancellationSource = new CancellationTokenSource(); + using var secondCancellationSource = new CancellationTokenSource(); + + async Task 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 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 ExecuteRequest(CancellationToken cancellationToken) + { + if (Interlocked.Increment(ref requestCount) == 1) + return Task.FromException(new InvalidOperationException("Request failed")); + + return Task.FromResult("result"); + } + + Assert.ThrowsAsync(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(); + var oldRequestStarted = CreateCompletionSource(); + var replacementCompletion = CreateCompletionSource(); + var replacementStarted = CreateCompletionSource(); + var requestCount = 0; + using var cancellationSource = new CancellationTokenSource(); + + async Task ExecuteOldRequest(CancellationToken cancellationToken) + { + Interlocked.Increment(ref requestCount); + oldRequestStarted.TrySetResult(true); + return await oldRequestCompletion.Task; + } + + async Task 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 CreateCompletionSource() + => new TaskCompletionSource(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>)requestProperty.GetValue(entry)!).Value; + } + } +} diff --git a/CryptoExchange.Net/Clients/RestApiClient.cs b/CryptoExchange.Net/Clients/RestApiClient.cs index 98c8c985..8924291b 100644 --- a/CryptoExchange.Net/Clients/RestApiClient.cs +++ b/CryptoExchange.Net/Clients/RestApiClient.cs @@ -98,6 +98,10 @@ namespace CryptoExchange.Net.Clients /// Configured environment name /// public abstract string EnvironmentName { get; } + /// + /// Request coalescer + /// + protected RequestCoalescer RequestCoalescer { get; } /// /// 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 /// Override the request weight for this request definition, for example when the weight depends on the parameters /// Specify the weight to apply to the individual rate limit guard for this request /// An additional optional suffix for the key selector. Can be used to make rate limiting work based on parameters. - /// - protected virtual async Task> SendAsync( + protected virtual Task> SendAsync( 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( + definition, + uriParameters, + bodyParameters, + cancellationToken, + additionalHeaders, + weight, + weightSingleLimiter, + rateLimitKeySuffix); + } + + var key = GetRequestCoalescingKey( + definition, + uriParameters); + + return RequestCoalescer.ExecuteAsync( + key, + requestCancellationToken => SendCoreAsync( + definition, + uriParameters, + bodyParameters, + requestCancellationToken, + additionalHeaders, + weight, + weightSingleLimiter, + rateLimitKeySuffix), + () => HttpResult.Fail( + Exchange, + new CancellationRequestedError()), + cancellationToken); + } + + /// + /// Send a request to the base address based on the request definition + /// + /// Response type + /// Request definition + /// Request query parameters + /// Request body parameters + /// Cancellation token + /// Additional headers for this request + /// Override the request weight for this request definition, for example when the weight depends on the parameters + /// Specify the weight to apply to the individual rate limit guard for this request + /// An additional optional suffix for the key selector. Can be used to make rate limiting work based on parameters. + protected virtual async Task> SendCoreAsync(RequestDefinition definition, + Parameters? uriParameters, + Parameters? bodyParameters, + CancellationToken cancellationToken, + Dictionary? 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(); } + /// + /// Whether a request can be coalesced or not + /// + protected virtual bool ShouldCoalesce( + RequestDefinition definition, + Dictionary? 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 + } + + /// + /// Get a unique key for coalescing requests based on the request definition, parameters, and the type of the expected response. + /// + protected virtual RequestCoalescingKey GetRequestCoalescingKey( + RequestDefinition definition, + Parameters? uriParameters) + { + return new RequestCoalescingKey( + definition.Method.Method, + definition.FullUrl, + uriParameters?.ToFormData(), + typeof(T)); + } + /// public virtual void SetOptions(UpdateOptions options) { diff --git a/CryptoExchange.Net/Logging/Extensions/RestApiClientLoggingExtensions.cs b/CryptoExchange.Net/Logging/Extensions/RestApiClientLoggingExtensions.cs index 901d935c..546b5cab 100644 --- a/CryptoExchange.Net/Logging/Extensions/RestApiClientLoggingExtensions.cs +++ b/CryptoExchange.Net/Logging/Extensions/RestApiClientLoggingExtensions.cs @@ -22,6 +22,7 @@ namespace CryptoExchange.Net.Logging.Extensions private static readonly Action _restApiCacheHit; private static readonly Action _restApiCacheNotHit; private static readonly Action _restApiCancellationRequested; + private static readonly Action _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( + 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); + } } } diff --git a/CryptoExchange.Net/Objects/Options/RestExchangeOptions.cs b/CryptoExchange.Net/Objects/Options/RestExchangeOptions.cs index dfbb87c1..ea0e1155 100644 --- a/CryptoExchange.Net/Objects/Options/RestExchangeOptions.cs +++ b/CryptoExchange.Net/Objects/Options/RestExchangeOptions.cs @@ -34,6 +34,11 @@ namespace CryptoExchange.Net.Objects.Options = new Version(1, 1); #endif + /// + /// Whether concurrent identical REST requests may share the same in-flight request when allowed by the request definition. + /// + public bool RequestCoalescingEnabled { get; set; } = true; + /// /// Http client keep alive interval for keeping connections open. Only applied when using dotnet8.0 or higher and dependency injection /// @@ -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; diff --git a/CryptoExchange.Net/Objects/RequestCoalescer.cs b/CryptoExchange.Net/Objects/RequestCoalescer.cs new file mode 100644 index 00000000..46dc6f2d --- /dev/null +++ b/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 +{ + /// + /// Request coalescer, used to detect multiple identical requests and wait for and return only a single actual request result + /// + public class RequestCoalescer + { + private sealed class RequestEntry + { + public RequestCoalescingKey Key { get; } + + public CancellationTokenSource CancellationSource { get; } = new(); + + public Lazy> 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 _requests = new(); + + /// + /// ctor + /// + public RequestCoalescer(ILogger logger) + { + _logger = logger; + } + + /// + /// Execute a request, coalescing identical requests into a single request + /// + public Task ExecuteAsync( + RequestCoalescingKey key, + Func> requestFactory, + Func 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( + RequestCoalescingKey key, + Func> requestFactory) + { + var entry = new RequestEntry(key); + + entry.Request = new Lazy>( + () => ExecuteCoreAsync( + entry, + async requestCancellationToken => + (await requestFactory(requestCancellationToken).ConfigureAwait(false))!), + LazyThreadSafetyMode.ExecutionAndPublication); + + return entry; + } + + private async Task ExecuteCoreAsync( + RequestEntry entry, + Func> requestFactory) + { + try + { + return await requestFactory(entry.CancellationSource.Token).ConfigureAwait(false); + } + finally + { + CompleteRequest(entry); + } + } + + private async Task WaitAsync( + RequestEntry entry, + Func cancellationResultFactory, + CancellationToken cancellationToken) + { + try + { + var requestTask = entry.Request.Value; + + if (!cancellationToken.CanBeCanceled) + return (TResult)await requestTask.ConfigureAwait(false); + + var cancellationCompletion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + using (cancellationToken.Register( + state => ((TaskCompletionSource)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(); + } + + /// + /// Remove an entry when it is still the current entry for its key. + /// Must only be called while holding . + /// + private void RemoveEntry(RequestEntry entry) + { + if (_requests.TryGetValue(entry.Key, out var current) + && ReferenceEquals(current, entry)) + { + _requests.Remove(entry.Key); + } + } + } + + /// + /// Key used to identify identical requests for coalescing + /// + /// Request method + /// Request URL + /// Request parameter string + /// Response type + public sealed record RequestCoalescingKey( + string Method, + string Url, + string? Parameters, + Type ResponseType); +} \ No newline at end of file diff --git a/CryptoExchange.Net/Objects/RequestDefinition.cs b/CryptoExchange.Net/Objects/RequestDefinition.cs index 3a782274..dbf02107 100644 --- a/CryptoExchange.Net/Objects/RequestDefinition.cs +++ b/CryptoExchange.Net/Objects/RequestDefinition.cs @@ -101,6 +101,11 @@ namespace CryptoExchange.Net.Objects } } + /// + /// Whether concurrent identical requests should be prevented from sharing the same request/response + /// + public bool PreventRequestCoalescing { get; set; } + /// /// ctor /// diff --git a/CryptoExchange.Net/Objects/RequestDefinitionCache.cs b/CryptoExchange.Net/Objects/RequestDefinitionCache.cs index 0a6477ad..c33a72db 100644 --- a/CryptoExchange.Net/Objects/RequestDefinitionCache.cs +++ b/CryptoExchange.Net/Objects/RequestDefinitionCache.cs @@ -53,6 +53,7 @@ namespace CryptoExchange.Net.Objects /// Try parse the response even when status is not success /// Force trailing `/` /// Optional request identifier override + /// Whether concurrent identical requests should be prevented from sharing the same request/response /// 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); } diff --git a/CryptoExchange.Net/Objects/Sockets/DataEvent.cs b/CryptoExchange.Net/Objects/Sockets/DataEvent.cs index 15caa7a7..7393e61c 100644 --- a/CryptoExchange.Net/Objects/Sockets/DataEvent.cs +++ b/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 }; } diff --git a/CryptoExchange.Net/SharedApis/Enums/SharedOrderBookSubscriptionType.cs b/CryptoExchange.Net/SharedApis/Enums/SharedOrderBookSubscriptionType.cs new file mode 100644 index 00000000..b4c8b4be --- /dev/null +++ b/CryptoExchange.Net/SharedApis/Enums/SharedOrderBookSubscriptionType.cs @@ -0,0 +1,21 @@ +using System; +using System.Collections.Generic; +using System.Text; + +namespace CryptoExchange.Net.SharedApis +{ + /// + /// The type of updates the incremental order book subscription produces + /// + public enum SharedOrderBookSubscriptionType + { + /// + /// Subscription produces an initial snapshot update, followed by incremental change updates + /// + SnapshotThenIncremental, + /// + /// Subscription produces only incremental change updates + /// + OnlyIncremental + } +} diff --git a/CryptoExchange.Net/SharedApis/Interfaces/V2/Endpoints/Futures/UserTrades/IGetFuturesUserTradeHistoryEndpoint.cs b/CryptoExchange.Net/SharedApis/Interfaces/V2/Endpoints/Futures/UserTrades/IGetFuturesUserTradeHistoryEndpoint.cs index b7f59942..6e37fdec 100644 --- a/CryptoExchange.Net/SharedApis/Interfaces/V2/Endpoints/Futures/UserTrades/IGetFuturesUserTradeHistoryEndpoint.cs +++ b/CryptoExchange.Net/SharedApis/Interfaces/V2/Endpoints/Futures/UserTrades/IGetFuturesUserTradeHistoryEndpoint.cs @@ -25,6 +25,6 @@ namespace CryptoExchange.Net.SharedApis /// Request info /// The pagination request from the previous request result property to continue pagination /// Cancellation token - Task> GetFuturesUserTradesAsync(GetUserTradesRequest request, PageRequest? nextPageToken = null, CancellationToken ct = default); + Task> GetFuturesUserTradeHistoryAsync(GetUserTradesRequest request, PageRequest? nextPageToken = null, CancellationToken ct = default); } } diff --git a/CryptoExchange.Net/SharedApis/Interfaces/V2/Operations/Futures/ISubscribeFuturesOrdersOperation.cs b/CryptoExchange.Net/SharedApis/Interfaces/V2/Operations/Futures/ISubscribeFuturesOrdersOperation.cs index f180b7da..d3eaec25 100644 --- a/CryptoExchange.Net/SharedApis/Interfaces/V2/Operations/Futures/ISubscribeFuturesOrdersOperation.cs +++ b/CryptoExchange.Net/SharedApis/Interfaces/V2/Operations/Futures/ISubscribeFuturesOrdersOperation.cs @@ -23,6 +23,6 @@ namespace CryptoExchange.Net.SharedApis /// Update handler /// Cancellation token, can be used to stop the updates /// - Task> SubscribeToFuturesOrderUpdatesAsync(SubscribeFuturesOrderRequest request, Action> handler, CancellationToken ct = default); + Task> SubscribeToFuturesOrderUpdatesAsync(SubscribeFuturesOrderRequest request, Action> handler, CancellationToken ct = default); } } diff --git a/CryptoExchange.Net/SharedApis/Interfaces/V2/Operations/ISubscribeIncrementalOrderBookOperation.cs b/CryptoExchange.Net/SharedApis/Interfaces/V2/Operations/ISubscribeIncrementalOrderBookOperation.cs new file mode 100644 index 00000000..13803377 --- /dev/null +++ b/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 +{ + /// + /// Client for subscribing to incremental order book updates for a symbol + /// + public interface ISubscribeIncrementalOrderBookOperation : ISharedSubscription + { + /// + /// Order book subscription options + /// + SubscribeOrderBookOptions SubscribeOrderBookOptions { get; } + + /// + /// Subscribe to incremental order book updates for a symbol + /// + /// Request info + /// Update handler + /// Cancellation token, can be used to stop the updates + /// + Task> SubscribeToOrderBookUpdatesAsync(SubscribeOrderBookRequest request, Action> handler, CancellationToken ct = default); + } +} diff --git a/CryptoExchange.Net/SharedApis/Interfaces/V2/Operations/Spot/ISubscribeSpotOrdersOperation.cs b/CryptoExchange.Net/SharedApis/Interfaces/V2/Operations/Spot/ISubscribeSpotOrdersOperation.cs index 5f29002d..28ca2bff 100644 --- a/CryptoExchange.Net/SharedApis/Interfaces/V2/Operations/Spot/ISubscribeSpotOrdersOperation.cs +++ b/CryptoExchange.Net/SharedApis/Interfaces/V2/Operations/Spot/ISubscribeSpotOrdersOperation.cs @@ -23,6 +23,6 @@ namespace CryptoExchange.Net.SharedApis /// Update handler /// Cancellation token, can be used to stop the updates /// - Task> SubscribeToSpotOrderUpdatesAsync(SubscribeSpotOrderRequest request, Action> handler, CancellationToken ct = default); + Task> SubscribeToSpotOrderUpdatesAsync(SubscribeSpotOrderRequest request, Action> handler, CancellationToken ct = default); } } diff --git a/CryptoExchange.Net/SharedApis/Models/Options/Endpoints/GetFuturesUserTradeHistoryOptions.cs b/CryptoExchange.Net/SharedApis/Models/Options/Endpoints/GetFuturesUserTradeHistoryOptions.cs index 653b5d7a..05e4d3a8 100644 --- a/CryptoExchange.Net/SharedApis/Models/Options/Endpoints/GetFuturesUserTradeHistoryOptions.cs +++ b/CryptoExchange.Net/SharedApis/Models/Options/Endpoints/GetFuturesUserTradeHistoryOptions.cs @@ -16,7 +16,7 @@ namespace CryptoExchange.Net.SharedApis /// ctor /// 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)) { } diff --git a/CryptoExchange.Net/SharedApis/Models/Options/Operations/SubscribeIncrementalOrderBookOptions.cs b/CryptoExchange.Net/SharedApis/Models/Options/Operations/SubscribeIncrementalOrderBookOptions.cs new file mode 100644 index 00000000..2b10ada4 --- /dev/null +++ b/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 +{ + /// + /// Options for subscribing to order book snapshot updates + /// + public class SubscribeIncrementalOrderBookOptions : CapabilityOptions + { + /// + public override string Description => "Subscribe to incremental order book updates for a symbol"; + + /// + /// Order book depths supported for updates + /// + public int[] SupportedLimits { get; } + + /// + /// The type of updates the subscription produces + /// + public SharedOrderBookSubscriptionType UpdateType { get; } + + /// + /// ctor + /// + public SubscribeIncrementalOrderBookOptions(string exchange, bool needsAuthentication, int[] limits, SharedOrderBookSubscriptionType updateType) + : base(exchange, needsAuthentication, nameof(ISubscribeIncrementalOrderBookOperation.SubscribeToOrderBookUpdatesAsync)) + { + SupportedLimits = limits; + UpdateType = updateType; + } + + /// + /// Validate a request + /// + 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); + } + } +} diff --git a/CryptoExchange.Net/SharedApis/ResponseModels/SharedFuturesOrder.cs b/CryptoExchange.Net/SharedApis/ResponseModels/SharedFuturesOrder.cs index 49daa0e5..4dc65791 100644 --- a/CryptoExchange.Net/SharedApis/ResponseModels/SharedFuturesOrder.cs +++ b/CryptoExchange.Net/SharedApis/ResponseModels/SharedFuturesOrder.cs @@ -73,10 +73,12 @@ namespace CryptoExchange.Net.SharedApis /// /// Asset the fee is in /// + [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; } /// /// Fee paid /// + [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; } /// /// Leverage @@ -94,6 +96,7 @@ namespace CryptoExchange.Net.SharedApis /// /// Last trade info, only available for websocket order updates if the API provides this data in the update /// + [Obsolete("Use SharedFuturesOrderUpdate.LastTrade instead, LastTrade is never filled on non websocket updates")] public SharedUserTrade? LastTrade { get; set; } /// diff --git a/CryptoExchange.Net/SharedApis/ResponseModels/SharedFuturesOrderUpdate.cs b/CryptoExchange.Net/SharedApis/ResponseModels/SharedFuturesOrderUpdate.cs new file mode 100644 index 00000000..05527ce7 --- /dev/null +++ b/CryptoExchange.Net/SharedApis/ResponseModels/SharedFuturesOrderUpdate.cs @@ -0,0 +1,38 @@ +using System; +using System.Diagnostics; + +namespace CryptoExchange.Net.SharedApis +{ + /// + /// Futures order info + /// + [DebuggerDisplay("{DebugView,nq}")] + public record SharedFuturesOrderUpdate : SharedFuturesOrder + { + /// + /// The info on the executed trade for this update + /// + 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 + } + + /// + /// ctor + /// + public SharedFuturesOrderUpdate( + SharedSymbol? sharedSymbol, + string symbol, + string orderId, + SharedOrderType orderType, + SharedOrderSide orderSide, + SharedOrderStatus orderStatus, + DateTime? createTime) + : base(sharedSymbol, symbol, orderId, orderType, orderSide, orderStatus, createTime) + { + } + } +} diff --git a/CryptoExchange.Net/SharedApis/ResponseModels/SharedOrderBook.cs b/CryptoExchange.Net/SharedApis/ResponseModels/SharedOrderBook.cs index 37e517b2..a631aa32 100644 --- a/CryptoExchange.Net/SharedApis/ResponseModels/SharedOrderBook.cs +++ b/CryptoExchange.Net/SharedApis/ResponseModels/SharedOrderBook.cs @@ -12,6 +12,10 @@ namespace CryptoExchange.Net.SharedApis /// public SharedQuantityType QuantityType { get; set; } /// + /// The sequence number this snapshot or update is at + /// + public long? SequenceNumber { get; set; } + /// /// Asks list /// public ISymbolOrderBookEntry[] Asks { get; set; } @@ -23,9 +27,10 @@ namespace CryptoExchange.Net.SharedApis /// /// ctor /// - 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; } diff --git a/CryptoExchange.Net/SharedApis/ResponseModels/SharedSpotOrder.cs b/CryptoExchange.Net/SharedApis/ResponseModels/SharedSpotOrder.cs index 0535922f..d4e8dfbc 100644 --- a/CryptoExchange.Net/SharedApis/ResponseModels/SharedSpotOrder.cs +++ b/CryptoExchange.Net/SharedApis/ResponseModels/SharedSpotOrder.cs @@ -65,10 +65,12 @@ namespace CryptoExchange.Net.SharedApis /// /// Asset the fee is in /// + [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; } /// /// Fee paid for the order /// + [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; } /// /// Timestamp the order was created @@ -79,8 +81,9 @@ namespace CryptoExchange.Net.SharedApis /// public DateTime? UpdateTime { get; set; } /// - /// 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 /// + [Obsolete("Use SharedSpotOrderUpdate.LastTrade instead, LastTrade is never filled on non websocket updates")] public SharedUserTrade? LastTrade { get; set; } /// diff --git a/CryptoExchange.Net/SharedApis/ResponseModels/SharedSpotOrderUpdate.cs b/CryptoExchange.Net/SharedApis/ResponseModels/SharedSpotOrderUpdate.cs new file mode 100644 index 00000000..022313ca --- /dev/null +++ b/CryptoExchange.Net/SharedApis/ResponseModels/SharedSpotOrderUpdate.cs @@ -0,0 +1,38 @@ +using System; +using System.Diagnostics; + +namespace CryptoExchange.Net.SharedApis +{ + /// + /// Spot order info + /// + [DebuggerDisplay("{DebugView,nq}")] + public record SharedSpotOrderUpdate : SharedSpotOrder + { + /// + /// The info on the executed trade for this update + /// + 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 + } + + /// + /// ctor + /// + public SharedSpotOrderUpdate( + SharedSymbol? sharedSymbol, + string symbol, + string orderId, + SharedOrderType orderType, + SharedOrderSide orderSide, + SharedOrderStatus orderStatus, + DateTime? createTime) + : base(sharedSymbol, symbol, orderId, orderType, orderSide, orderStatus, createTime) + { + } + } +} diff --git a/CryptoExchange.Net/Trackers/UserData/ItemTrackers/FuturesOrderTracker.cs b/CryptoExchange.Net/Trackers/UserData/ItemTrackers/FuturesOrderTracker.cs index 85116a23..92da259e 100644 --- a/CryptoExchange.Net/Trackers/UserData/ItemTrackers/FuturesOrderTracker.cs +++ b/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().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(_exchange, default!, default)); - return ExchangeHelpers.ProcessQueuedAsync( + return ExchangeHelpers.ProcessQueuedAsync( async handler => await _socketClient.SubscribeToFuturesOrderUpdatesAsync(new SubscribeFuturesOrderRequest(exchangeParameters: _exchangeParameters), handler, ct: _cts!.Token).ConfigureAwait(false), x => HandleUpdateAsync(UpdateSource.Push, x.Data))!; } diff --git a/CryptoExchange.Net/Trackers/UserData/ItemTrackers/FuturesUserTradeTracker.cs b/CryptoExchange.Net/Trackers/UserData/ItemTrackers/FuturesUserTradeTracker.cs index 1aa9fdb9..bfe0bdf1 100644 --- a/CryptoExchange.Net/Trackers/UserData/ItemTrackers/FuturesUserTradeTracker.cs +++ b/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; diff --git a/CryptoExchange.Net/Trackers/UserData/ItemTrackers/SpotOrderTracker.cs b/CryptoExchange.Net/Trackers/UserData/ItemTrackers/SpotOrderTracker.cs index 7f791635..ff5aefb8 100644 --- a/CryptoExchange.Net/Trackers/UserData/ItemTrackers/SpotOrderTracker.cs +++ b/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().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(_exchange!, default!, default)); - return ExchangeHelpers.ProcessQueuedAsync( + return ExchangeHelpers.ProcessQueuedAsync( async handler => await _socketClient.SubscribeToSpotOrderUpdatesAsync(new SubscribeSpotOrderRequest(exchangeParameters: _exchangeParameters), handler, ct: _cts!.Token).ConfigureAwait(false), x => HandleUpdateAsync(UpdateSource.Push, x.Data))!; }