Merge pull request #159 from frco9/bugfix/158-no-disconnect-on-poor-connection

fix: Catch websocket exception to detect disconnection & lock Emit to prevent race condition on OutgoingBytes

Author: doghappy

src/SocketIOClient/Processors/PingProcessor.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Diagnostics;
 
 namespace SocketIOClient.Processors
 {
@@ -8,10 +9,22 @@ public override async void Process(MessageContext ctx)
         {
             if (ctx.SocketIO.Options.EIO == 4)
             {
-                ctx.SocketIO.InvokePing();
-                DateTime pingTime = DateTime.Now;
-                await ctx.SocketIO.Socket.SendMessageAsync("3");
-                ctx.SocketIO.InvokePong(DateTime.Now - pingTime);
+                try
+                {
+                    ctx.SocketIO.InvokePing();
+                    DateTime pingTime = DateTime.Now;
+                    await ctx.SocketIO.Socket.SendMessageAsync("3");
+                    ctx.SocketIO.InvokePong(DateTime.Now - pingTime);
+                }
+                catch (System.Net.WebSockets.WebSocketException e)
+                {
+                    ctx.SocketIO.InvokeDisconnect(e.Message);
+                }
+                catch (Exception ex)
+                {
+                    Trace.TraceError(ex.ToString());
+                }
+                
             }
         }
     }

src/SocketIOClient/SocketIO.cs

@@ -1,6 +1,7 @@
 using SocketIOClient.ConnectInterval;
 using SocketIOClient.EioHandler;
 using SocketIOClient.EventArguments;
+using SocketIOClient.Exceptions;
 using SocketIOClient.JsonSerializer;
 using SocketIOClient.Response;
 using SocketIOClient.WebSocketClient;
@@ -112,6 +113,8 @@ public Uri ServerUri
 
         public IJsonSerializer JsonSerializer { get; set; }
 
+        static readonly SemaphoreSlim _semaphoreSlim = new SemaphoreSlim(1, 1);
+
         #region Socket.IO event
         public event EventHandler OnConnected;
         //public event EventHandler<string> OnConnectError;
@@ -299,15 +302,27 @@ private async Task EmityCoreAsync(string eventName, int packetId, string data, C
             }
             builder.Append(']');
             string message = builder.ToString();
-            await Socket.SendMessageAsync(message, cancellationToken);
-            if (OutGoingBytes.Count > 0)
+            try
             {
-                foreach (var item in OutGoingBytes)
+                await Socket.SendMessageAsync(message, cancellationToken);
+                if (OutGoingBytes.Count > 0)
                 {
-                    await Socket.SendMessageAsync(item, cancellationToken);
+                    foreach (var item in OutGoingBytes)
+                    {
+                        await Socket.SendMessageAsync(item, cancellationToken);
+                    }
+                    OutGoingBytes.Clear();
                 }
-                OutGoingBytes.Clear();
             }
+            catch (System.Net.WebSockets.WebSocketException e)
+            {
+                this.InvokeDisconnect(e.Message);
+            }
+            catch (InvalidSocketStateException e)
+            {
+                this.InvokeDisconnect(e.Message);
+            }
+            
         }
 
         internal async Task EmitCallbackAsync(int packetId, params object[] data)
@@ -364,8 +379,16 @@ public async Task EmitAsync(string eventName, params object[] data)
 
         public async Task EmitAsync(string eventName, CancellationToken cancellationToken, params object[] data)
         {
-            string dataString = GetDataString(data);
-            await EmityCoreAsync(eventName, -1, dataString, cancellationToken);
+            await _semaphoreSlim.WaitAsync();
+            try
+            {
+                string dataString = GetDataString(data);
+                await EmityCoreAsync(eventName, -1, dataString, cancellationToken);
+            }
+            finally
+            {
+                _semaphoreSlim.Release();
+            }
         }
 
         /// <summary>
@@ -382,9 +405,17 @@ public async Task EmitAsync(string eventName, Action<SocketIOResponse> ack, para
 
         public async Task EmitAsync(string eventName, CancellationToken cancellationToken, Action<SocketIOResponse> ack, params object[] data)
         {
-            Acks.Add(++PacketId, ack);
-            string dataString = GetDataString(data);
-            await EmityCoreAsync(eventName, PacketId, dataString, cancellationToken);
+            await _semaphoreSlim.WaitAsync();
+            try
+            {
+                Acks.Add(++PacketId, ack);
+                string dataString = GetDataString(data);
+                await EmityCoreAsync(eventName, PacketId, dataString, cancellationToken);
+            }
+            finally
+            {
+                _semaphoreSlim.Release();
+            }
         }
 
         internal void Open(OpenResponse openResponse)