Updated socketclient

CryptoExchange.Net/Sockets/BaseSocket.cs

@@ -32,9 +32,6 @@ namespace CryptoExchange.Net.Sockets
         protected HttpConnectProxy proxy;
 
         public int Id { get; }
-        public DateTime? DisconnectTime { get; set; }
-
-        public bool ShouldReconnect { get; set; }
         public bool Reconnecting { get; set; }
         public string Origin { get; set; }
 
@@ -42,7 +39,8 @@ namespace CryptoExchange.Net.Sockets
         public bool IsClosed => socket.State == WebSocketState.Closed;
         public bool IsOpen => socket.State == WebSocketState.Open;
         public SslProtocols SSLProtocols { get; set; } = SslProtocols.Tls12 | SslProtocols.Tls11 | SslProtocols.Tls;
-        public Func<byte[], string> DataInterpreter { get; set; }
+        public Func<byte[], string> DataInterpreterBytes { get; set; }
+        public Func<string, string> DataInterpreterString { get; set; }
 
         public DateTime LastActionTime { get; private set; }
         public TimeSpan Timeout { get; set; }
@@ -77,7 +75,7 @@ namespace CryptoExchange.Net.Sockets
 
         private void HandleByteData(byte[] data)
         {
-            var message = DataInterpreter(data);
+            var message = DataInterpreterBytes(data);
             Handle(messageHandlers, message);
         }
 
@@ -203,7 +201,13 @@ namespace CryptoExchange.Net.Sockets
                 socket.Opened += (o, s) => Handle(openHandlers);
                 socket.Closed += (o, s) => Handle(closeHandlers);
                 socket.Error += (o, s) => Handle(errorHandlers, s.Exception);
-                socket.MessageReceived += (o, s) => Handle(messageHandlers, s.Message);
+                socket.MessageReceived += (o, s) =>
+                {
+                    string data = s.Message;
+                    if (DataInterpreterString != null)
+                        data = DataInterpreterString(data);
+                    Handle(messageHandlers, data);
+                };
                 socket.DataReceived += (o, s) => HandleByteData(s.Data);
             }
 

CryptoExchange.Net/Sockets/SocketConnection.cs

@@ -0,0 +1,349 @@
+using CryptoExchange.Net.Interfaces;
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using CryptoExchange.Net.Logging;
+using Newtonsoft.Json;
+using Newtonsoft.Json.Linq;
+
+namespace CryptoExchange.Net.Sockets
+{
+    public class SocketConnection
+    {
+        public event Action ConnectionLost;
+        public event Action<TimeSpan> ConnectionRestored;
+        public event Action Closed;
+
+        public int HandlerCount
+        {
+            get { lock (handlersLock)
+                return handlers.Count(h => h.UserSubscription); }
+        }
+
+        public bool Authenticated { get; set; }
+        public bool Connected { get; private set; }
+
+
+        public IWebsocket Socket { get; set; }
+        public bool ShouldReconnect { get; set; }
+        public DateTime? DisconnectTime { get; set; }
+        public bool PausedActivity { get; set; }
+
+        private readonly List<SocketSubscription> handlers;
+        private readonly object handlersLock = new object();
+
+        private bool lostTriggered;
+        private readonly Log log;
+        private readonly SocketClient socketClient;
+
+        private readonly List<PendingRequest> pendingRequests;
+
+        public SocketConnection(SocketClient client, Log log, IWebsocket socket)
+        {
+            this.log = log;
+            socketClient = client;
+
+            pendingRequests = new List<PendingRequest>();
+
+            handlers = new List<SocketSubscription>();
+            Socket = socket;
+
+            Socket.Timeout = client.SocketTimeout;
+            Socket.OnMessage += ProcessMessage;
+            Socket.OnClose += () =>
+            {
+                if (lostTriggered)
+                    return;
+
+                DisconnectTime = DateTime.UtcNow;
+                lostTriggered = true;
+                
+                if (ShouldReconnect)
+                    ConnectionLost?.Invoke();
+            };
+            Socket.OnClose += SocketOnClose;
+            Socket.OnOpen += () =>
+            {
+                PausedActivity = false;
+                Connected = true;
+                if (lostTriggered)
+                {
+                    lostTriggered = false;
+                    ConnectionRestored?.Invoke(DisconnectTime.HasValue ? DateTime.UtcNow - DisconnectTime.Value: TimeSpan.FromSeconds(0));
+                }
+            };
+        }
+
+        public SocketSubscription AddHandler(object request, bool userSubscription, Action<SocketConnection, JToken> dataHandler)
+        {
+            var handler = new SocketSubscription(null, request, userSubscription, dataHandler);
+            lock (handlersLock)
+                handlers.Add(handler);
+            return handler;
+        }
+
+        public SocketSubscription AddHandler(string identifier, bool userSubscription, Action<SocketConnection, JToken> dataHandler)
+        {
+            var handler = new SocketSubscription(identifier, null, userSubscription, dataHandler);
+            lock (handlersLock)
+                handlers.Add(handler);
+            return handler;
+        }
+
+        public void ProcessMessage(string data)
+        {
+            log.Write(LogVerbosity.Debug, $"Socket {Socket.Id} received data: " + data);
+            
+            var tokenData = JToken.Parse(data);
+            
+            foreach (var pendingRequest in pendingRequests.ToList())
+            {
+                if (pendingRequest.Check(tokenData))
+                {
+                    pendingRequests.Remove(pendingRequest);
+                    return;
+                }
+            }
+
+            if (!HandleData(tokenData))
+            {
+                log.Write(LogVerbosity.Debug, "Message not handled: " + tokenData);
+            }
+        }
+
+        private bool HandleData(JToken tokenData)
+        {
+            SocketSubscription currentSubscription = null;
+            try
+            { 
+                bool handled = false;
+                var sw = Stopwatch.StartNew();
+                lock (handlersLock)
+                {
+                    foreach (var handler in handlers)
+                    {
+                        currentSubscription = handler;
+                        if (handler.Request == null)
+                        {
+                            if (socketClient.MessageMatchesHandler(tokenData, handler.Identifier))
+                            {
+                                handled = true;
+                                handler.MessageHandler(this, tokenData);
+                            }
+                        }
+                        else
+                        {
+                            if (socketClient.MessageMatchesHandler(tokenData, handler.Request))
+                            {
+                                handled = true;
+                                tokenData = socketClient.ProcessTokenData(tokenData);
+                                handler.MessageHandler(this, tokenData);
+                            }
+                        }
+                    }
+                }
+
+                sw.Stop();
+                if (sw.ElapsedMilliseconds > 500)
+                    log.Write(LogVerbosity.Warning, $"Socket {Socket.Id} message processing slow ({sw.ElapsedMilliseconds}ms), consider offloading data handling to another thread. " +
+                                                    "Data from this socket may arrive late or not at all if message processing is continuously slow.");
+                return handled;
+            }
+            catch (Exception ex)
+            {
+                log.Write(LogVerbosity.Error, $"Socket {Socket.Id} Exception during message processing\r\nException: {ex}\r\nData: {tokenData}");
+                currentSubscription?.InvokeExceptionHandler(ex);
+                return false;
+            }
+        }
+
+        public virtual async Task SendAndWait<T>(T obj, TimeSpan timeout, Func<JToken, bool> handler)
+        {
+            var pending = new PendingRequest(handler, timeout);
+            pendingRequests.Add(pending);
+            await Task.Run(() =>
+            {
+                Send(obj);
+                pending.Event.WaitOne(timeout);
+            }).ConfigureAwait(false);
+        }
+
+        /// <summary>
+        /// Send data to the websocket
+        /// </summary>
+        /// <typeparam name="T">The type of the object to send</typeparam>
+        /// <param name="obj">The object to send</param>
+        /// <param name="nullValueHandling">How null values should be serialized</param>
+        public virtual void Send<T>(T obj, NullValueHandling nullValueHandling = NullValueHandling.Ignore)
+        {
+            Send(JsonConvert.SerializeObject(obj, Formatting.None, new JsonSerializerSettings { NullValueHandling = nullValueHandling }));
+        }
+
+        /// <summary>
+        /// Send string data to the websocket
+        /// </summary>
+        /// <param name="data">The data to send</param>
+        public virtual void Send(string data)
+        {
+            log.Write(LogVerbosity.Debug, $"Socket {Socket.Id} sending data: {data}");
+            Socket.Send(data);
+        }
+
+        /// <summary>
+        /// Handler for a socket closing. Reconnects the socket if needed, or removes it from the active socket list if not
+        /// </summary>
+        protected virtual void SocketOnClose()
+        {
+            if (socketClient.AutoReconnect && ShouldReconnect)
+            {
+                if (Socket.Reconnecting)
+                    return; // Already reconnecting
+
+                Socket.Reconnecting = true;
+
+                log.Write(LogVerbosity.Info, $"Socket {Socket.Id} Connection lost, will try to reconnect after {socketClient.ReconnectInterval}");
+                Task.Run(async () =>
+                {
+                    while (ShouldReconnect)
+                    {
+                        Thread.Sleep(socketClient.ReconnectInterval);
+                        if (!ShouldReconnect)
+                        {
+                            // Should reconnect changed to false while waiting to reconnect
+                            Socket.Reconnecting = false;
+                            return;
+                        }
+
+                        Socket.Reset();
+                        if (!await Socket.Connect().ConfigureAwait(false))
+                        {
+                            log.Write(LogVerbosity.Debug, $"Socket {Socket.Id} failed to reconnect");
+                            continue;
+                        }
+
+                        var time = DisconnectTime;
+                        DisconnectTime = null;
+
+                        log.Write(LogVerbosity.Info, $"Socket {Socket.Id} reconnected after {DateTime.UtcNow - time}");
+
+                        var reconnectResult = await ProcessReconnect().ConfigureAwait(false);
+                        if (!reconnectResult)
+                            await Socket.Close().ConfigureAwait(false);
+                        else
+                            break;
+                    }
+
+                    Socket.Reconnecting = false;
+                });
+            }
+            else
+            {
+                log.Write(LogVerbosity.Info, $"Socket {Socket.Id} closed");
+                Socket.Dispose();
+                Closed?.Invoke();
+            }
+        }
+
+        public async Task<bool> ProcessReconnect()
+        {
+            if (Authenticated)
+            {
+                var authResult = await socketClient.AuthenticateSocket(this).ConfigureAwait(false);
+                if (!authResult.Success)
+                {
+                    log.Write(LogVerbosity.Info, "Authentication failed on reconnected socket. Disconnecting and reconnecting.");
+                    return false;
+                }
+
+                log.Write(LogVerbosity.Debug, "Authentication succeeded on reconnected socket.");
+            }
+
+            List<SocketSubscription> handlerList;
+            lock (handlersLock)
+                handlerList = handlers.Where(h => h.Request != null).ToList();
+            foreach (var handler in handlerList)
+            {
+                var resubResult = await socketClient.SubscribeAndWait(this, handler.Request, handler).ConfigureAwait(false);
+                if (!resubResult.Success)
+                {
+                    log.Write(LogVerbosity.Debug, "Resubscribing all subscriptions failed on reconnected socket. Disconnecting and reconnecting.");
+                    return false;
+                }
+            }
+
+            log.Write(LogVerbosity.Debug, "All subscription successfully resubscribed on reconnected socket.");
+            return true;
+        }
+
+        public async Task Close()
+        {
+            Connected = false;
+            ShouldReconnect = false;
+            lock (socketClient.socketLock)
+            {
+                if (socketClient.sockets.Contains(this))
+                    socketClient.sockets.Remove(this);
+            }
+
+            await Socket.Close().ConfigureAwait(false);
+            Socket.Dispose();
+        }
+
+        public async Task Close(SocketSubscription subscription)
+        {
+            if (subscription.Confirmed)
+                await socketClient.Unsubscribe(this, subscription).ConfigureAwait(false);
+
+            bool shouldCloseWrapper = false;
+            lock (handlersLock)
+            {
+                handlers.Remove(subscription);
+                if (handlers.Count(r => r.UserSubscription) == 0)
+                    shouldCloseWrapper = true;
+            }
+
+            if (shouldCloseWrapper)
+                await Close().ConfigureAwait(false);
+        }
+    }
+
+    public class PendingRequest
+    {
+        public Func<JToken, bool> Handler { get; }
+        public JToken Result { get; private set; }
+        public ManualResetEvent Event { get; }
+        public TimeSpan Timeout { get; }
+
+        private readonly DateTime startTime;
+
+        public PendingRequest(Func<JToken, bool> handler, TimeSpan timeout)
+        {
+            Handler = handler;
+            Event = new ManualResetEvent(false);
+            Timeout = timeout;
+            startTime = DateTime.UtcNow;
+        }
+
+        public bool Check(JToken data)
+        {
+            if (Handler(data))
+            {
+                Result = data;
+                Event.Set();
+                return true;
+            }
+
+            if (DateTime.UtcNow - startTime > Timeout)
+            {
+                // Timed out
+                Event.Set();
+                return true;
+            }
+
+            return false;
+        }
+    }
+}

CryptoExchange.Net/Sockets/SocketEvent.cs

@@ -1,40 +0,0 @@
-using CryptoExchange.Net.Objects;
-using System.Threading;
-
-namespace CryptoExchange.Net.Sockets
-{
-    public class SocketEvent
-    {
-        public string Name { get; set; }
-        public string WaitingId { get; set; }
-
-        private CallResult<bool> result;
-        private readonly ManualResetEvent setEvnt;
-
-        public SocketEvent(string name)
-        {
-            Name = name;
-            setEvnt = new ManualResetEvent(false);
-            result = new CallResult<bool>(false, new UnknownError("No response received"));
-        }
-
-        internal void Set(bool result, Error error)
-        {
-            this.result = new CallResult<bool>(result, error);
-            setEvnt.Set();
-            WaitingId = null;
-        }
-        
-        public CallResult<bool> Wait(int timeout = 5000)
-        {
-            setEvnt.WaitOne(timeout);
-            return result;
-        }
-
-        public void Reset()
-        {
-            setEvnt.Reset();
-            result = new CallResult<bool>(false, new UnknownError("No response received"));
-        }
-    }
-}

CryptoExchange.Net/Sockets/SocketRequest.cs

@@ -1,10 +0,0 @@
-using Newtonsoft.Json;
-
-namespace CryptoExchange.Net.Sockets
-{
-    public class SocketRequest
-    {
-        [JsonIgnore]
-        public bool Signed { get; set; }
-    }
-}

CryptoExchange.Net/Sockets/SocketSubscription.cs

@@ -1,161 +1,35 @@
-using CryptoExchange.Net.Interfaces;
-using CryptoExchange.Net.Objects;
+using System;
 using Newtonsoft.Json.Linq;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Threading.Tasks;
 
 namespace CryptoExchange.Net.Sockets
 {
     public class SocketSubscription
     {
-        public event Action ConnectionLost;
-        public event Action<TimeSpan> ConnectionRestored;
         public event Action<Exception> Exception;
 
         /// <summary>
         /// Message handlers for this subscription. Should return true if the message is handled and should not be distributed to the other handlers
         /// </summary>
-        public Dictionary<string, Func<SocketSubscription, JToken, bool>> MessageHandlers { get; set; }
-        public List<SocketEvent> Events { get; set; }
+        public Action<SocketConnection, JToken> MessageHandler { get; set; }
 
-        public IWebsocket Socket { get; set; }
-        public SocketRequest Request { get; set; }
-
-        public SocketType Type { get; set; }
-
-        private bool lostTriggered;
-        private readonly List<SocketEvent> waitingForEvents;
-        private object eventLock = new object();
-
-
-        public SocketSubscription(IWebsocket socket)
-        {
-            Socket = socket;
-            Events = new List<SocketEvent>();
-            waitingForEvents = new List<SocketEvent>();
-
-            MessageHandlers = new Dictionary<string, Func<SocketSubscription, JToken, bool>>();
-
-            Socket.OnClose += () =>
-            {
-                if (lostTriggered)
-                    return;
-
-                Socket.DisconnectTime = DateTime.UtcNow;
-                lostTriggered = true;
-
-                lock (eventLock)
-                {
-                    foreach (var events in Events)
-                        events.Reset();
-                }
-
-                if (Socket.ShouldReconnect)
-                    ConnectionLost?.Invoke();
-            };
-            Socket.OnOpen += () =>
-            {
-                if (lostTriggered)
-                {
-                    lostTriggered = false;
-                    ConnectionRestored?.Invoke(Socket.DisconnectTime.HasValue ? DateTime.UtcNow - Socket.DisconnectTime.Value: TimeSpan.FromSeconds(0));
-                }
-            };
-        }
-
-        public void AddEvent(string name)
-        {
-            lock (eventLock)
-                Events.Add(new SocketEvent(name));
-        }
-
-        public void SetEventByName(string name, bool success, Error error)
-        {
-            lock (eventLock)
-            {
-                var waitingEvent = waitingForEvents.SingleOrDefault(e => e.Name == name);
-                if (waitingEvent != null)
-                {
-                    waitingEvent.Set(success, error);
-                    waitingForEvents.Remove(waitingEvent);
-                }
-            }
-        }
-
-        public void SetEventById(string id, bool success, Error error)
-        {
-            lock (eventLock)
-            {
-                var waitingEvent = waitingForEvents.SingleOrDefault(e => e.WaitingId == id);
-                if (waitingEvent != null)
-                {
-                    waitingEvent.Set(success, error);
-                    waitingForEvents.Remove(waitingEvent);
-                }
-            }
-        }
-
-        public SocketEvent GetWaitingEvent(string name)
-        {
-            lock (eventLock)
-                return waitingForEvents.SingleOrDefault(w => w.Name == name);
-        }
+        public object Request { get; set; }
+        public string Identifier { get; set; }
+        public bool UserSubscription { get; set; }
         
-        public Task<CallResult<bool>> WaitForEvent(string name, TimeSpan timeout)
-        {
-            lock (eventLock)
-                return WaitForEvent(name, (int)Math.Round(timeout.TotalMilliseconds, 0));
-        }
+        public bool Confirmed { get; set; }
 
-        public Task<CallResult<bool>> WaitForEvent(string name, int timeout)
-        {
-            lock (eventLock)
-            {
-                var evnt = Events.Single(e => e.Name == name);
-                waitingForEvents.Add(evnt);
-                return Task.Run(() => evnt.Wait(timeout));
-            }
-        }
 
-        public Task<CallResult<bool>> WaitForEvent(string name, string id, TimeSpan timeout)
+        public SocketSubscription(string identifier, object request, bool userSubscription, Action<SocketConnection, JToken> dataHandler)
         {
-            lock (eventLock)
-                return WaitForEvent(name, id, (int)Math.Round(timeout.TotalMilliseconds, 0));
+            UserSubscription = userSubscription;
+            MessageHandler = dataHandler;
+            Identifier = identifier;
+            Request = request;
         }
-
-        public Task<CallResult<bool>> WaitForEvent(string name, string id, int timeout)
-        {
-            lock (eventLock)
-            {
-                var evnt = Events.Single(e => e.Name == name);
-                evnt.WaitingId = id;
-                waitingForEvents.Add(evnt);
-                return Task.Run(() => evnt.Wait(timeout));
-            }
-        }
-
-        public void ResetEvents()
-        {
-            lock (eventLock)
-            {
-                foreach (var waiting in waitingForEvents)
-                    waiting.Set(false, new UnknownError("Connection reset"));
-                waitingForEvents.Clear();
-            }
-        }
-
+       
         public void InvokeExceptionHandler(Exception e)
         {
             Exception?.Invoke(e);
         }
-
-        public async Task Close()
-        {
-            Socket.ShouldReconnect = false;
-            await Socket.Close().ConfigureAwait(false);
-            Socket.Dispose();
-        }
     }
 }

CryptoExchange.Net/Sockets/UpdateSubscription.cs

@@ -5,6 +5,7 @@ namespace CryptoExchange.Net.Sockets
 {
     public class UpdateSubscription
     {
+        private readonly SocketConnection connection;
         private readonly SocketSubscription subscription;
 
         /// <summary>
@@ -12,8 +13,8 @@ namespace CryptoExchange.Net.Sockets
         /// </summary>
         public event Action ConnectionLost
         {
-            add => subscription.ConnectionLost += value;
-            remove => subscription.ConnectionLost -= value;
+            add => connection.ConnectionLost += value;
+            remove => connection.ConnectionLost -= value;
         }
 
         /// <summary>
@@ -21,8 +22,8 @@ namespace CryptoExchange.Net.Sockets
         /// </summary>
         public event Action<TimeSpan> ConnectionRestored
         {
-            add => subscription.ConnectionRestored += value;
-            remove => subscription.ConnectionRestored -= value;
+            add => connection.ConnectionRestored += value;
+            remove => connection.ConnectionRestored -= value;
         }
 
         /// <summary>
@@ -37,11 +38,12 @@ namespace CryptoExchange.Net.Sockets
         /// <summary>
         /// The id of the socket
         /// </summary>
-        public int Id => subscription.Socket.Id;
+        public int Id => connection.Socket.Id;
 
-        public UpdateSubscription(SocketSubscription sub)
+        public UpdateSubscription(SocketConnection connection, SocketSubscription subscription)
         {
-            subscription = sub;
+            this.connection = connection;
+            this.subscription = subscription;
         }
 
         /// <summary>
@@ -50,7 +52,7 @@ namespace CryptoExchange.Net.Sockets
         /// <returns></returns>
         public async Task Close()
         {
-            await subscription.Close().ConfigureAwait(false);
+            await connection.Close(subscription).ConfigureAwait(false);
         }
     }
 }