mirror of
https://github.com/JKorf/CryptoExchange.Net.git
synced 2026-08-13 17:33:02 +00:00
Compare commits
24 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| ea9375d582 | |||
| 2cf3c93e5e | |||
| ca888d8e41 | |||
| 2040b1c175 | |||
| d451c18821 | |||
| c13dfa4461 | |||
| c2080ef75f | |||
| 6b252e8024 | |||
| d06bd5f176 | |||
| d55fc8da65 | |||
| 01184f2c5d | |||
| cadc93c2f0 | |||
| 2600a51461 | |||
| 9e6a86ba8b | |||
| c4430d63fa | |||
| f3e1cfef33 | |||
| cc3053719c | |||
| cd6907e601 | |||
| 8fe00693bd | |||
| fb90d1e015 | |||
| 4b44861e43 | |||
| e42ca4ab5a | |||
| 5b97f6dd67 | |||
| a9813ecb0a |
@@ -6,10 +6,10 @@
|
|||||||
</PropertyGroup>
|
</PropertyGroup>
|
||||||
|
|
||||||
<ItemGroup>
|
<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="Moq" Version="4.16.1" />
|
||||||
<packagereference Include="NUnit" Version="3.13.2"></packagereference>
|
<PackageReference Include="NUnit" Version="3.13.2"></PackageReference>
|
||||||
<packagereference Include="NUnit3TestAdapter" Version="4.2.0"></packagereference>
|
<PackageReference Include="NUnit3TestAdapter" Version="4.2.0"></PackageReference>
|
||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|
||||||
<ItemGroup>
|
<ItemGroup>
|
||||||
|
|||||||
@@ -135,7 +135,7 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
|
|||||||
throw new NotImplementedException();
|
throw new NotImplementedException();
|
||||||
}
|
}
|
||||||
|
|
||||||
protected override TimeSyncInfo GetTimeSyncInfo()
|
public override TimeSyncInfo GetTimeSyncInfo()
|
||||||
{
|
{
|
||||||
throw new NotImplementedException();
|
throw new NotImplementedException();
|
||||||
}
|
}
|
||||||
@@ -161,7 +161,7 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
|
|||||||
throw new NotImplementedException();
|
throw new NotImplementedException();
|
||||||
}
|
}
|
||||||
|
|
||||||
protected override TimeSyncInfo GetTimeSyncInfo()
|
public override TimeSyncInfo GetTimeSyncInfo()
|
||||||
{
|
{
|
||||||
throw new NotImplementedException();
|
throw new NotImplementedException();
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -40,6 +40,8 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
|
|||||||
|
|
||||||
public Uri Uri => new Uri("");
|
public Uri Uri => new Uri("");
|
||||||
|
|
||||||
|
public TimeSpan KeepAliveInterval { get; set; }
|
||||||
|
|
||||||
public static int lastId = 0;
|
public static int lastId = 0;
|
||||||
public static object lastIdLock = new object();
|
public static object lastIdLock = new object();
|
||||||
|
|
||||||
|
|||||||
@@ -63,6 +63,7 @@ namespace CryptoExchange.Net
|
|||||||
log = new Log(name);
|
log = new Log(name);
|
||||||
log.UpdateWriters(options.LogWriters);
|
log.UpdateWriters(options.LogWriters);
|
||||||
log.Level = options.LogLevel;
|
log.Level = options.LogLevel;
|
||||||
|
options.OnLoggingChanged += HandleLogConfigChange;
|
||||||
|
|
||||||
ClientOptions = options;
|
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>
|
/// <summary>
|
||||||
/// Dispose
|
/// Dispose
|
||||||
/// </summary>
|
/// </summary>
|
||||||
public virtual void Dispose()
|
public virtual void Dispose()
|
||||||
{
|
{
|
||||||
log.Write(LogLevel.Debug, "Disposing client");
|
log.Write(LogLevel.Debug, "Disposing client");
|
||||||
|
ClientOptions.OnLoggingChanged -= HandleLogConfigChange;
|
||||||
foreach (var client in ApiClients)
|
foreach (var client in ApiClients)
|
||||||
client.Dispose();
|
client.Dispose();
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -172,11 +172,17 @@ namespace CryptoExchange.Net
|
|||||||
|
|
||||||
if (signed)
|
if (signed)
|
||||||
{
|
{
|
||||||
var syncTimeResult = await apiClient.SyncTimeAsync().ConfigureAwait(false);
|
var syncTask = apiClient.SyncTimeAsync();
|
||||||
if (!syncTimeResult)
|
var timeSyncInfo = apiClient.GetTimeSyncInfo();
|
||||||
|
if (timeSyncInfo.TimeSyncState.LastSyncTime == default)
|
||||||
{
|
{
|
||||||
log.Write(LogLevel.Debug, $"[{requestId}] Failed to sync time, aborting request: " + syncTimeResult.Error);
|
// Initially with first request we'll need to wait for the time syncing, if it's not the first request we can just continue
|
||||||
return syncTimeResult.As<IRequest>(default);
|
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);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -36,9 +36,9 @@ namespace CryptoExchange.Net
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
protected internal readonly SemaphoreSlim semaphoreSlim = new(1);
|
protected internal readonly SemaphoreSlim semaphoreSlim = new(1);
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// The max amount of concurrent socket connections
|
/// Keep alive interval for websocket connection
|
||||||
/// </summary>
|
/// </summary>
|
||||||
protected int MaxSocketConnections { get; set; } = 9999;
|
protected TimeSpan KeepAliveInterval { get; set; } = TimeSpan.FromSeconds(10);
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Delegate used for processing byte data received from socket connections before it is processed by handlers
|
/// Delegate used for processing byte data received from socket connections before it is processed by handlers
|
||||||
/// </summary>
|
/// </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>
|
/// <summary>
|
||||||
/// Client options
|
/// Client options
|
||||||
/// </summary>
|
/// </summary>
|
||||||
@@ -164,7 +178,7 @@ namespace CryptoExchange.Net
|
|||||||
return new CallResult<UpdateSubscription>(new InvalidOperationError("Client disposed, can't subscribe"));
|
return new CallResult<UpdateSubscription>(new InvalidOperationError("Client disposed, can't subscribe"));
|
||||||
|
|
||||||
SocketConnection socketConnection;
|
SocketConnection socketConnection;
|
||||||
SocketSubscription subscription;
|
SocketSubscription? subscription;
|
||||||
var released = false;
|
var released = false;
|
||||||
// Wait for a semaphore here, so we only connect 1 socket at a time.
|
// 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
|
// This is necessary for being able to see if connections can be combined
|
||||||
@@ -179,23 +193,34 @@ namespace CryptoExchange.Net
|
|||||||
|
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
// Get a new or existing socket connection
|
while (true)
|
||||||
socketConnection = GetSocketConnection(apiClient, url, authenticated);
|
|
||||||
|
|
||||||
// Add a subscription on the socket connection
|
|
||||||
subscription = AddSubscription(request, identifier, true, socketConnection, dataHandler);
|
|
||||||
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
|
// Get a new or existing socket connection
|
||||||
semaphoreSlim.Release();
|
socketConnection = GetSocketConnection(apiClient, url, authenticated);
|
||||||
released = true;
|
|
||||||
|
// 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
|
finally
|
||||||
{
|
{
|
||||||
@@ -443,9 +468,6 @@ namespace CryptoExchange.Net
|
|||||||
/// <param name="message"></param>
|
/// <param name="message"></param>
|
||||||
/// <returns></returns>
|
/// <returns></returns>
|
||||||
protected internal virtual JToken ProcessTokenData(JToken message)
|
protected internal virtual JToken ProcessTokenData(JToken message)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
{
|
{
|
||||||
return message;
|
return message;
|
||||||
}
|
}
|
||||||
@@ -460,7 +482,7 @@ namespace CryptoExchange.Net
|
|||||||
/// <param name="connection">The socket connection the handler is on</param>
|
/// <param name="connection">The socket connection the handler is on</param>
|
||||||
/// <param name="dataHandler">The handler of the data received</param>
|
/// <param name="dataHandler">The handler of the data received</param>
|
||||||
/// <returns></returns>
|
/// <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)
|
void InternalHandler(MessageEvent messageEvent)
|
||||||
{
|
{
|
||||||
@@ -484,7 +506,8 @@ namespace CryptoExchange.Net
|
|||||||
var subscription = request == null
|
var subscription = request == null
|
||||||
? SocketSubscription.CreateForIdentifier(NextId(), identifier!, userSubscription, InternalHandler)
|
? SocketSubscription.CreateForIdentifier(NextId(), identifier!, userSubscription, InternalHandler)
|
||||||
: SocketSubscription.CreateForRequest(NextId(), request, userSubscription, InternalHandler);
|
: SocketSubscription.CreateForRequest(NextId(), request, userSubscription, InternalHandler);
|
||||||
connection.AddSubscription(subscription);
|
if (!connection.AddSubscription(subscription))
|
||||||
|
return null;
|
||||||
return subscription;
|
return subscription;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -510,13 +533,14 @@ namespace CryptoExchange.Net
|
|||||||
/// <returns></returns>
|
/// <returns></returns>
|
||||||
protected virtual SocketConnection GetSocketConnection(SocketApiClient apiClient, string address, bool authenticated)
|
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.ApiClient.GetType() == apiClient.GetType())
|
||||||
&& (s.Value.Authenticated == authenticated || !authenticated) && s.Value.Connected).OrderBy(s => s.Value.SubscriptionCount).FirstOrDefault();
|
&& (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;
|
var result = socketResult.Equals(default(KeyValuePair<int, SocketConnection>)) ? null : socketResult.Value;
|
||||||
if (result != null)
|
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
|
// Use existing socket if it has less than target connections OR it has the least connections and we can't make new
|
||||||
return result;
|
return result;
|
||||||
@@ -574,6 +598,7 @@ namespace CryptoExchange.Net
|
|||||||
if (ClientOptions.Proxy != null)
|
if (ClientOptions.Proxy != null)
|
||||||
socket.SetProxy(ClientOptions.Proxy);
|
socket.SetProxy(ClientOptions.Proxy);
|
||||||
|
|
||||||
|
socket.KeepAliveInterval = KeepAliveInterval;
|
||||||
socket.Timeout = ClientOptions.SocketNoDataTimeout;
|
socket.Timeout = ClientOptions.SocketNoDataTimeout;
|
||||||
socket.DataInterpreterBytes = dataInterpreterBytes;
|
socket.DataInterpreterBytes = dataInterpreterBytes;
|
||||||
socket.DataInterpreterString = dataInterpreterString;
|
socket.DataInterpreterString = dataInterpreterString;
|
||||||
|
|||||||
@@ -18,7 +18,7 @@ namespace CryptoExchange.Net
|
|||||||
/// Get time sync info for an API client
|
/// Get time sync info for an API client
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <returns></returns>
|
/// <returns></returns>
|
||||||
protected abstract TimeSyncInfo GetTimeSyncInfo();
|
public abstract TimeSyncInfo GetTimeSyncInfo();
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Get time offset for an API client
|
/// Get time offset for an API client
|
||||||
|
|||||||
@@ -31,7 +31,7 @@ namespace CryptoExchange.Net.Converters
|
|||||||
if(reader.TokenType is JsonToken.Integer)
|
if(reader.TokenType is JsonToken.Integer)
|
||||||
{
|
{
|
||||||
var longValue = (long)reader.Value;
|
var longValue = (long)reader.Value;
|
||||||
if (longValue == 0)
|
if (longValue == 0 || longValue == -1)
|
||||||
return objectType == typeof(DateTime) ? default(DateTime): null;
|
return objectType == typeof(DateTime) ? default(DateTime): null;
|
||||||
if (longValue < 19999999999)
|
if (longValue < 19999999999)
|
||||||
return ConvertFromSeconds(longValue);
|
return ConvertFromSeconds(longValue);
|
||||||
@@ -45,6 +45,9 @@ namespace CryptoExchange.Net.Converters
|
|||||||
else if (reader.TokenType is JsonToken.Float)
|
else if (reader.TokenType is JsonToken.Float)
|
||||||
{
|
{
|
||||||
var doubleValue = (double)reader.Value;
|
var doubleValue = (double)reader.Value;
|
||||||
|
if (doubleValue == 0 || doubleValue == -1)
|
||||||
|
return objectType == typeof(DateTime) ? default(DateTime) : null;
|
||||||
|
|
||||||
if (doubleValue < 19999999999)
|
if (doubleValue < 19999999999)
|
||||||
return ConvertFromSeconds(doubleValue);
|
return ConvertFromSeconds(doubleValue);
|
||||||
|
|
||||||
@@ -56,6 +59,9 @@ namespace CryptoExchange.Net.Converters
|
|||||||
if (string.IsNullOrWhiteSpace(stringValue))
|
if (string.IsNullOrWhiteSpace(stringValue))
|
||||||
return null;
|
return null;
|
||||||
|
|
||||||
|
if (string.IsNullOrWhiteSpace(stringValue) || stringValue == "0" || stringValue == "-1")
|
||||||
|
return objectType == typeof(DateTime) ? default(DateTime) : null;
|
||||||
|
|
||||||
if (stringValue.Length == 8)
|
if (stringValue.Length == 8)
|
||||||
{
|
{
|
||||||
// Parse 20211103 format
|
// Parse 20211103 format
|
||||||
|
|||||||
@@ -6,16 +6,16 @@
|
|||||||
<PackageId>CryptoExchange.Net</PackageId>
|
<PackageId>CryptoExchange.Net</PackageId>
|
||||||
<Authors>JKorf</Authors>
|
<Authors>JKorf</Authors>
|
||||||
<Description>A base package for implementing cryptocurrency API's</Description>
|
<Description>A base package for implementing cryptocurrency API's</Description>
|
||||||
<PackageVersion>5.1.9</PackageVersion>
|
<PackageVersion>5.1.12</PackageVersion>
|
||||||
<AssemblyVersion>5.1.9</AssemblyVersion>
|
<AssemblyVersion>5.1.12</AssemblyVersion>
|
||||||
<FileVersion>5.1.9</FileVersion>
|
<FileVersion>5.1.12</FileVersion>
|
||||||
<PackageRequireLicenseAcceptance>false</PackageRequireLicenseAcceptance>
|
<PackageRequireLicenseAcceptance>false</PackageRequireLicenseAcceptance>
|
||||||
<RepositoryType>git</RepositoryType>
|
<RepositoryType>git</RepositoryType>
|
||||||
<RepositoryUrl>https://github.com/JKorf/CryptoExchange.Net.git</RepositoryUrl>
|
<RepositoryUrl>https://github.com/JKorf/CryptoExchange.Net.git</RepositoryUrl>
|
||||||
<PackageProjectUrl>https://github.com/JKorf/CryptoExchange.Net</PackageProjectUrl>
|
<PackageProjectUrl>https://github.com/JKorf/CryptoExchange.Net</PackageProjectUrl>
|
||||||
<NeutralLanguage>en</NeutralLanguage>
|
<NeutralLanguage>en</NeutralLanguage>
|
||||||
<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
|
<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
|
||||||
<PackageReleaseNotes>5.1.9 - Added latency to the timesync calculation, Small fix for exception in socket close handling</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>
|
<Nullable>enable</Nullable>
|
||||||
<LangVersion>9.0</LangVersion>
|
<LangVersion>9.0</LangVersion>
|
||||||
<PackageLicenseExpression>MIT</PackageLicenseExpression>
|
<PackageLicenseExpression>MIT</PackageLicenseExpression>
|
||||||
|
|||||||
@@ -426,6 +426,7 @@ namespace CryptoExchange.Net
|
|||||||
var uriBuilder = new UriBuilder();
|
var uriBuilder = new UriBuilder();
|
||||||
uriBuilder.Scheme = baseUri.Scheme;
|
uriBuilder.Scheme = baseUri.Scheme;
|
||||||
uriBuilder.Host = baseUri.Host;
|
uriBuilder.Host = baseUri.Host;
|
||||||
|
uriBuilder.Port = baseUri.Port;
|
||||||
uriBuilder.Path = baseUri.AbsolutePath;
|
uriBuilder.Path = baseUri.AbsolutePath;
|
||||||
var httpValueCollection = HttpUtility.ParseQueryString(string.Empty);
|
var httpValueCollection = HttpUtility.ParseQueryString(string.Empty);
|
||||||
foreach (var parameter in parameters)
|
foreach (var parameter in parameters)
|
||||||
@@ -454,6 +455,7 @@ namespace CryptoExchange.Net
|
|||||||
var uriBuilder = new UriBuilder();
|
var uriBuilder = new UriBuilder();
|
||||||
uriBuilder.Scheme = baseUri.Scheme;
|
uriBuilder.Scheme = baseUri.Scheme;
|
||||||
uriBuilder.Host = baseUri.Host;
|
uriBuilder.Host = baseUri.Host;
|
||||||
|
uriBuilder.Port = baseUri.Port;
|
||||||
uriBuilder.Path = baseUri.AbsolutePath;
|
uriBuilder.Path = baseUri.AbsolutePath;
|
||||||
var httpValueCollection = HttpUtility.ParseQueryString(string.Empty);
|
var httpValueCollection = HttpUtility.ParseQueryString(string.Empty);
|
||||||
foreach (var parameter in parameters)
|
foreach (var parameter in parameters)
|
||||||
|
|||||||
@@ -27,6 +27,16 @@ namespace CryptoExchange.Net.Interfaces
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
public double IncomingKbps { get; }
|
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>
|
/// <summary>
|
||||||
/// Unsubscribe from a stream using the subscription id received when starting the subscription
|
/// Unsubscribe from a stream using the subscription id received when starting the subscription
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
|||||||
@@ -77,6 +77,10 @@ namespace CryptoExchange.Net.Interfaces
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
TimeSpan Timeout { get; set; }
|
TimeSpan Timeout { get; set; }
|
||||||
/// <summary>
|
/// <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
|
/// Set a proxy to use when connecting
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="proxy"></param>
|
/// <param name="proxy"></param>
|
||||||
|
|||||||
@@ -26,6 +26,8 @@ namespace CryptoExchange.Net.Logging
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
public string ClientName { get; set; }
|
public string ClientName { get; set; }
|
||||||
|
|
||||||
|
private readonly object _lock = new object();
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// ctor
|
/// ctor
|
||||||
/// </summary>
|
/// </summary>
|
||||||
@@ -42,7 +44,8 @@ namespace CryptoExchange.Net.Logging
|
|||||||
/// <param name="textWriters"></param>
|
/// <param name="textWriters"></param>
|
||||||
public void UpdateWriters(List<ILogger> textWriters)
|
public void UpdateWriters(List<ILogger> textWriters)
|
||||||
{
|
{
|
||||||
writers = textWriters;
|
lock (_lock)
|
||||||
|
writers = textWriters;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
@@ -56,16 +59,19 @@ namespace CryptoExchange.Net.Logging
|
|||||||
return;
|
return;
|
||||||
|
|
||||||
var logMessage = $"{ClientName,-10} | {message}";
|
var logMessage = $"{ClientName,-10} | {message}";
|
||||||
foreach (var writer in writers.ToList())
|
lock (_lock)
|
||||||
{
|
{
|
||||||
try
|
foreach (var writer in writers)
|
||||||
{
|
{
|
||||||
writer.Log(logLevel, logMessage);
|
try
|
||||||
}
|
{
|
||||||
catch (Exception e)
|
writer.Log(logLevel, logMessage);
|
||||||
{
|
}
|
||||||
// Can't write to the logging so where else to output..
|
catch (Exception e)
|
||||||
Trace.WriteLine($"{DateTime.Now:yyyy/MM/dd HH:mm:ss:fff} | Warning | Failed to write log to writer {writer.GetType()}: " + e.ToLogString());
|
{
|
||||||
|
// 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());
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -14,15 +14,35 @@ namespace CryptoExchange.Net.Objects
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
public class BaseOptions
|
public class BaseOptions
|
||||||
{
|
{
|
||||||
|
internal event Action? OnLoggingChanged;
|
||||||
|
|
||||||
|
private LogLevel _logLevel = LogLevel.Information;
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// The minimum log level to output
|
/// The minimum log level to output
|
||||||
/// </summary>
|
/// </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>
|
/// <summary>
|
||||||
/// The log writers
|
/// The log writers
|
||||||
/// </summary>
|
/// </summary>
|
||||||
public List<ILogger> LogWriters { get; set; } = new List<ILogger> { new DebugLogger() };
|
public List<ILogger> LogWriters
|
||||||
|
{
|
||||||
|
get => _logWriters;
|
||||||
|
set
|
||||||
|
{
|
||||||
|
_logWriters = value;
|
||||||
|
OnLoggingChanged?.Invoke();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// If true, the CallResult and DataEvent objects will also include the originally received json data in the OriginalData property
|
/// 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>
|
/// </summary>
|
||||||
public int? SocketSubscriptionsCombineTarget { get; set; }
|
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>
|
/// <summary>
|
||||||
/// ctor
|
/// ctor
|
||||||
/// </summary>
|
/// </summary>
|
||||||
@@ -213,12 +238,13 @@ namespace CryptoExchange.Net.Objects
|
|||||||
SocketResponseTimeout = baseOptions.SocketResponseTimeout;
|
SocketResponseTimeout = baseOptions.SocketResponseTimeout;
|
||||||
SocketNoDataTimeout = baseOptions.SocketNoDataTimeout;
|
SocketNoDataTimeout = baseOptions.SocketNoDataTimeout;
|
||||||
SocketSubscriptionsCombineTarget = baseOptions.SocketSubscriptionsCombineTarget;
|
SocketSubscriptionsCombineTarget = baseOptions.SocketSubscriptionsCombineTarget;
|
||||||
|
MaxSocketConnections = baseOptions.MaxSocketConnections;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
public override string ToString()
|
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}";
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -258,24 +258,32 @@ namespace CryptoExchange.Net.OrderBook
|
|||||||
}
|
}
|
||||||
|
|
||||||
_subscription = startResult.Data;
|
_subscription = startResult.Data;
|
||||||
_subscription.ConnectionLost += () =>
|
_subscription.ConnectionLost += HandleConnectionLost;
|
||||||
{
|
_subscription.ConnectionClosed += HandleConnectionClosed;
|
||||||
log.Write(LogLevel.Warning, $"{Id} order book {Symbol} connection lost");
|
_subscription.ConnectionRestored += HandleConnectionRestored;
|
||||||
Status = OrderBookStatus.Reconnecting;
|
|
||||||
Reset();
|
|
||||||
};
|
|
||||||
_subscription.ConnectionClosed += () =>
|
|
||||||
{
|
|
||||||
log.Write(LogLevel.Warning, $"{Id} order book {Symbol} disconnected");
|
|
||||||
Status = OrderBookStatus.Disconnected;
|
|
||||||
_ = StopAsync();
|
|
||||||
};
|
|
||||||
|
|
||||||
_subscription.ConnectionRestored += async time => await ResyncAsync().ConfigureAwait(false);
|
|
||||||
Status = OrderBookStatus.Synced;
|
Status = OrderBookStatus.Synced;
|
||||||
return new CallResult<bool>(true);
|
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/>
|
/// <inheritdoc/>
|
||||||
public async Task StopAsync()
|
public async Task StopAsync()
|
||||||
{
|
{
|
||||||
@@ -286,8 +294,12 @@ namespace CryptoExchange.Net.OrderBook
|
|||||||
if (_processTask != null)
|
if (_processTask != null)
|
||||||
await _processTask.ConfigureAwait(false);
|
await _processTask.ConfigureAwait(false);
|
||||||
|
|
||||||
if (_subscription != null)
|
if (_subscription != null) {
|
||||||
await _subscription.CloseAsync().ConfigureAwait(false);
|
await _subscription.CloseAsync().ConfigureAwait(false);
|
||||||
|
_subscription.ConnectionLost -= HandleConnectionLost;
|
||||||
|
_subscription.ConnectionClosed -= HandleConnectionClosed;
|
||||||
|
_subscription.ConnectionRestored -= HandleConnectionRestored;
|
||||||
|
}
|
||||||
log.Write(LogLevel.Trace, $"{Id} order book {Symbol} stopped");
|
log.Write(LogLevel.Trace, $"{Id} order book {Symbol} stopped");
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -601,13 +613,13 @@ namespace CryptoExchange.Net.OrderBook
|
|||||||
|
|
||||||
private async Task ProcessQueue()
|
private async Task ProcessQueue()
|
||||||
{
|
{
|
||||||
while (Status != OrderBookStatus.Disconnected)
|
while (Status != OrderBookStatus.Disconnected && Status != OrderBookStatus.Disposed)
|
||||||
{
|
{
|
||||||
await _queueEvent.WaitAsync().ConfigureAwait(false);
|
await _queueEvent.WaitAsync().ConfigureAwait(false);
|
||||||
|
|
||||||
while (_processQueue.TryDequeue(out var item))
|
while (_processQueue.TryDequeue(out var item))
|
||||||
{
|
{
|
||||||
if (Status == OrderBookStatus.Disconnected)
|
if (Status == OrderBookStatus.Disconnected || Status == OrderBookStatus.Disposed)
|
||||||
break;
|
break;
|
||||||
|
|
||||||
if (_stopProcessing)
|
if (_stopProcessing)
|
||||||
|
|||||||
@@ -31,9 +31,12 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
private readonly IDictionary<string, string> cookies;
|
private readonly IDictionary<string, string> cookies;
|
||||||
private readonly IDictionary<string, string> headers;
|
private readonly IDictionary<string, string> headers;
|
||||||
private CancellationTokenSource _ctsSource;
|
private CancellationTokenSource _ctsSource;
|
||||||
|
private ApiProxy? _proxy;
|
||||||
|
|
||||||
private readonly List<DateTime> _outgoingMessages;
|
private readonly List<DateTime> _outgoingMessages;
|
||||||
private DateTime _lastReceivedMessagesUpdate;
|
private DateTime _lastReceivedMessagesUpdate;
|
||||||
|
private bool _closed;
|
||||||
|
private bool _disposed;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Received messages, the size and the timstamp
|
/// Received messages, the size and the timstamp
|
||||||
@@ -122,6 +125,9 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
public TimeSpan Timeout { get; set; }
|
public TimeSpan Timeout { get; set; }
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public TimeSpan KeepAliveInterval { get; set; }
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
public double IncomingKbps
|
public double IncomingKbps
|
||||||
{
|
{
|
||||||
@@ -204,6 +210,8 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
public virtual void SetProxy(ApiProxy proxy)
|
public virtual void SetProxy(ApiProxy proxy)
|
||||||
{
|
{
|
||||||
|
_proxy = proxy;
|
||||||
|
|
||||||
if (!Uri.TryCreate($"{proxy.Host}:{proxy.Port}", UriKind.Absolute, out var uri))
|
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));
|
throw new ArgumentException("Proxy settings invalid, {proxy.Host}:{proxy.Port} not a valid URI", nameof(proxy));
|
||||||
|
|
||||||
@@ -276,6 +284,10 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
/// <returns></returns>
|
/// <returns></returns>
|
||||||
private async Task CloseInternalAsync()
|
private async Task CloseInternalAsync()
|
||||||
{
|
{
|
||||||
|
if (_closed || _disposed)
|
||||||
|
return;
|
||||||
|
|
||||||
|
_closed = true;
|
||||||
_ctsSource.Cancel();
|
_ctsSource.Cancel();
|
||||||
_sendEvent.Set();
|
_sendEvent.Set();
|
||||||
|
|
||||||
@@ -288,6 +300,15 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
catch(Exception)
|
catch(Exception)
|
||||||
{ } // Can sometimes throw an exception when socket is in aborted state due to timing
|
{ } // 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");
|
log.Write(LogLevel.Debug, $"Socket {Id} closed");
|
||||||
Handle(closeHandlers);
|
Handle(closeHandlers);
|
||||||
}
|
}
|
||||||
@@ -297,7 +318,11 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
public void Dispose()
|
public void Dispose()
|
||||||
{
|
{
|
||||||
|
if (_disposed)
|
||||||
|
return;
|
||||||
|
|
||||||
log.Write(LogLevel.Debug, $"Socket {Id} disposing");
|
log.Write(LogLevel.Debug, $"Socket {Id} disposing");
|
||||||
|
_disposed = true;
|
||||||
_socket.Dispose();
|
_socket.Dispose();
|
||||||
_ctsSource.Dispose();
|
_ctsSource.Dispose();
|
||||||
|
|
||||||
@@ -317,6 +342,9 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
while (_sendBuffer.TryDequeue(out _)) { } // Clear send buffer
|
while (_sendBuffer.TryDequeue(out _)) { } // Clear send buffer
|
||||||
|
|
||||||
_socket = CreateSocket();
|
_socket = CreateSocket();
|
||||||
|
if (_proxy != null)
|
||||||
|
SetProxy(_proxy);
|
||||||
|
_closed = false;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
@@ -332,7 +360,7 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
socket.Options.Cookies = cookieContainer;
|
socket.Options.Cookies = cookieContainer;
|
||||||
foreach (var header in headers)
|
foreach (var header in headers)
|
||||||
socket.Options.SetRequestHeader(header.Key, header.Value);
|
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
|
socket.Options.SetBuffer(65536, 65536); // Setting it to anything bigger than 65536 throws an exception in .net framework
|
||||||
return socket;
|
return socket;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -139,7 +139,25 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
private readonly BaseSocketClient socketClient;
|
private readonly BaseSocketClient socketClient;
|
||||||
|
|
||||||
private readonly List<PendingRequest> pendingRequests;
|
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>
|
/// <summary>
|
||||||
/// The underlying websocket
|
/// The underlying websocket
|
||||||
@@ -163,9 +181,13 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
subscriptions = new List<SocketSubscription>();
|
subscriptions = new List<SocketSubscription>();
|
||||||
_socket = socket;
|
_socket = socket;
|
||||||
|
|
||||||
|
_reconnectWaitEvent = new AsyncResetEvent(false, true);
|
||||||
|
|
||||||
_socket.Timeout = client.ClientOptions.SocketNoDataTimeout;
|
_socket.Timeout = client.ClientOptions.SocketNoDataTimeout;
|
||||||
_socket.OnMessage += ProcessMessage;
|
_socket.OnMessage += ProcessMessage;
|
||||||
_socket.OnOpen += SocketOnOpen;
|
_socket.OnOpen += SocketOnOpen;
|
||||||
|
_socket.OnClose += () => _reconnectWaitEvent.Set();
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
@@ -176,7 +198,11 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
{
|
{
|
||||||
var connected = await _socket.ConnectAsync().ConfigureAwait(false);
|
var connected = await _socket.ConnectAsync().ConfigureAwait(false);
|
||||||
if (connected)
|
if (connected)
|
||||||
StartProcessingTask();
|
{
|
||||||
|
Status = SocketStatus.Connected;
|
||||||
|
_socketReconnectTask = ReconnectWatcherAsync();
|
||||||
|
_socketProcessTask = _socket.ProcessAsync();
|
||||||
|
}
|
||||||
|
|
||||||
return connected;
|
return connected;
|
||||||
}
|
}
|
||||||
@@ -205,6 +231,9 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
/// <returns></returns>
|
/// <returns></returns>
|
||||||
public async Task CloseAsync()
|
public async Task CloseAsync()
|
||||||
{
|
{
|
||||||
|
if (Status == SocketStatus.Closed || Status == SocketStatus.Disposed)
|
||||||
|
return;
|
||||||
|
|
||||||
ShouldReconnect = false;
|
ShouldReconnect = false;
|
||||||
if (socketClient.socketConnections.ContainsKey(SocketId))
|
if (socketClient.socketConnections.ContainsKey(SocketId))
|
||||||
socketClient.socketConnections.TryRemove(SocketId, out _);
|
socketClient.socketConnections.TryRemove(SocketId, out _);
|
||||||
@@ -217,11 +246,14 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
subscription.CancellationTokenRegistration.Value.Dispose();
|
subscription.CancellationTokenRegistration.Value.Dispose();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
while (Status == SocketStatus.Reconnecting)
|
||||||
|
// Wait for reconnecting to finish
|
||||||
|
await Task.Delay(100).ConfigureAwait(false);
|
||||||
|
|
||||||
await _socket.CloseAsync().ConfigureAwait(false);
|
await _socket.CloseAsync().ConfigureAwait(false);
|
||||||
|
if(_socketProcessTask != null)
|
||||||
if (_socketProcessReconnectTask != null)
|
await _socketProcessTask.ConfigureAwait(false);
|
||||||
await _socketProcessReconnectTask.ConfigureAwait(false);
|
|
||||||
|
|
||||||
_socket.Dispose();
|
_socket.Dispose();
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -232,37 +264,40 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
/// <returns></returns>
|
/// <returns></returns>
|
||||||
public async Task CloseAsync(SocketSubscription subscription)
|
public async Task CloseAsync(SocketSubscription subscription)
|
||||||
{
|
{
|
||||||
if (!_socket.IsOpen)
|
if (Status == SocketStatus.Closing || Status == SocketStatus.Closed || Status == SocketStatus.Disposed)
|
||||||
return;
|
return;
|
||||||
|
|
||||||
|
log.Write(LogLevel.Trace, $"Socket {SocketId} closing subscription {subscription.Id}");
|
||||||
if (subscription.CancellationTokenRegistration.HasValue)
|
if (subscription.CancellationTokenRegistration.HasValue)
|
||||||
subscription.CancellationTokenRegistration.Value.Dispose();
|
subscription.CancellationTokenRegistration.Value.Dispose();
|
||||||
|
|
||||||
if (subscription.Confirmed)
|
if (subscription.Confirmed && _socket.IsOpen)
|
||||||
await socketClient.UnsubscribeAsync(this, subscription).ConfigureAwait(false);
|
await socketClient.UnsubscribeAsync(this, subscription).ConfigureAwait(false);
|
||||||
|
|
||||||
bool shouldCloseConnection;
|
bool shouldCloseConnection;
|
||||||
lock (subscriptionLock)
|
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)
|
if (shouldCloseConnection)
|
||||||
|
{
|
||||||
|
log.Write(LogLevel.Trace, $"Socket {SocketId} closing as there are no more subscriptions");
|
||||||
await CloseAsync().ConfigureAwait(false);
|
await CloseAsync().ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
lock (subscriptionLock)
|
lock (subscriptionLock)
|
||||||
subscriptions.Remove(subscription);
|
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()
|
private async Task ReconnectAsync()
|
||||||
{
|
{
|
||||||
// Fail all pending requests
|
// Fail all pending requests
|
||||||
@@ -283,7 +318,7 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
if (!lostTriggered)
|
if (!lostTriggered)
|
||||||
{
|
{
|
||||||
lostTriggered = true;
|
lostTriggered = true;
|
||||||
ConnectionLost?.Invoke();
|
_ = Task.Run(() => ConnectionLost?.Invoke());
|
||||||
}
|
}
|
||||||
|
|
||||||
while (ShouldReconnect)
|
while (ShouldReconnect)
|
||||||
@@ -326,7 +361,8 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Successfully reconnected, start processing
|
// Successfully reconnected, start processing
|
||||||
StartProcessingTask();
|
Status = SocketStatus.Connected;
|
||||||
|
_socketProcessTask = _socket.ProcessAsync();
|
||||||
|
|
||||||
ReconnectTry = 0;
|
ReconnectTry = 0;
|
||||||
var time = DisconnectTime;
|
var time = DisconnectTime;
|
||||||
@@ -372,7 +408,7 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
if (lostTriggered)
|
if (lostTriggered)
|
||||||
{
|
{
|
||||||
lostTriggered = false;
|
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;
|
break;
|
||||||
@@ -396,6 +432,7 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
public void Dispose()
|
public void Dispose()
|
||||||
{
|
{
|
||||||
|
Status = SocketStatus.Disposed;
|
||||||
_socket.Dispose();
|
_socket.Dispose();
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -458,7 +495,7 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
|
|
||||||
var total = DateTime.UtcNow - timestamp;
|
var total = DateTime.UtcNow - timestamp;
|
||||||
if (userProcessTime.TotalMilliseconds > 500)
|
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.");
|
"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)");
|
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
|
/// Add a subscription to this connection
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="subscription"></param>
|
/// <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);
|
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>
|
/// <summary>
|
||||||
@@ -614,6 +658,22 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
PausedActivity = false;
|
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()
|
private async Task<CallResult<bool>> ProcessReconnectAsync()
|
||||||
{
|
{
|
||||||
if (!_socket.IsOpen)
|
if (!_socket.IsOpen)
|
||||||
@@ -671,5 +731,36 @@ namespace CryptoExchange.Net.Sockets
|
|||||||
|
|
||||||
return await socketClient.SubscribeAndWaitAsync(this, socketSubscription.Request!, socketSubscription).ConfigureAwait(false);
|
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
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -18,6 +18,30 @@ 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)
|
Alternatively, sponsor me on Github using [Github Sponsors](https://github.com/sponsors/JKorf)
|
||||||
|
|
||||||
## Release notes
|
## 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
|
* Version 5.1.9 - 08 May 2022
|
||||||
* Added latency to the timesync calculation
|
* Added latency to the timesync calculation
|
||||||
* Small fix for exception in socket close handling
|
* Small fix for exception in socket close handling
|
||||||
|
|||||||
+4
-1
@@ -61,4 +61,7 @@ var client = new BinanceClient(new BinanceClientOptions
|
|||||||
BaseAddress = BinanceApiAddresses.TestNet.UsdFuturesRestClientAddress
|
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.
|
||||||
Reference in New Issue
Block a user