Handle disconnection / reconnection

src/StatsdClient/StatsSender.cs

@@ -11,6 +11,7 @@ internal class StatsSender : IStatsSender, IDisposable
         private static readonly TimeSpan NoBufferSpaceAvailableWait = TimeSpan.FromMilliseconds(10);
         private readonly Socket _socket;
         private readonly int _noBufferSpaceAvailableRetryCount;
+        private readonly EndPoint _endPoint;
 
         private StatsSender(
             StatsSenderTransportType transport,
@@ -20,6 +21,7 @@ private StatsSender(
             TimeSpan? bufferFullBlockDuration)
         {
             TransportType = transport;
+            _endPoint = endPoint;
             if (bufferFullBlockDuration.HasValue)
             {
                 _noBufferSpaceAvailableRetryCount = (int)(bufferFullBlockDuration.Value.TotalMilliseconds
@@ -52,8 +54,6 @@ private StatsSender(
             {
                 // It is not supported on Windows for Dgram with UDP.
             }
-
-            _socket.Connect(endPoint);
         }
 
         public StatsSenderTransportType TransportType { get; }
@@ -90,7 +90,7 @@ public bool Send(byte[] buffer, int length)
             {
                 try
                 {
-                    _socket.Send(buffer, 0, length, SocketFlags.None);
+                    _socket.SendTo(buffer, 0, length, SocketFlags.None, _endPoint);
                     return true;
                 }
                 catch (SocketException e) when (e.SocketErrorCode == SocketError.NoBufferSpaceAvailable)

src/StatsdClient/Worker/AsynchronousWorker.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Threading.Tasks;
 
 namespace StatsdClient.Worker
@@ -63,9 +64,16 @@ public void Dispose()
             }
 
             _terminate = true;
-            foreach (var worker in _workers)
+            try
             {
-                worker.Wait();
+                foreach (var worker in _workers)
+                {
+                    worker.Wait();
+                }
+            }
+            catch (Exception e)
+            {
+                Debug.WriteLine(e.Message);
             }
 
             _workers.Clear();
@@ -77,29 +85,36 @@ private void Dequeue()
 
             while (true)
             {
-                if (_queue.TryDequeue(out var v))
-                {
-                    _handler.OnNewValue(v);
-                    waitDuration = MinWaitDuration;
-                }
-                else
+                try
                 {
-                    if (_terminate)
+                    if (_queue.TryDequeue(out var v))
                     {
-                        _handler.OnShutdown();
-                        return;
+                        _handler.OnNewValue(v);
+                        waitDuration = MinWaitDuration;
                     }
-
-                    if (_handler.OnIdle())
+                    else
                     {
-                        _waiter.Wait(waitDuration);
-                        waitDuration = waitDuration + waitDuration;
-                        if (waitDuration > MaxWaitDuration)
+                        if (_terminate)
+                        {
+                            _handler.OnShutdown();
+                            return;
+                        }
+
+                        if (_handler.OnIdle())
                         {
-                            waitDuration = MaxWaitDuration;
+                            _waiter.Wait(waitDuration);
+                            waitDuration = waitDuration + waitDuration;
+                            if (waitDuration > MaxWaitDuration)
+                            {
+                                waitDuration = MaxWaitDuration;
+                            }
                         }
                     }
                 }
+                catch (Exception e)
+                {
+                    Debug.WriteLine(e.Message);
+                }
             }
         }
     }

tests/StatsdClient.Tests/DogStatsdServiceReconnectionTests.cs

@@ -0,0 +1,68 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using NUnit.Framework;
+using StatsdClient;
+using Tests.Utils;
+
+namespace Tests
+{
+    [TestFixture]
+    public class DogStatsdServiceReconnectionTests
+    {
+        [Test]
+        public void UDPReconnection()
+        {
+            CheckReconnection(new StatsdConfig
+            {
+                StatsdServerName = "127.0.0.1",
+                StatsdPort = 1234,
+            });
+        }
+
+#if !OS_WINDOWS
+        [Test]
+        public void UDSReconnection()
+        {
+            using (var temporaryPath = new TemporaryPath())
+            {
+                CheckReconnection(new StatsdConfig
+                {
+                    StatsdServerName = StatsdBuilder.UnixDomainSocketPrefix + temporaryPath.Path,
+                });
+            }
+        }
+#endif
+
+        private static void CheckReconnection(StatsdConfig config)
+        {
+            SocketServer server = null;
+
+            try
+            {
+                server = new SocketServer(config);
+                using (var service = new DogStatsdService())
+                {
+                    service.Configure(config);
+                    service.Increment("test1");
+                    Assert.AreEqual("test1:1|c", server.Stop().Single());
+                    server.Dispose();
+
+                    // Send a metric when the server is not running.
+                    service.Increment("test2");
+                    Task.Delay(TimeSpan.FromMilliseconds(500)).Wait();
+
+                    // Restart the server
+                    server = new SocketServer(config, removeUDSFileBeforeStarting: true);
+                    service.Increment("test3");
+                    service.Dispose();
+                    Assert.AreEqual("test3:1|c", server.Stop().Last());
+                }
+            }
+            finally
+            {
+                server?.Dispose();
+            }
+        }
+    }
+}

tests/StatsdClient.Tests/utils/SocketServer.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.IO;
 using System.Net;
 using System.Net.Sockets;
 using System.Threading;
@@ -18,7 +19,7 @@ internal class SocketServer : IDisposable
 
         private volatile bool _shutdown = false;
 
-        public SocketServer(StatsdConfig config)
+        public SocketServer(StatsdConfig config, bool removeUDSFileBeforeStarting = false)
         {
             EndPoint endPoint;
             int bufferSize;
@@ -27,6 +28,11 @@ public SocketServer(StatsdConfig config)
             if (serverName.StartsWith(StatsdBuilder.UnixDomainSocketPrefix))
             {
                 serverName = serverName.Substring(StatsdBuilder.UnixDomainSocketPrefix.Length);
+                if (removeUDSFileBeforeStarting)
+                {
+                    File.Delete(serverName);
+                }
+
                 _server = new Socket(AddressFamily.Unix, SocketType.Dgram, ProtocolType.Unspecified);
                 endPoint = new UnixEndPoint(serverName);
                 bufferSize = config.StatsdMaxUnixDomainSocketPacketSize;