Websocket refactoring

Binance.Net.UnitTests/BinanceClientTests.cs

@@ -18,10 +18,11 @@
 using System.Reflection;
 using System.Diagnostics;
 using Binance.Net.Objects.Models.Spot;
-using CryptoExchange.Net.Sockets;
 using Binance.Net.Clients;
 using Binance.Net.Clients.SpotApi;
+using Binance.Net.ExtensionMethods;
 using Binance.Net.Objects.Options;
+using CryptoExchange.Net.Objects.Sockets;
 
 namespace Binance.Net.UnitTests
 {

Binance.Net.UnitTests/BinanceSocketClientTests.cs

@@ -11,21 +11,25 @@
 using Microsoft.Extensions.Logging;
 using Binance.Net.Objects.Models.Futures.Socket;
 using Binance.Net.Objects.Options;
+using Binance.Net.Objects.Sockets;
+using CryptoExchange.Net;
 
 namespace Binance.Net.UnitTests
 {
     [TestFixture()]
     public class BinanceNetTest
     {
         [TestCase()]
-        public void SubscribingToKlineStream_Should_TriggerWhenKlineStreamMessageIsReceived()
+        public async Task SubscribingToKlineStream_Should_TriggerWhenKlineStreamMessageIsReceived()
         {
             // arrange
             var socket = new TestSocket();
             var client = TestHelpers.CreateSocketClient(socket);
 
             IBinanceStreamKlineData result = null;
-            client.SpotApi.ExchangeData.SubscribeToKlineUpdatesAsync("ETHBTC", KlineInterval.OneMinute, (test) => result = test.Data);
+            var subTask = client.SpotApi.ExchangeData.SubscribeToKlineUpdatesAsync("ETHBTC", KlineInterval.OneMinute, (test) => result = test.Data);
+            await socket.InvokeMessage(new BinanceSocketQueryResponse { Id = ExchangeHelpers.LastId() - 1 });
+            await subTask;
 
             var data = new BinanceCombinedStream<BinanceStreamKlineData>()
             {
@@ -58,7 +62,7 @@ public void SubscribingToKlineStream_Should_TriggerWhenKlineStreamMessageIsRecei
             };
 
             // act
-            socket.InvokeMessage(data);
+            await socket.InvokeMessage(data);
 
             // assert
             Assert.IsNotNull(result);
@@ -67,14 +71,16 @@ public void SubscribingToKlineStream_Should_TriggerWhenKlineStreamMessageIsRecei
         }
 
         [TestCase()]
-        public void SubscribingToContinuousKlineStream_Should_TriggerWhenContinuousKlineStreamMessageIsReceived()
+        public async Task SubscribingToContinuousKlineStream_Should_TriggerWhenContinuousKlineStreamMessageIsReceived()
         {
             // arrange
             var socket = new TestSocket();
             var client = TestHelpers.CreateSocketClient(socket);
 
             IBinanceStreamKlineData result = null;
-            client.UsdFuturesApi.SubscribeToContinuousContractKlineUpdatesAsync("ETHBTC", ContractType.Perpetual, KlineInterval.OneMinute, (test) => result = test.Data);
+            var subTask = client.UsdFuturesApi.SubscribeToContinuousContractKlineUpdatesAsync("ETHBTC", ContractType.Perpetual, KlineInterval.OneMinute, (test) => result = test.Data);
+            await socket.InvokeMessage(new BinanceSocketQueryResponse { Id = ExchangeHelpers.LastId() - 1 });
+            await subTask;
 
             var data = new BinanceCombinedStream<BinanceStreamContinuousKlineData>()
             {
@@ -108,7 +114,7 @@ public void SubscribingToContinuousKlineStream_Should_TriggerWhenContinuousKline
             };
 
             // act
-            socket.InvokeMessage(data);
+            await socket.InvokeMessage(data);
 
             // assert
             Assert.IsNotNull(result);
@@ -124,7 +130,9 @@ public async Task SubscribingToSymbolTicker_Should_TriggerWhenSymbolTickerStream
             var client = TestHelpers.CreateSocketClient(socket);
 
             IBinanceTick result = null;
-            await client.SpotApi.ExchangeData.SubscribeToTickerUpdatesAsync("ETHBTC", (test) => result = test.Data);
+            var subTask = client.SpotApi.ExchangeData.SubscribeToTickerUpdatesAsync("ETHBTC", (test) => result = test.Data);
+            await socket.InvokeMessage(new BinanceSocketQueryResponse { Id = ExchangeHelpers.LastId() - 1 });
+            await subTask;
 
             var data = new BinanceCombinedStream<BinanceStreamTick>()
             {
@@ -145,7 +153,7 @@ public async Task SubscribingToSymbolTicker_Should_TriggerWhenSymbolTickerStream
             };
 
             // act
-            socket.InvokeMessage(data);
+            await socket.InvokeMessage(data);
 
             // assert
             Assert.IsNotNull(result);
@@ -160,7 +168,9 @@ public async Task SubscribingToAllSymbolTicker_Should_TriggerWhenAllSymbolTicker
             var client = TestHelpers.CreateSocketClient(socket);
 
             IBinanceTick[] result = null;
-            await client.SpotApi.ExchangeData.SubscribeToAllTickerUpdatesAsync((test) => result = test.Data.ToArray());
+            var subTask = client.SpotApi.ExchangeData.SubscribeToAllTickerUpdatesAsync((test) => result = test.Data.ToArray());
+            await socket.InvokeMessage(new BinanceSocketQueryResponse { Id = ExchangeHelpers.LastId() - 1 });
+            await subTask;
 
             var data = new BinanceCombinedStream<BinanceStreamTick[]>
             {
@@ -185,7 +195,7 @@ public async Task SubscribingToAllSymbolTicker_Should_TriggerWhenAllSymbolTicker
             };
 
             // act
-            socket.InvokeMessage(data);
+            await socket.InvokeMessage(data);
 
             // assert
             Assert.IsNotNull(result);
@@ -200,7 +210,9 @@ public async Task SubscribingToTradeStream_Should_TriggerWhenTradeStreamMessageI
             var client = TestHelpers.CreateSocketClient(socket);
 
             BinanceStreamTrade result = null;
-            await client.SpotApi.ExchangeData.SubscribeToTradeUpdatesAsync("ETHBTC", (test) => result = test.Data);
+            var subTask = client.SpotApi.ExchangeData.SubscribeToTradeUpdatesAsync("ETHBTC", (test) => result = test.Data);
+            await socket.InvokeMessage(new BinanceSocketQueryResponse { Id = ExchangeHelpers.LastId() - 1 });
+            await subTask;
 
             var data = new BinanceCombinedStream<BinanceStreamTrade>()
             {
@@ -220,7 +232,7 @@ public async Task SubscribingToTradeStream_Should_TriggerWhenTradeStreamMessageI
             };
 
             // act
-            socket.InvokeMessage(data);
+            await socket.InvokeMessage(data);
 
             // assert
             Assert.IsNotNull(result);
@@ -235,7 +247,9 @@ public async Task SubscribingToUserStream_Should_TriggerWhenAccountUpdateStreamM
             var client = TestHelpers.CreateSocketClient(socket);
 
             BinanceStreamBalanceUpdate result = null;
-            await client.SpotApi.Account.SubscribeToUserDataUpdatesAsync("test", null, null, null, (test) => result = test.Data);
+            var subTask = client.SpotApi.Account.SubscribeToUserDataUpdatesAsync("test", null, null, null, (test) => result = test.Data);
+            await socket.InvokeMessage(new BinanceSocketQueryResponse { Id = ExchangeHelpers.LastId() - 1 });
+            await subTask;
 
             var data = new BinanceCombinedStream<BinanceStreamBalanceUpdate>
             {
@@ -251,22 +265,24 @@ public async Task SubscribingToUserStream_Should_TriggerWhenAccountUpdateStreamM
             };
 
             // act
-            socket.InvokeMessage(data);
+            await socket.InvokeMessage(data);
 
             // assert
             Assert.IsNotNull(result);
             Assert.IsTrue(TestHelpers.AreEqual(data.Data, result, "ListenKey"));
         }
 
         [TestCase()]
-        public void SubscribingToUserStream_Should_TriggerWhenOcoOrderUpdateStreamMessageIsReceived()
+        public async Task SubscribingToUserStream_Should_TriggerWhenOcoOrderUpdateStreamMessageIsReceived()
         {
             // arrange
             var socket = new TestSocket();
             var client = TestHelpers.CreateSocketClient(socket);
 
             BinanceStreamOrderList result = null;
-            client.SpotApi.Account.SubscribeToUserDataUpdatesAsync("test", null, (test) => result = test.Data, null, null);
+            var subTask = client.SpotApi.Account.SubscribeToUserDataUpdatesAsync("test", null, (test) => result = test.Data, null, null);
+            await socket.InvokeMessage(new BinanceSocketQueryResponse { Id = ExchangeHelpers.LastId() - 1 });
+            await subTask;
 
             var data = new BinanceCombinedStream<BinanceStreamOrderList>
             {
@@ -301,7 +317,7 @@ public void SubscribingToUserStream_Should_TriggerWhenOcoOrderUpdateStreamMessag
             };
 
             // act
-            socket.InvokeMessage(data);
+            await socket.InvokeMessage(data);
 
             // assert
             Assert.IsNotNull(result);
@@ -311,14 +327,16 @@ public void SubscribingToUserStream_Should_TriggerWhenOcoOrderUpdateStreamMessag
         }
 
         [TestCase()]
-        public void SubscribingToUserStream_Should_TriggerWhenOrderUpdateStreamMessageIsReceived()
+        public async Task SubscribingToUserStream_Should_TriggerWhenOrderUpdateStreamMessageIsReceived()
         {
             // arrange
             var socket = new TestSocket();
             var client = TestHelpers.CreateSocketClient(socket);
 
             BinanceStreamOrderUpdate result = null;
-            client.SpotApi.Account.SubscribeToUserDataUpdatesAsync("test", (test) => result = test.Data, null, null, null);
+            var subTask = client.SpotApi.Account.SubscribeToUserDataUpdatesAsync("test", (test) => result = test.Data, null, null, null);
+            await socket.InvokeMessage(new BinanceSocketQueryResponse { Id = ExchangeHelpers.LastId() - 1 });
+            await subTask;
 
             var data = new BinanceCombinedStream<BinanceStreamOrderUpdate>
             {
@@ -351,7 +369,7 @@ public void SubscribingToUserStream_Should_TriggerWhenOrderUpdateStreamMessageIs
             };
 
             // act
-            socket.InvokeMessage(data);
+            await socket.InvokeMessage(data);
 
             // assert
             Assert.IsNotNull(result);

Binance.Net.UnitTests/TestImplementations/TestHelpers.cs

@@ -14,7 +14,7 @@
 using Binance.Net.Objects.Models.Futures;
 using Binance.Net.Objects.Options;
 using CryptoExchange.Net.Interfaces;
-using CryptoExchange.Net.Sockets;
+using CryptoExchange.Net.Objects.Sockets;
 using Microsoft.Extensions.Logging;
 using Moq;
 using Newtonsoft.Json;

Binance.Net.UnitTests/TestImplementations/TestSocket.cs

@@ -1,4 +1,6 @@
 using System;
+using System.IO;
+using System.Net.WebSockets;
 using System.Security.Authentication;
 using System.Text;
 using System.Threading.Tasks;
@@ -10,18 +12,18 @@ namespace Binance.Net.UnitTests.TestImplementations
 {
     public class TestSocket: IWebsocket
     {
-        public bool CanConnect { get; set; }
+        public bool CanConnect { get; set; } = true;
         public bool Connected { get; set; }
 
-        public event Action OnClose;
-        public event Action<string> OnMessage;
+        public event Func<Task> OnClose;
 #pragma warning disable 0067
-        public event Action<int> OnRequestSent;
-        public event Action<Exception> OnError;
-        public event Action OnOpen;
-        public event Action OnReconnecting;
-        public event Action OnReconnected;
+        public event Func<Task> OnReconnected;
+        public event Func<Task> OnReconnecting;
+        public event Func<Exception, Task> OnError;
 #pragma warning restore 0067
+        public event Func<int, Task> OnRequestSent;
+        public event Func<WebSocketMessageType, Stream, Task> OnStreamMessage;
+        public event Func<Task> OnOpen;
 
         public int Id { get; }
         public bool ShouldReconnect { get; set; }
@@ -90,14 +92,16 @@ public void InvokeOpen()
             OnOpen?.Invoke();
         }
 
-        public void InvokeMessage(string data)
+        public async Task InvokeMessage(string data)
         {
-            OnMessage?.Invoke(data);
+            var stream = new MemoryStream(Encoding.UTF8.GetBytes(data));
+            await OnStreamMessage?.Invoke(WebSocketMessageType.Text, stream);
         }
 
-        public void InvokeMessage<T>(T data)
+        public async Task InvokeMessage<T>(T data)
         {
-            OnMessage?.Invoke(JsonConvert.SerializeObject(data));
+            var stream = new MemoryStream(Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(data)));
+            await OnStreamMessage?.Invoke(WebSocketMessageType.Text, stream);
         }
 
         public void SetProxy(ApiProxy proxy)

Binance.Net/Binance.Net.csproj

@@ -1,15 +1,15 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
     <TargetFrameworks>netstandard2.0;netstandard2.1</TargetFrameworks>
-    <LangVersion>8.0</LangVersion>
+    <LangVersion>10.0</LangVersion>
     <Nullable>enable</Nullable>
   </PropertyGroup>
   <PropertyGroup>
     <PackageId>Binance.Net</PackageId>
     <Authors>JKorf</Authors>
-    <PackageVersion>9.4.0</PackageVersion>
-    <AssemblyVersion>9.4.0</AssemblyVersion>
-    <FileVersion>9.4.0</FileVersion>
+    <PackageVersion>9.5.0-beta1</PackageVersion>
+    <AssemblyVersion>9.5.0</AssemblyVersion>
+    <FileVersion>9.5.0</FileVersion>
     <Description>Binance.Net is a .Net wrapper for the Binance API, including Binance Futures. It includes all features the API provides, REST API and Websocket, using clear and readable objects including but not limited to Reading market info, Placing and managing orders and Reading balances and funds</Description>
     <PackageRequireLicenseAcceptance>false</PackageRequireLicenseAcceptance>
     <PackageTags>Binance Binance.Net C# .Net CryptoCurrency Exchange API wrapper</PackageTags>
@@ -21,7 +21,7 @@
     <PackageLicenseExpression>MIT</PackageLicenseExpression>
     <NeutralLanguage>en</NeutralLanguage>
     <GeneratePackageOnBuild>true</GeneratePackageOnBuild>
-    <PackageReleaseNotes>9.4.0 - Added futures Trading.GetFundingInfoAsync endpoint, Added autoRepayAtCancel and selfTradePreventionMode parameters to SpotApi.Trading.PlaceMarginOCOOrderAsync, Added MarkPrice to futures funding rate models, Updated and combined futures order models, Updated withdraw history model</PackageReleaseNotes>
+    <PackageReleaseNotes>9.5.0-beta1 - Updated CryptoExchange.Net and implemented reworked websocket message handling. For release notes for the CryptoExchange.Net base library see: https://github.com/JKorf/CryptoExchange.Net/tree/beta?tab=readme-ov-file#release-notes, Fixed issue in DI registration causing http client to not be correctly injected</PackageReleaseNotes>
   </PropertyGroup>
   <PropertyGroup Label="Deterministic Build" Condition="'$(Configuration)' == 'Release'">
     <PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -48,8 +48,8 @@
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
-    <PackageReference Include="CryptoExchange.Net" Version="6.2.3" />
-    <PackageReference Include="Microsoft.CodeAnalysis.NetAnalyzers" Version="6.0.0">
+    <PackageReference Include="CryptoExchange.Net" Version="7.0.0" />
+    <PackageReference Include="Microsoft.CodeAnalysis.NetAnalyzers" Version="8.0.0">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>