Merge WebSocketAsync into master

.gitignore

@@ -263,4 +263,5 @@ __pycache__/
 # Trader binary and log files
 *.bin
 *.log*
-ExchangeSharpConsole/Properties/launchSettings.json
+launchSettings.json
+**/PublishProfiles/*

ExchangeSharp/API/Common/BaseAPI.cs

@@ -496,7 +496,7 @@ public async Task<T> MakeJsonRequestAsync<T>(string url, string baseUrl = null,
         /// <param name="messageCallback">Callback for messages</param>
         /// <param name="connectCallback">Connect callback</param>
         /// <returns>Web socket - dispose of the wrapper to shutdown the socket</returns>
-        public IWebSocket ConnectWebSocket
+        public Task<IWebSocket> ConnectWebSocket
         (
             string url,
             Func<IWebSocket, byte[], Task> messageCallback,
@@ -524,7 +524,7 @@ public IWebSocket ConnectWebSocket
                 wrapper.Disconnected += disconnectCallback;
             }
             wrapper.Start();
-            return wrapper;
+            return Task.FromResult<IWebSocket>(wrapper);
         }
 
         /// <summary>

ExchangeSharp/API/Exchanges/Binance/ExchangeBinanceAPI.cs

@@ -241,7 +241,7 @@ protected override async Task<IEnumerable<KeyValuePair<string, ExchangeTicker>>>
 			return tickers;
 		}
 
-		protected override IWebSocket OnGetTickersWebSocket(Action<IReadOnlyCollection<KeyValuePair<string, ExchangeTicker>>> callback, params string[] symbols)
+		protected override Task<IWebSocket> OnGetTickersWebSocket(Action<IReadOnlyCollection<KeyValuePair<string, ExchangeTicker>>> callback, params string[] symbols)
 		{
 			return ConnectWebSocket("/stream?streams=!ticker@arr", (_socket, msg) =>
 			{
@@ -261,7 +261,7 @@ protected override IWebSocket OnGetTickersWebSocket(Action<IReadOnlyCollection<K
 			});
 		}
 
-		protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
+		protected override Task<IWebSocket> OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
 		{
 			/*
 	    {
@@ -297,7 +297,7 @@ protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, Exc
 			});
 		}
 
-		protected override IWebSocket OnGetDeltaOrderBookWebSocket(Action<ExchangeOrderBook> callback, int maxCount = 20, params string[] marketSymbols)
+		protected override Task<IWebSocket> OnGetDeltaOrderBookWebSocket(Action<ExchangeOrderBook> callback, int maxCount = 20, params string[] marketSymbols)
 		{
 			if (marketSymbols == null || marketSymbols.Length == 0)
 			{
@@ -1060,7 +1060,7 @@ protected override async Task<IEnumerable<ExchangeTransaction>> OnGetDepositHist
 			return transactions;
 		}
 
-		protected override IWebSocket OnUserDataWebSocket(Action<object> callback, string listenKey)
+		protected override Task<IWebSocket> OnUserDataWebSocket(Action<object> callback, string listenKey)
 		{
 			return ConnectWebSocket($"/ws/{listenKey}", (_socket, msg) =>
 			{

ExchangeSharp/API/Exchanges/BitMEX/ExchangeBitMEXAPI.cs

@@ -224,7 +224,7 @@ protected override async Task<IEnumerable<ExchangeMarket>> OnGetMarketSymbolsMet
             return markets;
         }
 
-        protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
+        protected override Task<IWebSocket> OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
         {
             /*
 {"table":"trade","action":"partial","keys":[],
@@ -270,7 +270,7 @@ protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, Exc
             });
         }
 
-        protected override IWebSocket OnGetDeltaOrderBookWebSocket(Action<ExchangeOrderBook> callback, int maxCount = 20, params string[] marketSymbols)
+        protected override Task<IWebSocket> OnGetDeltaOrderBookWebSocket(Action<ExchangeOrderBook> callback, int maxCount = 20, params string[] marketSymbols)
         {
             /*
 {"info":"Welcome to the BitMEX Realtime API.","version":"2018-06-29T18:05:14.000Z","timestamp":"2018-07-05T14:22:26.267Z","docs":"https://www.bitmex.com/app/wsAPI","limit":{"remaining":39}}

ExchangeSharp/API/Exchanges/Bitfinex/ExchangeBitfinexAPI.cs

@@ -213,7 +213,7 @@ protected override async Task<IEnumerable<KeyValuePair<string, ExchangeTicker>>>
             return tickers;
         }
 
-        protected override IWebSocket OnGetTickersWebSocket(Action<IReadOnlyCollection<KeyValuePair<string, ExchangeTicker>>> callback, params string[] marketSymbols)
+        protected override Task<IWebSocket> OnGetTickersWebSocket(Action<IReadOnlyCollection<KeyValuePair<string, ExchangeTicker>>> callback, params string[] marketSymbols)
         {
             Dictionary<int, string> channelIdToSymbol = new Dictionary<int, string>();
             return ConnectWebSocket(string.Empty, (_socket, msg) =>
@@ -251,7 +251,7 @@ protected override IWebSocket OnGetTickersWebSocket(Action<IReadOnlyCollection<K
             });
         }
 
-        protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
+        protected override Task<IWebSocket> OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
         {
             Dictionary<int, string> channelIdToSymbol = new Dictionary<int, string>();
 			if (marketSymbols == null || marketSymbols.Length == 0)
@@ -534,7 +534,7 @@ protected override async Task<IEnumerable<ExchangeOrderResult>> OnGetCompletedOr
             return await GetOrderDetailsInternalV1(new string[] { marketSymbol }, afterDate);
         }
 
-        protected override IWebSocket OnGetCompletedOrderDetailsWebSocket(Action<ExchangeOrderResult> callback)
+        protected override Task<IWebSocket> OnGetCompletedOrderDetailsWebSocket(Action<ExchangeOrderResult> callback)
         {
             return ConnectWebSocket(string.Empty, (_socket, msg) =>
             {

ExchangeSharp/API/Exchanges/Bitstamp/ExchangeBitstampAPI.cs

@@ -507,7 +507,7 @@ protected override async Task<ExchangeWithdrawalResponse> OnWithdrawAsync(Exchan
             };
         }
 
-		protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
+		protected override Task<IWebSocket> OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
 		{
 			if (marketSymbols == null || marketSymbols.Length == 0)
 			{

ExchangeSharp/API/Exchanges/Bittrex/ExchangeBittrexAPI_WebSocket.cs

@@ -50,15 +50,12 @@ public BittrexWebSocketManager() : base("https://socket.bittrex.com/signalr", "c
             /// Subscribe to all market summaries
             /// </summary>
             /// <param name="callback">Callback</param>
+            /// <param name="marketSymbols">Symbols</param>
             /// <returns>IDisposable to close the socket</returns>
-            public IWebSocket SubscribeToSummaryDeltas(Action<string> callback)
+            public async Task<IWebSocket> SubscribeToSummaryDeltas(Func<string, Task> callback, params string[] marketSymbols)
             {
                 SignalrManager.SignalrSocketConnection conn = new SignalrManager.SignalrSocketConnection(this);
-                Task.Run(async () => await conn.OpenAsync("uS", (s) =>
-                {
-                    callback(s);
-                    return Task.CompletedTask;
-                }));
+                await conn.OpenAsync("uS", callback);
                 return conn;
             }
 
@@ -68,27 +65,29 @@ public IWebSocket SubscribeToSummaryDeltas(Action<string> callback)
             /// <param name="callback">Callback</param>
             /// <param name="marketSymbols">The market symbols to subscribe to</param>
             /// <returns>IDisposable to close the socket</returns>
-            public IWebSocket SubscribeToExchangeDeltas(Func<string, Task> callback, params string[] marketSymbols)
+            public async Task<IWebSocket> SubscribeToExchangeDeltas(Func<string, Task> callback, params string[] marketSymbols)
             {
                 SignalrManager.SignalrSocketConnection conn = new SignalrManager.SignalrSocketConnection(this);
                 List<object[]> paramList = new List<object[]>();
                 foreach (string marketSymbol in marketSymbols)
                 {
                     paramList.Add(new object[] { marketSymbol });
                 }
-                Task.Run(async () => await conn.OpenAsync("uE", async (s) =>
-                {
-                    await callback(s);
-                }, 0, paramList.ToArray()));
+                await conn.OpenAsync("uE", callback, 0, paramList.ToArray());
                 return conn;
             }
         }
 
         private BittrexWebSocketManager webSocket;
 
-        protected override IWebSocket OnGetTickersWebSocket(Action<IReadOnlyCollection<KeyValuePair<string, ExchangeTicker>>> callback, params string[] symbols)
+        protected override Task<IWebSocket> OnGetTickersWebSocket(Action<IReadOnlyCollection<KeyValuePair<string, ExchangeTicker>>> callback, params string[] marketSymbols)
         {
-            void innerCallback(string json)
+            HashSet<string> filter = new HashSet<string>();
+            foreach (string marketSymbol in marketSymbols)
+            {
+                filter.Add(marketSymbol);
+            }
+            Task innerCallback(string json)
             {
                 #region sample json
                 /*
@@ -122,6 +121,10 @@ void innerCallback(string json)
                 foreach (JToken ticker in token)
                 {
                     string marketName = ticker["M"].ToStringInvariant();
+                    if (filter.Count != 0 && !filter.Contains(marketName))
+                    {
+                        continue;
+                    }
                     var (baseCurrency, quoteCurrency) = ExchangeMarketSymbolToCurrencies(marketName);
                     decimal last = ticker["l"].ConvertInvariant<decimal>();
                     decimal ask = ticker["A"].ConvertInvariant<decimal>();
@@ -147,11 +150,12 @@ void innerCallback(string json)
                     freshTickers[marketName] = t;
                 }
                 callback(freshTickers);
+                return Task.CompletedTask;
             }
-            return new BittrexWebSocketManager().SubscribeToSummaryDeltas(innerCallback);
+            return new BittrexWebSocketManager().SubscribeToSummaryDeltas(innerCallback, marketSymbols);
         }
 
-        protected override IWebSocket OnGetDeltaOrderBookWebSocket
+        protected override Task<IWebSocket> OnGetDeltaOrderBookWebSocket
         (
             Action<ExchangeOrderBook> callback,
             int maxCount = 20,
@@ -222,7 +226,7 @@ Task innerCallback(string json)
             return new BittrexWebSocketManager().SubscribeToExchangeDeltas(innerCallback, marketSymbols);
         }
 
-		protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
+		protected override Task<IWebSocket> OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
 		{
 			if (marketSymbols == null || marketSymbols.Length == 0)
 			{

ExchangeSharp/API/Exchanges/Coinbase/ExchangeCoinbaseAPI.cs

@@ -293,7 +293,7 @@ protected override async Task<IEnumerable<KeyValuePair<string, ExchangeTicker>>>
             }
         }
 
-        protected override IWebSocket OnGetDeltaOrderBookWebSocket(Action<ExchangeOrderBook> callback, int maxCount = 20, params string[] marketSymbols)
+        protected override Task<IWebSocket> OnGetDeltaOrderBookWebSocket(Action<ExchangeOrderBook> callback, int maxCount = 20, params string[] marketSymbols)
         {
             return ConnectWebSocket(string.Empty, (_socket, msg) =>
             {
@@ -362,7 +362,7 @@ protected override IWebSocket OnGetDeltaOrderBookWebSocket(Action<ExchangeOrderB
             });
         }
 
-        protected override IWebSocket OnGetTickersWebSocket(Action<IReadOnlyCollection<KeyValuePair<string, ExchangeTicker>>> callback, params string[] marketSymbols)
+        protected override Task<IWebSocket> OnGetTickersWebSocket(Action<IReadOnlyCollection<KeyValuePair<string, ExchangeTicker>>> callback, params string[] marketSymbols)
         {
             return ConnectWebSocket("/", (_socket, msg) =>
             {
@@ -393,7 +393,7 @@ protected override IWebSocket OnGetTickersWebSocket(Action<IReadOnlyCollection<K
             });
         }
 
-        protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
+        protected override Task<IWebSocket> OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
         {
 			if (marketSymbols == null || marketSymbols.Length == 0)
 			{

ExchangeSharp/API/Exchanges/Digifinex/ExchangeDigifinexAPI.cs

@@ -387,7 +387,7 @@ protected override async Task OnCancelOrderAsync(string orderId, string marketSy
 
         #region WebSocket APIs
 
-        protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
+        protected override Task<IWebSocket> OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
         {
             if (callback == null)
                 return null;
@@ -454,7 +454,7 @@ await callback.Invoke(new KeyValuePair<string, ExchangeTrade>(
             });
         }
 
-        protected override IWebSocket OnGetDeltaOrderBookWebSocket(Action<ExchangeOrderBook> callback, int maxCount = 20, params string[] marketSymbols)
+        protected override Task<IWebSocket> OnGetDeltaOrderBookWebSocket(Action<ExchangeOrderBook> callback, int maxCount = 20, params string[] marketSymbols)
         {
             if (callback == null)
             {

ExchangeSharp/API/Exchanges/Gemini/ExchangeGeminiAPI.cs

@@ -284,7 +284,7 @@ protected override async Task OnCancelOrderAsync(string orderId, string marketSy
             await MakeJsonRequestAsync<JToken>("/order/cancel", null, new Dictionary<string, object>{ { "nonce", nonce }, { "order_id", orderId } });
         }
 
-		protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
+		protected override Task<IWebSocket> OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
 		{
 			//{
 			//  "type": "l2_updates",

ExchangeSharp/API/Exchanges/Hitbtc/ExchangeHitbtcAPI.cs

@@ -440,7 +440,7 @@ protected override async Task<ExchangeWithdrawalResponse> OnWithdrawAsync(Exchan
 
 		// working on it. Hitbtc has extensive support for sockets, including trading
 
-		protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
+		protected override Task<IWebSocket> OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
 		{
 			if (marketSymbols == null || marketSymbols.Length == 0)
 			{

ExchangeSharp/API/Exchanges/Huobi/ExchangeHuobiAPI.cs

@@ -220,7 +220,7 @@ protected async override Task<IEnumerable<KeyValuePair<string, ExchangeTicker>>>
             return tickers;
         }
 
-        protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
+        protected override Task<IWebSocket> OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
         {
             return ConnectWebSocket(string.Empty, async (_socket, msg) =>
             {
@@ -287,7 +287,7 @@ protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, Exc
             });
         }
 
-        protected override IWebSocket OnGetDeltaOrderBookWebSocket(Action<ExchangeOrderBook> callback, int maxCount = 20, params string[] marketSymbols)
+        protected override Task<IWebSocket> OnGetDeltaOrderBookWebSocket(Action<ExchangeOrderBook> callback, int maxCount = 20, params string[] marketSymbols)
         {
             return ConnectWebSocket(string.Empty, async (_socket, msg) =>
             {

ExchangeSharp/API/Exchanges/Kraken/ExchangeKrakenAPI.cs

@@ -828,7 +828,7 @@ protected override async Task OnCancelOrderAsync(string orderId, string marketSy
             await MakeJsonRequestAsync<JToken>("/0/private/CancelOrder", null, payload);
         }
 
-		protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
+		protected override Task<IWebSocket> OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
 		{
 			if (marketSymbols == null || marketSymbols.Length == 0)
 			{

ExchangeSharp/API/Exchanges/KuCoin/ExchangeKuCoinAPI.cs

@@ -433,7 +433,7 @@ protected override async Task<ExchangeWithdrawalResponse> OnWithdrawAsync(Exchan
 
         #region Websockets
 
-        protected override IWebSocket OnGetTickersWebSocket(Action<IReadOnlyCollection<KeyValuePair<string, ExchangeTicker>>> callback, params string[] marketSymbols)
+        protected override Task<IWebSocket> OnGetTickersWebSocket(Action<IReadOnlyCollection<KeyValuePair<string, ExchangeTicker>>> callback, params string[] marketSymbols)
         {
             var websocketUrlToken = GetWebsocketBulletToken();
             return ConnectWebSocket(
@@ -463,7 +463,7 @@ protected override IWebSocket OnGetTickersWebSocket(Action<IReadOnlyCollection<K
                 );
         }
 
-        protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
+        protected override Task<IWebSocket> OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
         {
 			//{
 			//  "id":"5c24c5da03aa673885cd67aa",

ExchangeSharp/API/Exchanges/OKGroup/OKGroupCommon.cs

@@ -192,7 +192,7 @@ void parseData(Tuple<JToken, string> data)
             return tickers;
         }
 
-        protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
+        protected override Task<IWebSocket> OnGetTradesWebSocket(Func<KeyValuePair<string, ExchangeTrade>, Task> callback, params string[] marketSymbols)
         {
 			/*
 			 spot request:
@@ -228,7 +228,7 @@ protected override IWebSocket OnGetTradesWebSocket(Func<KeyValuePair<string, Exc
 				});
         }
 
-        protected override IWebSocket OnGetDeltaOrderBookWebSocket(Action<ExchangeOrderBook> callback, int maxCount = 20, params string[] marketSymbols)
+        protected override Task<IWebSocket> OnGetDeltaOrderBookWebSocket(Action<ExchangeOrderBook> callback, int maxCount = 20, params string[] marketSymbols)
         {
 			/*
 			 request:
@@ -633,7 +633,7 @@ private ExchangeOrderResult ParseOrder(JToken token)
             return result;
         }
 
-        private IWebSocket ConnectWebSocketOkex(Func<IWebSocket, Task> connected, Func<IWebSocket, string, string[], JToken, Task> callback, int symbolArrayIndex = 3)
+        private Task<IWebSocket> ConnectWebSocketOkex(Func<IWebSocket, Task> connected, Func<IWebSocket, string, string[], JToken, Task> callback, int symbolArrayIndex = 3)
         {
 			Timer pingTimer = null;
             return ConnectWebSocket(url: string.Empty, messageCallback: async (_socket, msg) =>
@@ -683,7 +683,7 @@ private IWebSocket ConnectWebSocketOkex(Func<IWebSocket, Task> connected, Func<I
 			});
         }
 
-        private IWebSocket ConnectPrivateWebSocketOkex(Func<IWebSocket, Task> connected, Func<IWebSocket, string, string[], JToken, Task> callback, int symbolArrayIndex = 3)
+        private Task<IWebSocket> ConnectPrivateWebSocketOkex(Func<IWebSocket, Task> connected, Func<IWebSocket, string, string[], JToken, Task> callback, int symbolArrayIndex = 3)
         {
             return ConnectWebSocketOkex(async (_socket) =>
             {