1
0
mirror of https://github.com/JKorf/CryptoExchange.Net.git synced 2026-08-12 17:03:10 +00:00

Compare commits

...

27 Commits

Author SHA1 Message Date
JKorf ea9375d582 Updated version 2022-06-12 15:36:04 +02:00
JKorf 2cf3c93e5e Cleanup 2022-06-12 15:35:35 +02:00
JKorf ca888d8e41 Updated version 2022-06-12 15:31:03 +02:00
JKorf 2040b1c175 Fixed proxy setting not used on reconnecting socket 2022-06-12 15:26:11 +02:00
JKorf d451c18821 No longer waiting for timesyncing to complete when it's not the first request 2022-06-12 15:21:22 +02:00
JKorf c13dfa4461 Updated socket reconnection 2022-06-12 15:10:10 +02:00
JKorf c2080ef75f Made MaxSocketConnections a setting, added support for changing log settings after creating client 2022-06-11 13:31:39 +02:00
Jan Korf 6b252e8024 Update TestSocket.cs 2022-05-24 22:36:55 +02:00
Jan Korf d06bd5f176 Updated version 2022-05-24 18:56:37 +02:00
Jan Korf d55fc8da65 Merge pull request #144 from tamaw/fix/missing-port-on-baseuri
Fixed: copying the port number when using a custom BaseAddress
2022-05-24 15:14:38 +02:00
Tama Waddell 01184f2c5d Added port to the other overloaded method 2022-05-24 21:13:08 +10:00
Jan Korf cadc93c2f0 Merge pull request #143 from andriibratanin/bugfix/fix-nuget-discovery
Fix NuGet packages discovery for some IDEs
2022-05-24 10:21:21 +02:00
Tama Waddell 2600a51461 Included copying the port when using SetParameters 2022-05-24 15:16:06 +10:00
Andrii Bratanin 9e6a86ba8b Fix wrong case in csproj files of tests projects #142 2022-05-24 01:31:25 +03:00
Jan Korf c4430d63fa Added KeepAliveInterval setting for socket connections 2022-05-23 22:05:04 +02:00
Jan Korf f3e1cfef33 Updated version 2022-05-22 15:51:48 +02:00
Jan Korf cc3053719c Make socket ConnectionLost run in a separate task to prevent issue with long running/exceptions in the handler 2022-05-22 15:46:30 +02:00
Jan Korf cd6907e601 Merge branch 'master' of https://github.com/jkorf/CryptoExchange.Net 2022-05-22 14:35:07 +02:00
Jan Korf 8fe00693bd Merge pull request #141 from nathan-datusarator/master
Add checks for Disposed
2022-05-22 11:41:55 +02:00
Jan Korf fb90d1e015 Fixed exception when disposing client in reconnecting state 2022-05-22 11:35:39 +02:00
Jan Korf 4b44861e43 Added additional cases for no null/default handling in DateTimeConverter 2022-05-22 11:35:17 +02:00
Jan Korf e42ca4ab5a Update FAQ.md 2022-05-21 10:17:59 +02:00
Nathan Pfluger 5b97f6dd67 Move Subscription Events into non-lambda so they can be removed on StopAsync 2022-05-12 10:00:44 -07:00
Nathan Pfluger a9813ecb0a Add checks for Disposed 2022-05-12 09:05:27 -07:00
Jan Korf c7069a4049 Updated version 2022-05-08 16:28:17 +02:00
Jan Korf 5683ae0b3c Small fix when closing socket 2022-05-08 16:25:45 +02:00
Jan Korf 1c8cf5ac98 Updated timestamp calculation to include latency 2022-05-08 15:23:47 +02:00
19 changed files with 365 additions and 99 deletions
@@ -6,10 +6,10 @@
</PropertyGroup>
<ItemGroup>
<packagereference Include="Microsoft.NET.Test.Sdk" Version="17.1.0-preview-20211130-02"></packagereference>
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.1.0-preview-20211130-02"></PackageReference>
<PackageReference Include="Moq" Version="4.16.1" />
<packagereference Include="NUnit" Version="3.13.2"></packagereference>
<packagereference Include="NUnit3TestAdapter" Version="4.2.0"></packagereference>
<PackageReference Include="NUnit" Version="3.13.2"></PackageReference>
<PackageReference Include="NUnit3TestAdapter" Version="4.2.0"></PackageReference>
</ItemGroup>
<ItemGroup>
@@ -135,7 +135,7 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
throw new NotImplementedException();
}
protected override TimeSyncInfo GetTimeSyncInfo()
public override TimeSyncInfo GetTimeSyncInfo()
{
throw new NotImplementedException();
}
@@ -161,7 +161,7 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
throw new NotImplementedException();
}
protected override TimeSyncInfo GetTimeSyncInfo()
public override TimeSyncInfo GetTimeSyncInfo()
{
throw new NotImplementedException();
}
@@ -40,6 +40,8 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
public Uri Uri => new Uri("");
public TimeSpan KeepAliveInterval { get; set; }
public static int lastId = 0;
public static object lastIdLock = new object();
+11
View File
@@ -63,6 +63,7 @@ namespace CryptoExchange.Net
log = new Log(name);
log.UpdateWriters(options.LogWriters);
log.Level = options.LogLevel;
options.OnLoggingChanged += HandleLogConfigChange;
ClientOptions = options;
@@ -282,12 +283,22 @@ namespace CryptoExchange.Net
}
}
/// <summary>
/// Handle a change in the client options log config
/// </summary>
private void HandleLogConfigChange()
{
log.UpdateWriters(ClientOptions.LogWriters);
log.Level = ClientOptions.LogLevel;
}
/// <summary>
/// Dispose
/// </summary>
public virtual void Dispose()
{
log.Write(LogLevel.Debug, "Disposing client");
ClientOptions.OnLoggingChanged -= HandleLogConfigChange;
foreach (var client in ApiClients)
client.Dispose();
}
+10 -4
View File
@@ -172,11 +172,17 @@ namespace CryptoExchange.Net
if (signed)
{
var syncTimeResult = await apiClient.SyncTimeAsync().ConfigureAwait(false);
if (!syncTimeResult)
var syncTask = apiClient.SyncTimeAsync();
var timeSyncInfo = apiClient.GetTimeSyncInfo();
if (timeSyncInfo.TimeSyncState.LastSyncTime == default)
{
log.Write(LogLevel.Debug, $"[{requestId}] Failed to sync time, aborting request: " + syncTimeResult.Error);
return syncTimeResult.As<IRequest>(default);
// Initially with first request we'll need to wait for the time syncing, if it's not the first request we can just continue
var syncTimeResult = await syncTask.ConfigureAwait(false);
if (!syncTimeResult)
{
log.Write(LogLevel.Debug, $"[{requestId}] Failed to sync time, aborting request: " + syncTimeResult.Error);
return syncTimeResult.As<IRequest>(default);
}
}
}
+50 -25
View File
@@ -36,9 +36,9 @@ namespace CryptoExchange.Net
/// </summary>
protected internal readonly SemaphoreSlim semaphoreSlim = new(1);
/// <summary>
/// The max amount of concurrent socket connections
/// Keep alive interval for websocket connection
/// </summary>
protected int MaxSocketConnections { get; set; } = 9999;
protected TimeSpan KeepAliveInterval { get; set; } = TimeSpan.FromSeconds(10);
/// <summary>
/// Delegate used for processing byte data received from socket connections before it is processed by handlers
/// </summary>
@@ -92,6 +92,20 @@ namespace CryptoExchange.Net
}
}
/// <inheritdoc />
public int CurrentConnections => socketConnections.Count;
/// <inheritdoc />
public int CurrentSubscriptions
{
get
{
if (!socketConnections.Any())
return 0;
return socketConnections.Sum(s => s.Value.SubscriptionCount);
}
}
/// <summary>
/// Client options
/// </summary>
@@ -164,7 +178,7 @@ namespace CryptoExchange.Net
return new CallResult<UpdateSubscription>(new InvalidOperationError("Client disposed, can't subscribe"));
SocketConnection socketConnection;
SocketSubscription subscription;
SocketSubscription? subscription;
var released = false;
// Wait for a semaphore here, so we only connect 1 socket at a time.
// This is necessary for being able to see if connections can be combined
@@ -179,23 +193,34 @@ namespace CryptoExchange.Net
try
{
// Get a new or existing socket connection
socketConnection = GetSocketConnection(apiClient, url, authenticated);
// Add a subscription on the socket connection
subscription = AddSubscription(request, identifier, true, socketConnection, dataHandler);
if (ClientOptions.SocketSubscriptionsCombineTarget == 1)
while (true)
{
// Only 1 subscription per connection, so no need to wait for connection since a new subscription will create a new connection anyway
semaphoreSlim.Release();
released = true;
// Get a new or existing socket connection
socketConnection = GetSocketConnection(apiClient, url, authenticated);
// Add a subscription on the socket connection
subscription = AddSubscription(request, identifier, true, socketConnection, dataHandler);
if (subscription == null)
{
log.Write(LogLevel.Trace, $"Socket {socketConnection.SocketId} failed to add subscription, retrying on different connection");
continue;
}
if (ClientOptions.SocketSubscriptionsCombineTarget == 1)
{
// Only 1 subscription per connection, so no need to wait for connection since a new subscription will create a new connection anyway
semaphoreSlim.Release();
released = true;
}
var needsConnecting = !socketConnection.Connected;
var connectResult = await ConnectIfNeededAsync(socketConnection, authenticated).ConfigureAwait(false);
if (!connectResult)
return new CallResult<UpdateSubscription>(connectResult.Error!);
break;
}
var needsConnecting = !socketConnection.Connected;
var connectResult = await ConnectIfNeededAsync(socketConnection, authenticated).ConfigureAwait(false);
if (!connectResult)
return new CallResult<UpdateSubscription>(connectResult.Error!);
}
finally
{
@@ -443,9 +468,6 @@ namespace CryptoExchange.Net
/// <param name="message"></param>
/// <returns></returns>
protected internal virtual JToken ProcessTokenData(JToken message)
{
return message;
}
@@ -460,7 +482,7 @@ namespace CryptoExchange.Net
/// <param name="connection">The socket connection the handler is on</param>
/// <param name="dataHandler">The handler of the data received</param>
/// <returns></returns>
protected virtual SocketSubscription AddSubscription<T>(object? request, string? identifier, bool userSubscription, SocketConnection connection, Action<DataEvent<T>> dataHandler)
protected virtual SocketSubscription? AddSubscription<T>(object? request, string? identifier, bool userSubscription, SocketConnection connection, Action<DataEvent<T>> dataHandler)
{
void InternalHandler(MessageEvent messageEvent)
{
@@ -484,7 +506,8 @@ namespace CryptoExchange.Net
var subscription = request == null
? SocketSubscription.CreateForIdentifier(NextId(), identifier!, userSubscription, InternalHandler)
: SocketSubscription.CreateForRequest(NextId(), request, userSubscription, InternalHandler);
connection.AddSubscription(subscription);
if (!connection.AddSubscription(subscription))
return null;
return subscription;
}
@@ -510,13 +533,14 @@ namespace CryptoExchange.Net
/// <returns></returns>
protected virtual SocketConnection GetSocketConnection(SocketApiClient apiClient, string address, bool authenticated)
{
var socketResult = socketConnections.Where(s => s.Value.Uri.ToString().TrimEnd('/') == address.TrimEnd('/')
var socketResult = socketConnections.Where(s => (s.Value.Status == SocketConnection.SocketStatus.None || s.Value.Status == SocketConnection.SocketStatus.Connected)
&& s.Value.Uri.ToString().TrimEnd('/') == address.TrimEnd('/')
&& (s.Value.ApiClient.GetType() == apiClient.GetType())
&& (s.Value.Authenticated == authenticated || !authenticated) && s.Value.Connected).OrderBy(s => s.Value.SubscriptionCount).FirstOrDefault();
var result = socketResult.Equals(default(KeyValuePair<int, SocketConnection>)) ? null : socketResult.Value;
if (result != null)
{
if (result.SubscriptionCount < ClientOptions.SocketSubscriptionsCombineTarget || (socketConnections.Count >= MaxSocketConnections && socketConnections.All(s => s.Value.SubscriptionCount >= ClientOptions.SocketSubscriptionsCombineTarget)))
if (result.SubscriptionCount < ClientOptions.SocketSubscriptionsCombineTarget || (socketConnections.Count >= ClientOptions.MaxSocketConnections && socketConnections.All(s => s.Value.SubscriptionCount >= ClientOptions.SocketSubscriptionsCombineTarget)))
{
// Use existing socket if it has less than target connections OR it has the least connections and we can't make new
return result;
@@ -574,6 +598,7 @@ namespace CryptoExchange.Net
if (ClientOptions.Proxy != null)
socket.SetProxy(ClientOptions.Proxy);
socket.KeepAliveInterval = KeepAliveInterval;
socket.Timeout = ClientOptions.SocketNoDataTimeout;
socket.DataInterpreterBytes = dataInterpreterBytes;
socket.DataInterpreterString = dataInterpreterString;
+2 -2
View File
@@ -18,7 +18,7 @@ namespace CryptoExchange.Net
/// Get time sync info for an API client
/// </summary>
/// <returns></returns>
protected abstract TimeSyncInfo GetTimeSyncInfo();
public abstract TimeSyncInfo GetTimeSyncInfo();
/// <summary>
/// Get time offset for an API client
@@ -92,7 +92,7 @@ namespace CryptoExchange.Net
}
// Calculate time offset between local and server
var offset = result.Data - localTime;
var offset = result.Data - (localTime.AddMilliseconds(result.ResponseTime!.Value.TotalMilliseconds / 2));
timeSyncParams.UpdateTimeOffset(offset);
timeSyncParams.TimeSyncState.Semaphore.Release();
}
@@ -31,7 +31,7 @@ namespace CryptoExchange.Net.Converters
if(reader.TokenType is JsonToken.Integer)
{
var longValue = (long)reader.Value;
if (longValue == 0)
if (longValue == 0 || longValue == -1)
return objectType == typeof(DateTime) ? default(DateTime): null;
if (longValue < 19999999999)
return ConvertFromSeconds(longValue);
@@ -45,6 +45,9 @@ namespace CryptoExchange.Net.Converters
else if (reader.TokenType is JsonToken.Float)
{
var doubleValue = (double)reader.Value;
if (doubleValue == 0 || doubleValue == -1)
return objectType == typeof(DateTime) ? default(DateTime) : null;
if (doubleValue < 19999999999)
return ConvertFromSeconds(doubleValue);
@@ -56,6 +59,9 @@ namespace CryptoExchange.Net.Converters
if (string.IsNullOrWhiteSpace(stringValue))
return null;
if (string.IsNullOrWhiteSpace(stringValue) || stringValue == "0" || stringValue == "-1")
return objectType == typeof(DateTime) ? default(DateTime) : null;
if (stringValue.Length == 8)
{
// Parse 20211103 format
+4 -4
View File
@@ -6,16 +6,16 @@
<PackageId>CryptoExchange.Net</PackageId>
<Authors>JKorf</Authors>
<Description>A base package for implementing cryptocurrency API's</Description>
<PackageVersion>5.1.8</PackageVersion>
<AssemblyVersion>5.1.8</AssemblyVersion>
<FileVersion>5.1.8</FileVersion>
<PackageVersion>5.1.12</PackageVersion>
<AssemblyVersion>5.1.12</AssemblyVersion>
<FileVersion>5.1.12</FileVersion>
<PackageRequireLicenseAcceptance>false</PackageRequireLicenseAcceptance>
<RepositoryType>git</RepositoryType>
<RepositoryUrl>https://github.com/JKorf/CryptoExchange.Net.git</RepositoryUrl>
<PackageProjectUrl>https://github.com/JKorf/CryptoExchange.Net</PackageProjectUrl>
<NeutralLanguage>en</NeutralLanguage>
<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
<PackageReleaseNotes>5.1.8 - Cleanup socket code, fixed an issue which could cause connections to never reconnect when connection was lost, Added support for sending requests which expect an empty response, Fixed issue with the DateTimeConverter date interpretation</PackageReleaseNotes>
<PackageReleaseNotes>5.1.12 - Changed time sync so requests no longer wait for it to complete unless it's the first time, Made log client options changable after client creation, Fixed proxy setting not used when reconnecting socket, Changed MaxSocketConnections to a client options, Updated socket reconnection logic</PackageReleaseNotes>
<Nullable>enable</Nullable>
<LangVersion>9.0</LangVersion>
<PackageLicenseExpression>MIT</PackageLicenseExpression>
+2
View File
@@ -426,6 +426,7 @@ namespace CryptoExchange.Net
var uriBuilder = new UriBuilder();
uriBuilder.Scheme = baseUri.Scheme;
uriBuilder.Host = baseUri.Host;
uriBuilder.Port = baseUri.Port;
uriBuilder.Path = baseUri.AbsolutePath;
var httpValueCollection = HttpUtility.ParseQueryString(string.Empty);
foreach (var parameter in parameters)
@@ -454,6 +455,7 @@ namespace CryptoExchange.Net
var uriBuilder = new UriBuilder();
uriBuilder.Scheme = baseUri.Scheme;
uriBuilder.Host = baseUri.Host;
uriBuilder.Port = baseUri.Port;
uriBuilder.Path = baseUri.AbsolutePath;
var httpValueCollection = HttpUtility.ParseQueryString(string.Empty);
foreach (var parameter in parameters)
@@ -27,6 +27,16 @@ namespace CryptoExchange.Net.Interfaces
/// </summary>
public double IncomingKbps { get; }
/// <summary>
/// The current amount of connections to the API from this client. A connection can have multiple subscriptions.
/// </summary>
public int CurrentConnections { get; }
/// <summary>
/// The current amount of subscriptions running from the client
/// </summary>
public int CurrentSubscriptions { get; }
/// <summary>
/// Unsubscribe from a stream using the subscription id received when starting the subscription
/// </summary>
@@ -77,6 +77,10 @@ namespace CryptoExchange.Net.Interfaces
/// </summary>
TimeSpan Timeout { get; set; }
/// <summary>
/// The interval at which to send a ping frame to the server
/// </summary>
TimeSpan KeepAliveInterval { get; set; }
/// <summary>
/// Set a proxy to use when connecting
/// </summary>
/// <param name="proxy"></param>
+15 -9
View File
@@ -26,6 +26,8 @@ namespace CryptoExchange.Net.Logging
/// </summary>
public string ClientName { get; set; }
private readonly object _lock = new object();
/// <summary>
/// ctor
/// </summary>
@@ -42,7 +44,8 @@ namespace CryptoExchange.Net.Logging
/// <param name="textWriters"></param>
public void UpdateWriters(List<ILogger> textWriters)
{
writers = textWriters;
lock (_lock)
writers = textWriters;
}
/// <summary>
@@ -56,16 +59,19 @@ namespace CryptoExchange.Net.Logging
return;
var logMessage = $"{ClientName,-10} | {message}";
foreach (var writer in writers.ToList())
lock (_lock)
{
try
foreach (var writer in writers)
{
writer.Log(logLevel, logMessage);
}
catch (Exception e)
{
// Can't write to the logging so where else to output..
Trace.WriteLine($"{DateTime.Now:yyyy/MM/dd HH:mm:ss:fff} | Warning | Failed to write log to writer {writer.GetType()}: " + e.ToLogString());
try
{
writer.Log(logLevel, logMessage);
}
catch (Exception e)
{
// Can't write to the logging so where else to output..
Trace.WriteLine($"{DateTime.Now:yyyy/MM/dd HH:mm:ss:fff} | Warning | Failed to write log to writer {writer.GetType()}: " + e.ToLogString());
}
}
}
}
+29 -3
View File
@@ -14,15 +14,35 @@ namespace CryptoExchange.Net.Objects
/// </summary>
public class BaseOptions
{
internal event Action? OnLoggingChanged;
private LogLevel _logLevel = LogLevel.Information;
/// <summary>
/// The minimum log level to output
/// </summary>
public LogLevel LogLevel { get; set; } = LogLevel.Information;
public LogLevel LogLevel
{
get => _logLevel;
set
{
_logLevel = value;
OnLoggingChanged?.Invoke();
}
}
private List<ILogger> _logWriters = new List<ILogger> { new DebugLogger() };
/// <summary>
/// The log writers
/// </summary>
public List<ILogger> LogWriters { get; set; } = new List<ILogger> { new DebugLogger() };
public List<ILogger> LogWriters
{
get => _logWriters;
set
{
_logWriters = value;
OnLoggingChanged?.Invoke();
}
}
/// <summary>
/// If true, the CallResult and DataEvent objects will also include the originally received json data in the OriginalData property
@@ -189,6 +209,11 @@ namespace CryptoExchange.Net.Objects
/// </summary>
public int? SocketSubscriptionsCombineTarget { get; set; }
/// <summary>
/// The max amount of connections to make to the server. Can be used for API's which only allow a certain number of connections. Changing this to a high value might cause issues.
/// </summary>
public int? MaxSocketConnections { get; set; }
/// <summary>
/// ctor
/// </summary>
@@ -213,12 +238,13 @@ namespace CryptoExchange.Net.Objects
SocketResponseTimeout = baseOptions.SocketResponseTimeout;
SocketNoDataTimeout = baseOptions.SocketNoDataTimeout;
SocketSubscriptionsCombineTarget = baseOptions.SocketSubscriptionsCombineTarget;
MaxSocketConnections = baseOptions.MaxSocketConnections;
}
/// <inheritdoc />
public override string ToString()
{
return $"{base.ToString()}, AutoReconnect: {AutoReconnect}, ReconnectInterval: {ReconnectInterval}, MaxReconnectTries: {MaxReconnectTries}, MaxResubscribeTries: {MaxResubscribeTries}, MaxConcurrentResubscriptionsPerSocket: {MaxConcurrentResubscriptionsPerSocket}, SocketResponseTimeout: {SocketResponseTimeout:c}, SocketNoDataTimeout: {SocketNoDataTimeout}, SocketSubscriptionsCombineTarget: {SocketSubscriptionsCombineTarget}";
return $"{base.ToString()}, AutoReconnect: {AutoReconnect}, ReconnectInterval: {ReconnectInterval}, MaxReconnectTries: {MaxReconnectTries}, MaxResubscribeTries: {MaxResubscribeTries}, MaxConcurrentResubscriptionsPerSocket: {MaxConcurrentResubscriptionsPerSocket}, SocketResponseTimeout: {SocketResponseTimeout:c}, SocketNoDataTimeout: {SocketNoDataTimeout}, SocketSubscriptionsCombineTarget: {SocketSubscriptionsCombineTarget}, MaxSocketConnections: {MaxSocketConnections}";
}
}
+28 -16
View File
@@ -258,24 +258,32 @@ namespace CryptoExchange.Net.OrderBook
}
_subscription = startResult.Data;
_subscription.ConnectionLost += () =>
{
log.Write(LogLevel.Warning, $"{Id} order book {Symbol} connection lost");
Status = OrderBookStatus.Reconnecting;
Reset();
};
_subscription.ConnectionClosed += () =>
{
log.Write(LogLevel.Warning, $"{Id} order book {Symbol} disconnected");
Status = OrderBookStatus.Disconnected;
_ = StopAsync();
};
_subscription.ConnectionLost += HandleConnectionLost;
_subscription.ConnectionClosed += HandleConnectionClosed;
_subscription.ConnectionRestored += HandleConnectionRestored;
_subscription.ConnectionRestored += async time => await ResyncAsync().ConfigureAwait(false);
Status = OrderBookStatus.Synced;
return new CallResult<bool>(true);
}
private void HandleConnectionLost() {
log.Write(LogLevel.Warning, $"{Id} order book {Symbol} connection lost");
if (Status != OrderBookStatus.Disposed) {
Status = OrderBookStatus.Reconnecting;
Reset();
}
}
private void HandleConnectionClosed() {
log.Write(LogLevel.Warning, $"{Id} order book {Symbol} disconnected");
Status = OrderBookStatus.Disconnected;
_ = StopAsync();
}
private async void HandleConnectionRestored(TimeSpan _) {
await ResyncAsync().ConfigureAwait(false);
}
/// <inheritdoc/>
public async Task StopAsync()
{
@@ -286,8 +294,12 @@ namespace CryptoExchange.Net.OrderBook
if (_processTask != null)
await _processTask.ConfigureAwait(false);
if (_subscription != null)
if (_subscription != null) {
await _subscription.CloseAsync().ConfigureAwait(false);
_subscription.ConnectionLost -= HandleConnectionLost;
_subscription.ConnectionClosed -= HandleConnectionClosed;
_subscription.ConnectionRestored -= HandleConnectionRestored;
}
log.Write(LogLevel.Trace, $"{Id} order book {Symbol} stopped");
}
@@ -601,13 +613,13 @@ namespace CryptoExchange.Net.OrderBook
private async Task ProcessQueue()
{
while (Status != OrderBookStatus.Disconnected)
while (Status != OrderBookStatus.Disconnected && Status != OrderBookStatus.Disposed)
{
await _queueEvent.WaitAsync().ConfigureAwait(false);
while (_processQueue.TryDequeue(out var item))
{
if (Status == OrderBookStatus.Disconnected)
if (Status == OrderBookStatus.Disconnected || Status == OrderBookStatus.Disposed)
break;
if (_stopProcessing)
@@ -31,9 +31,12 @@ namespace CryptoExchange.Net.Sockets
private readonly IDictionary<string, string> cookies;
private readonly IDictionary<string, string> headers;
private CancellationTokenSource _ctsSource;
private ApiProxy? _proxy;
private readonly List<DateTime> _outgoingMessages;
private DateTime _lastReceivedMessagesUpdate;
private bool _closed;
private bool _disposed;
/// <summary>
/// Received messages, the size and the timstamp
@@ -122,6 +125,9 @@ namespace CryptoExchange.Net.Sockets
/// <inheritdoc />
public TimeSpan Timeout { get; set; }
/// <inheritdoc />
public TimeSpan KeepAliveInterval { get; set; }
/// <inheritdoc />
public double IncomingKbps
{
@@ -204,6 +210,8 @@ namespace CryptoExchange.Net.Sockets
/// <inheritdoc />
public virtual void SetProxy(ApiProxy proxy)
{
_proxy = proxy;
if (!Uri.TryCreate($"{proxy.Host}:{proxy.Port}", UriKind.Absolute, out var uri))
throw new ArgumentException("Proxy settings invalid, {proxy.Host}:{proxy.Port} not a valid URI", nameof(proxy));
@@ -276,12 +284,31 @@ namespace CryptoExchange.Net.Sockets
/// <returns></returns>
private async Task CloseInternalAsync()
{
if (_closed || _disposed)
return;
_closed = true;
_ctsSource.Cancel();
_sendEvent.Set();
if (_socket.State == WebSocketState.Open)
await _socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "Closing", default).ConfigureAwait(false);
{
try
{
await _socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "Closing", default).ConfigureAwait(false);
}
catch(Exception)
{ } // Can sometimes throw an exception when socket is in aborted state due to timing
}
else if(_socket.State == WebSocketState.CloseReceived)
{
try
{
await _socket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", default).ConfigureAwait(false);
}
catch (Exception)
{ } // Can sometimes throw an exception when socket is in aborted state due to timing
}
log.Write(LogLevel.Debug, $"Socket {Id} closed");
Handle(closeHandlers);
}
@@ -291,7 +318,11 @@ namespace CryptoExchange.Net.Sockets
/// </summary>
public void Dispose()
{
if (_disposed)
return;
log.Write(LogLevel.Debug, $"Socket {Id} disposing");
_disposed = true;
_socket.Dispose();
_ctsSource.Dispose();
@@ -311,6 +342,9 @@ namespace CryptoExchange.Net.Sockets
while (_sendBuffer.TryDequeue(out _)) { } // Clear send buffer
_socket = CreateSocket();
if (_proxy != null)
SetProxy(_proxy);
_closed = false;
}
/// <summary>
@@ -326,7 +360,7 @@ namespace CryptoExchange.Net.Sockets
socket.Options.Cookies = cookieContainer;
foreach (var header in headers)
socket.Options.SetRequestHeader(header.Key, header.Value);
socket.Options.KeepAliveInterval = TimeSpan.FromSeconds(10);
socket.Options.KeepAliveInterval = KeepAliveInterval;
socket.Options.SetBuffer(65536, 65536); // Setting it to anything bigger than 65536 throws an exception in .net framework
return socket;
}
+117 -26
View File
@@ -139,7 +139,25 @@ namespace CryptoExchange.Net.Sockets
private readonly BaseSocketClient socketClient;
private readonly List<PendingRequest> pendingRequests;
private Task? _socketProcessReconnectTask;
private Task? _socketProcessTask;
private Task? _socketReconnectTask;
private readonly AsyncResetEvent _reconnectWaitEvent;
private SocketStatus _status;
/// <summary>
/// Status of the socket connection
/// </summary>
public SocketStatus Status
{
get => _status;
private set
{
var oldStatus = _status;
_status = value;
log.Write(LogLevel.Trace, $"Socket {SocketId} status changed from {oldStatus} to {_status}");
}
}
/// <summary>
/// The underlying websocket
@@ -163,9 +181,13 @@ namespace CryptoExchange.Net.Sockets
subscriptions = new List<SocketSubscription>();
_socket = socket;
_reconnectWaitEvent = new AsyncResetEvent(false, true);
_socket.Timeout = client.ClientOptions.SocketNoDataTimeout;
_socket.OnMessage += ProcessMessage;
_socket.OnOpen += SocketOnOpen;
_socket.OnClose += () => _reconnectWaitEvent.Set();
}
/// <summary>
@@ -176,7 +198,11 @@ namespace CryptoExchange.Net.Sockets
{
var connected = await _socket.ConnectAsync().ConfigureAwait(false);
if (connected)
StartProcessingTask();
{
Status = SocketStatus.Connected;
_socketReconnectTask = ReconnectWatcherAsync();
_socketProcessTask = _socket.ProcessAsync();
}
return connected;
}
@@ -205,6 +231,9 @@ namespace CryptoExchange.Net.Sockets
/// <returns></returns>
public async Task CloseAsync()
{
if (Status == SocketStatus.Closed || Status == SocketStatus.Disposed)
return;
ShouldReconnect = false;
if (socketClient.socketConnections.ContainsKey(SocketId))
socketClient.socketConnections.TryRemove(SocketId, out _);
@@ -217,11 +246,14 @@ namespace CryptoExchange.Net.Sockets
subscription.CancellationTokenRegistration.Value.Dispose();
}
}
while (Status == SocketStatus.Reconnecting)
// Wait for reconnecting to finish
await Task.Delay(100).ConfigureAwait(false);
await _socket.CloseAsync().ConfigureAwait(false);
if (_socketProcessReconnectTask != null)
await _socketProcessReconnectTask.ConfigureAwait(false);
if(_socketProcessTask != null)
await _socketProcessTask.ConfigureAwait(false);
_socket.Dispose();
}
@@ -232,37 +264,40 @@ namespace CryptoExchange.Net.Sockets
/// <returns></returns>
public async Task CloseAsync(SocketSubscription subscription)
{
if (!_socket.IsOpen)
if (Status == SocketStatus.Closing || Status == SocketStatus.Closed || Status == SocketStatus.Disposed)
return;
log.Write(LogLevel.Trace, $"Socket {SocketId} closing subscription {subscription.Id}");
if (subscription.CancellationTokenRegistration.HasValue)
subscription.CancellationTokenRegistration.Value.Dispose();
if (subscription.Confirmed)
if (subscription.Confirmed && _socket.IsOpen)
await socketClient.UnsubscribeAsync(this, subscription).ConfigureAwait(false);
bool shouldCloseConnection;
lock (subscriptionLock)
shouldCloseConnection = !subscriptions.Any(r => r.UserSubscription && subscription != r);
{
if (Status == SocketStatus.Closing)
{
log.Write(LogLevel.Trace, $"Socket {SocketId} already closing");
return;
}
shouldCloseConnection = subscriptions.All(r => !r.UserSubscription || r == subscription);
if (shouldCloseConnection)
Status = SocketStatus.Closing;
}
if (shouldCloseConnection)
{
log.Write(LogLevel.Trace, $"Socket {SocketId} closing as there are no more subscriptions");
await CloseAsync().ConfigureAwait(false);
}
lock (subscriptionLock)
subscriptions.Remove(subscription);
}
private void StartProcessingTask()
{
log.Write(LogLevel.Trace, $"Starting {SocketId} process/reconnect task");
_socketProcessReconnectTask = Task.Run(async () =>
{
await _socket.ProcessAsync().ConfigureAwait(false);
await ReconnectAsync().ConfigureAwait(false);
log.Write(LogLevel.Trace, $"Process/reconnect {SocketId} task finished");
});
}
private async Task ReconnectAsync()
{
// Fail all pending requests
@@ -283,7 +318,7 @@ namespace CryptoExchange.Net.Sockets
if (!lostTriggered)
{
lostTriggered = true;
ConnectionLost?.Invoke();
_ = Task.Run(() => ConnectionLost?.Invoke());
}
while (ShouldReconnect)
@@ -326,7 +361,8 @@ namespace CryptoExchange.Net.Sockets
}
// Successfully reconnected, start processing
StartProcessingTask();
Status = SocketStatus.Connected;
_socketProcessTask = _socket.ProcessAsync();
ReconnectTry = 0;
var time = DisconnectTime;
@@ -372,7 +408,7 @@ namespace CryptoExchange.Net.Sockets
if (lostTriggered)
{
lostTriggered = false;
_ = Task.Run(() => ConnectionRestored?.Invoke(time.HasValue ? DateTime.UtcNow - time.Value : TimeSpan.FromSeconds(0))).ConfigureAwait(false);
_ = Task.Run(() => ConnectionRestored?.Invoke(time.HasValue ? DateTime.UtcNow - time.Value : TimeSpan.FromSeconds(0)));
}
break;
@@ -396,6 +432,7 @@ namespace CryptoExchange.Net.Sockets
/// </summary>
public void Dispose()
{
Status = SocketStatus.Disposed;
_socket.Dispose();
}
@@ -458,7 +495,7 @@ namespace CryptoExchange.Net.Sockets
var total = DateTime.UtcNow - timestamp;
if (userProcessTime.TotalMilliseconds > 500)
log.Write(LogLevel.Debug, $"Socket {SocketId} message processing slow ({(int)total.TotalMilliseconds}ms), consider offloading data handling to another thread. " +
log.Write(LogLevel.Debug, $"Socket {SocketId} message processing slow ({(int)total.TotalMilliseconds}ms, {(int)userProcessTime.TotalMilliseconds}ms user code), consider offloading data handling to another thread. " +
"Data from this socket may arrive late or not at all if message processing is continuously slow.");
log.Write(LogLevel.Trace, $"Socket {SocketId} message processed in {(int)total.TotalMilliseconds}ms, ({(int)userProcessTime.TotalMilliseconds}ms user code)");
@@ -468,10 +505,17 @@ namespace CryptoExchange.Net.Sockets
/// Add a subscription to this connection
/// </summary>
/// <param name="subscription"></param>
public void AddSubscription(SocketSubscription subscription)
public bool AddSubscription(SocketSubscription subscription)
{
lock(subscriptionLock)
lock (subscriptionLock)
{
if (Status != SocketStatus.None && Status != SocketStatus.Connected)
return false;
subscriptions.Add(subscription);
log.Write(LogLevel.Trace, $"Socket {SocketId} adding new subscription with id {subscription.Id}, total subscriptions on connection: {subscriptions.Count}");
return true;
}
}
/// <summary>
@@ -614,6 +658,22 @@ namespace CryptoExchange.Net.Sockets
PausedActivity = false;
}
private async Task ReconnectWatcherAsync()
{
while (true)
{
await _reconnectWaitEvent.WaitAsync().ConfigureAwait(false);
if (!ShouldReconnect)
return;
Status = SocketStatus.Reconnecting;
await ReconnectAsync().ConfigureAwait(false);
if (!ShouldReconnect)
return;
}
}
private async Task<CallResult<bool>> ProcessReconnectAsync()
{
if (!_socket.IsOpen)
@@ -671,5 +731,36 @@ namespace CryptoExchange.Net.Sockets
return await socketClient.SubscribeAndWaitAsync(this, socketSubscription.Request!, socketSubscription).ConfigureAwait(false);
}
/// <summary>
/// Status of the socket connection
/// </summary>
public enum SocketStatus
{
/// <summary>
/// None/Initial
/// </summary>
None,
/// <summary>
/// Connected
/// </summary>
Connected,
/// <summary>
/// Reconnecting
/// </summary>
Reconnecting,
/// <summary>
/// Closing
/// </summary>
Closing,
/// <summary>
/// Closed
/// </summary>
Closed,
/// <summary>
/// Disposed
/// </summary>
Disposed
}
}
}
+28
View File
@@ -18,6 +18,34 @@ I develop and maintain this package on my own for free in my spare time. Donatio
Alternatively, sponsor me on Github using [Github Sponsors](https://github.com/sponsors/JKorf)
## Release notes
* Version 5.1.12 - 12 Jun 2022
* Changed time sync so requests no longer wait for it to complete unless it's the first time
* Made log client options changable after client creation
* Fixed proxy setting not used when reconnecting socket
* Changed MaxSocketConnections to a client options
* Updated socket reconnection logic
* Version 5.1.12 - 12 Jun 2022
* Changed time sync so requests no longer wait for it to complete unless it's the first time
* Made log client options changable after client creation
* Fixed proxy setting not used when reconnecting socket
* Updated socket reconnection logic
* Version 5.1.11 - 24 May 2022
* Added KeepAliveInterval setting
* Fixed port not being copied when setting parameters on request
* Fixed inconsistent PackageReference casing in csproj
* Version 5.1.10 - 22 May 2022
* Fixed order book reconnecting while Diposed
* Fixed exception when disposing socket client while reconnecting
* Added additional null/default checking in DateTimeConverter
* Changed ConnectionLost subscription event to run in seperate task to prevent exception/longer operations from intervering with reconnecting
* Version 5.1.9 - 08 May 2022
* Added latency to the timesync calculation
* Small fix for exception in socket close handling
* Version 5.1.8 - 01 May 2022
* Cleanup socket code, fixed an issue which could cause connections to never reconnect when connection was lost
* Added support for sending requests which expect an empty response
+4 -1
View File
@@ -61,4 +61,7 @@ var client = new BinanceClient(new BinanceClientOptions
BaseAddress = BinanceApiAddresses.TestNet.UsdFuturesRestClientAddress
}
});
```
```
### How are timezones handled / Timestamps are off by xx
Exchange API's treat all timestamps as UTC, both incoming and outgoing. The client libraries do no conversion so be sure to use UTC as well.