1
0
mirror of https://github.com/JKorf/CryptoExchange.Net.git synced 2026-08-14 01:42:55 +00:00

Compare commits

..

4 Commits

Author SHA1 Message Date
JKorf 99c331b389 Updated version 2022-07-10 16:41:14 +02:00
JKorf 70f8bd203a Finished up websocket refactoring 2022-07-10 16:36:00 +02:00
JKorf 89b517c936 wip 2022-07-08 20:24:58 +02:00
JKorf 91e33cc42c wip 2022-07-07 22:17:55 +02:00
10 changed files with 56 additions and 44 deletions
@@ -42,7 +42,7 @@ namespace CryptoExchange.Net.UnitTests
socket.CanConnect = canConnect; socket.CanConnect = canConnect;
//act //act
var connectResult = client.ConnectSocketSub(new SocketConnection(client, null, socket, null)); var connectResult = client.ConnectSocketSub(new SocketConnection(client, null, socket));
//assert //assert
Assert.IsTrue(connectResult.Success == canConnect); Assert.IsTrue(connectResult.Success == canConnect);
@@ -57,10 +57,10 @@ namespace CryptoExchange.Net.UnitTests
socket.ShouldReconnect = true; socket.ShouldReconnect = true;
socket.CanConnect = true; socket.CanConnect = true;
socket.DisconnectTime = DateTime.UtcNow; socket.DisconnectTime = DateTime.UtcNow;
var sub = new SocketConnection(client, null, socket, null); var sub = new SocketConnection(client, null, socket);
var rstEvent = new ManualResetEvent(false); var rstEvent = new ManualResetEvent(false);
JToken result = null; JToken result = null;
sub.AddSubscription(SocketSubscription.CreateForIdentifier(10, "TestHandler", true, false, (messageEvent) => sub.AddSubscription(SocketSubscription.CreateForIdentifier(10, "TestHandler", true, (messageEvent) =>
{ {
result = messageEvent.JsonData; result = messageEvent.JsonData;
rstEvent.Set(); rstEvent.Set();
@@ -85,10 +85,10 @@ namespace CryptoExchange.Net.UnitTests
socket.ShouldReconnect = true; socket.ShouldReconnect = true;
socket.CanConnect = true; socket.CanConnect = true;
socket.DisconnectTime = DateTime.UtcNow; socket.DisconnectTime = DateTime.UtcNow;
var sub = new SocketConnection(client, null, socket, null); var sub = new SocketConnection(client, null, socket);
var rstEvent = new ManualResetEvent(false); var rstEvent = new ManualResetEvent(false);
string original = null; string original = null;
sub.AddSubscription(SocketSubscription.CreateForIdentifier(10, "TestHandler", true, false, (messageEvent) => sub.AddSubscription(SocketSubscription.CreateForIdentifier(10, "TestHandler", true, (messageEvent) =>
{ {
original = messageEvent.OriginalData; original = messageEvent.OriginalData;
rstEvent.Set(); rstEvent.Set();
@@ -103,6 +103,34 @@ namespace CryptoExchange.Net.UnitTests
Assert.IsTrue(original == (enabled ? "{\"property\": 123}" : null)); Assert.IsTrue(original == (enabled ? "{\"property\": 123}" : null));
} }
[TestCase]
public void DisconnectedSocket_Should_Reconnect()
{
// arrange
bool reconnected = false;
var client = new TestSocketClient(new TestOptions() { ReconnectInterval = TimeSpan.Zero, LogLevel = LogLevel.Debug });
var socket = client.CreateSocket();
socket.ShouldReconnect = true;
socket.CanConnect = true;
socket.DisconnectTime = DateTime.UtcNow;
var sub = new SocketConnection(client, null, socket);
sub.ShouldReconnect = true;
client.ConnectSocketSub(sub);
var rstEvent = new ManualResetEvent(false);
sub.ConnectionRestored += (a) =>
{
reconnected = true;
rstEvent.Set();
};
// act
socket.InvokeClose();
rstEvent.WaitOne(1000);
// assert
Assert.IsTrue(reconnected);
}
[TestCase()] [TestCase()]
public void UnsubscribingStream_Should_CloseTheSocket() public void UnsubscribingStream_Should_CloseTheSocket()
{ {
@@ -110,11 +138,9 @@ namespace CryptoExchange.Net.UnitTests
var client = new TestSocketClient(new TestOptions() { ReconnectInterval = TimeSpan.Zero, LogLevel = LogLevel.Debug }); var client = new TestSocketClient(new TestOptions() { ReconnectInterval = TimeSpan.Zero, LogLevel = LogLevel.Debug });
var socket = client.CreateSocket(); var socket = client.CreateSocket();
socket.CanConnect = true; socket.CanConnect = true;
var sub = new SocketConnection(client, null, socket, null); var sub = new SocketConnection(client, null, socket);
client.ConnectSocketSub(sub); client.ConnectSocketSub(sub);
var us = SocketSubscription.CreateForIdentifier(10, "Test", true, false, (e) => { }); var ups = new UpdateSubscription(sub, SocketSubscription.CreateForIdentifier(10, "Test", true, (e) => {}));
var ups = new UpdateSubscription(sub, us);
sub.AddSubscription(us);
// act // act
client.UnsubscribeAsync(ups).Wait(); client.UnsubscribeAsync(ups).Wait();
@@ -132,8 +158,8 @@ namespace CryptoExchange.Net.UnitTests
var socket2 = client.CreateSocket(); var socket2 = client.CreateSocket();
socket1.CanConnect = true; socket1.CanConnect = true;
socket2.CanConnect = true; socket2.CanConnect = true;
var sub1 = new SocketConnection(client, null, socket1, null); var sub1 = new SocketConnection(client, null, socket1);
var sub2 = new SocketConnection(client, null, socket2, null); var sub2 = new SocketConnection(client, null, socket2);
client.ConnectSocketSub(sub1); client.ConnectSocketSub(sub1);
client.ConnectSocketSub(sub2); client.ConnectSocketSub(sub2);
@@ -152,7 +178,7 @@ namespace CryptoExchange.Net.UnitTests
var client = new TestSocketClient(new TestOptions() { ReconnectInterval = TimeSpan.Zero, LogLevel = LogLevel.Debug }); var client = new TestSocketClient(new TestOptions() { ReconnectInterval = TimeSpan.Zero, LogLevel = LogLevel.Debug });
var socket = client.CreateSocket(); var socket = client.CreateSocket();
socket.CanConnect = false; socket.CanConnect = false;
var sub = new SocketConnection(client, null, socket, null); var sub = new SocketConnection(client, null, socket);
// act // act
var connectResult = client.ConnectSocketSub(sub); var connectResult = client.ConnectSocketSub(sub);
@@ -13,8 +13,6 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
public bool Connected { get; set; } public bool Connected { get; set; }
public event Action OnClose; public event Action OnClose;
public event Action OnReconnected;
public event Action OnReconnecting;
public event Action<string> OnMessage; public event Action<string> OnMessage;
public event Action<Exception> OnError; public event Action<Exception> OnError;
public event Action OnOpen; public event Action OnOpen;
@@ -95,7 +93,6 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
{ {
Connected = false; Connected = false;
DisconnectTime = DateTime.UtcNow; DisconnectTime = DateTime.UtcNow;
Reconnecting = true;
OnClose?.Invoke(); OnClose?.Invoke();
} }
@@ -118,6 +115,11 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
{ {
OnError?.Invoke(error); OnError?.Invoke(error);
} }
public Task ReconnectAsync() => Task.CompletedTask;
public async Task ProcessAsync()
{
while (Connected)
await Task.Delay(50);
}
} }
} }
@@ -22,13 +22,13 @@ namespace CryptoExchange.Net.UnitTests.TestImplementations
{ {
SubClient = new TestSubSocketClient(exchangeOptions, exchangeOptions.SubOptions); SubClient = new TestSubSocketClient(exchangeOptions, exchangeOptions.SubOptions);
SocketFactory = new Mock<IWebsocketFactory>().Object; SocketFactory = new Mock<IWebsocketFactory>().Object;
Mock.Get(SocketFactory).Setup(f => f.CreateWebsocket(It.IsAny<Log>(), It.IsAny<WebSocketParameters>())).Returns(new TestSocket()); Mock.Get(SocketFactory).Setup(f => f.CreateWebsocket(It.IsAny<Log>(), It.IsAny<string>())).Returns(new TestSocket());
} }
public TestSocket CreateSocket() public TestSocket CreateSocket()
{ {
Mock.Get(SocketFactory).Setup(f => f.CreateWebsocket(It.IsAny<Log>(), It.IsAny<WebSocketParameters>())).Returns(new TestSocket()); Mock.Get(SocketFactory).Setup(f => f.CreateWebsocket(It.IsAny<Log>(), It.IsAny<string>())).Returns(new TestSocket());
return (TestSocket)CreateSocket("https://localhost:123/"); return (TestSocket)CreateSocket("123");
} }
public CallResult<bool> ConnectSocketSub(SocketConnection sub) public CallResult<bool> ConnectSocketSub(SocketConnection sub)
+1 -1
View File
@@ -320,7 +320,7 @@ namespace CryptoExchange.Net
responseStream.Close(); responseStream.Close();
response.Close(); response.Close();
var parseResult = ValidateJson(data); var parseResult = ValidateJson(data);
var error = parseResult.Success ? ParseErrorResponse(parseResult.Data) : new ServerError(data)!; var error = parseResult.Success ? ParseErrorResponse(parseResult.Data) : parseResult.Error!;
if(error.Code == null || error.Code == 0) if(error.Code == null || error.Code == 0)
error.Code = (int)response.StatusCode; error.Code = (int)response.StatusCode;
return new WebCallResult<T>(statusCode, headers, sw.Elapsed, data, request.Uri.ToString(), request.Content, request.Method, request.GetHeaders(), default, error); return new WebCallResult<T>(statusCode, headers, sw.Elapsed, data, request.Uri.ToString(), request.Content, request.Method, request.GetHeaders(), default, error);
@@ -392,14 +392,11 @@ namespace CryptoExchange.Net
if (!authenticated || socket.Authenticated) if (!authenticated || socket.Authenticated)
return new CallResult<bool>(true); return new CallResult<bool>(true);
log.Write(LogLevel.Debug, $"Attempting to authenticate {socket.SocketId}");
var result = await AuthenticateSocketAsync(socket).ConfigureAwait(false); var result = await AuthenticateSocketAsync(socket).ConfigureAwait(false);
if (!result) if (!result)
{ {
await socket.CloseAsync().ConfigureAwait(false);
log.Write(LogLevel.Warning, $"Socket {socket.SocketId} authentication failed"); log.Write(LogLevel.Warning, $"Socket {socket.SocketId} authentication failed");
if(socket.Connected)
await socket.CloseAsync().ConfigureAwait(false);
result.Error!.Message = "Authentication failed: " + result.Error.Message; result.Error!.Message = "Authentication failed: " + result.Error.Message;
return new CallResult<bool>(result.Error); return new CallResult<bool>(result.Error);
} }
+4 -4
View File
@@ -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.2.1</PackageVersion> <PackageVersion>5.2.0</PackageVersion>
<AssemblyVersion>5.2.1</AssemblyVersion> <AssemblyVersion>5.2.0</AssemblyVersion>
<FileVersion>5.2.1</FileVersion> <FileVersion>5.2.0</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.2.1 - Fixed socket reconnect issue, Fixed `message not handled` messages after unsubscribing, Fixed error returning for non-json error responses</PackageReleaseNotes> <PackageReleaseNotes>5.2.0 - Refactored websocket code, removed some clutter and simplified, Added ReconnectAsync and GetSubscriptionsState methods on socket clients</PackageReleaseNotes>
<Nullable>enable</Nullable> <Nullable>enable</Nullable>
<LangVersion>9.0</LangVersion> <LangVersion>9.0</LangVersion>
<PackageLicenseExpression>MIT</PackageLicenseExpression> <PackageLicenseExpression>MIT</PackageLicenseExpression>
@@ -250,7 +250,7 @@ namespace CryptoExchange.Net.Sockets
/// <inheritdoc /> /// <inheritdoc />
public virtual async Task ReconnectAsync() public virtual async Task ReconnectAsync()
{ {
if (_processState != ProcessState.Processing && IsOpen) if (_processState != ProcessState.Processing)
return; return;
_log.Write(LogLevel.Debug, $"Socket {Id} reconnect requested"); _log.Write(LogLevel.Debug, $"Socket {Id} reconnect requested");
@@ -385,7 +385,7 @@ namespace CryptoExchange.Net.Sockets
if (!subscriptions.Contains(subscription)) if (!subscriptions.Contains(subscription))
return; return;
subscription.Closed = true; subscriptions.Remove(subscription);
} }
if (Status == SocketStatus.Closing || Status == SocketStatus.Closed || Status == SocketStatus.Disposed) if (Status == SocketStatus.Closing || Status == SocketStatus.Closed || Status == SocketStatus.Disposed)
@@ -407,7 +407,7 @@ namespace CryptoExchange.Net.Sockets
return; return;
} }
shouldCloseConnection = subscriptions.All(r => !r.UserSubscription || r.Closed); shouldCloseConnection = subscriptions.All(r => !r.UserSubscription);
if (shouldCloseConnection) if (shouldCloseConnection)
Status = SocketStatus.Closing; Status = SocketStatus.Closing;
} }
@@ -417,9 +417,6 @@ namespace CryptoExchange.Net.Sockets
log.Write(LogLevel.Debug, $"Socket {SocketId} closing as there are no more subscriptions"); log.Write(LogLevel.Debug, $"Socket {SocketId} closing as there are no more subscriptions");
await CloseAsync().ConfigureAwait(false); await CloseAsync().ConfigureAwait(false);
} }
lock (subscriptionLock)
subscriptions.Remove(subscription);
} }
/// <summary> /// <summary>
@@ -48,11 +48,6 @@ namespace CryptoExchange.Net.Sockets
/// </summary> /// </summary>
public bool Authenticated { get; set; } public bool Authenticated { get; set; }
/// <summary>
/// Whether we're closing this subscription and a socket connection shouldn't be kept open for it
/// </summary>
public bool Closed { get; set; }
/// <summary> /// <summary>
/// Cancellation token registration, should be disposed when subscription is closed. Used for closing the subscription with /// Cancellation token registration, should be disposed when subscription is closed. Used for closing the subscription with
/// a provided cancelation token /// a provided cancelation token
-5
View File
@@ -18,11 +18,6 @@ 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.2.1 - 16 Jul 2022
* Fixed socket reconnect issue
* Fixed `message not handled` messages after unsubscribing
* Fixed error returning for non-json error responses
* Version 5.2.0 - 10 Jul 2022 * Version 5.2.0 - 10 Jul 2022
* Refactored websocket code, removed some clutter and simplified * Refactored websocket code, removed some clutter and simplified
* Added ReconnectAsync and GetSubscriptionsState methods on socket clients * Added ReconnectAsync and GetSubscriptionsState methods on socket clients