ping tarantool instead of socket ping, tbd

Author: Dmitry-Bryliuk

src/tarantool.client/Converters/PingPacketConverter.cs

@@ -0,0 +1,24 @@
+using System;
+
+using ProGaudi.MsgPack.Light;
+
+using ProGaudi.Tarantool.Client.Model.Requests;
+
+namespace ProGaudi.Tarantool.Client.Converters
+{
+    internal class PingPacketConverter : IMsgPackConverter<PingRequest>
+    {
+        public void Initialize(MsgPackContext context)
+        {
+        }
+
+        public void Write(PingRequest value, IMsgPackWriter writer)
+        {
+        }
+
+        public PingRequest Read(IMsgPackReader reader)
+        {
+            throw new NotImplementedException();
+        }
+    }
+}

src/tarantool.client/LogicalConnectionManager.cs

@@ -1,14 +1,14 @@
-using System.Threading.Tasks;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+using ProGaudi.Tarantool.Client.Model;
 using ProGaudi.Tarantool.Client.Model.Requests;
 using ProGaudi.Tarantool.Client.Model.Responses;
+using ProGaudi.Tarantool.Client.Utils;
 
 namespace ProGaudi.Tarantool.Client
 {
-    using System.Threading;
-
-    using ProGaudi.Tarantool.Client.Model;
-    using ProGaudi.Tarantool.Client.Utils;
-
     public class LogicalConnectionManager : ILogicalConnection
     {
         private readonly ClientOptions _clientOptions;
@@ -21,16 +21,38 @@ public class LogicalConnectionManager : ILogicalConnection
 
         private readonly AutoResetEvent _reconnectAvailable = new AutoResetEvent(true);
 
+        private Timer _timer;
+
+        private int _disposing;
+
         private const int connectionTimeout = 1000;
 
+        private const int connectionPingInterval = 1000;
+
         public LogicalConnectionManager(ClientOptions options)
         {
             _clientOptions = options;
         }
 
         public void Dispose()
         {
+            if (Interlocked.Exchange(ref _disposing, 1) > 0)
+            {
+                return;
+            }
+
             Interlocked.Exchange(ref _droppableLogicalConnection, null)?.Dispose();
+
+            var savedTimer = Interlocked.Exchange(ref _timer, null);
+            if (savedTimer != null)
+            {
+                using (var timerDisposedEvent = new ManualResetEvent(false))
+                {
+                    savedTimer.Dispose(timerDisposedEvent);
+                    // this will guarantee that all callbacks finished
+                    timerDisposedEvent.WaitOne();
+                }
+            }
         }
 
         public async Task Connect()
@@ -46,6 +68,40 @@ public async Task Connect()
             _connected.Set();
 
             _clientOptions.LogWriter?.WriteLine($"{nameof(LogicalConnectionManager)}: Connected...");
+
+            _timer = new Timer(x => CheckPing(), null, connectionPingInterval, Timeout.Infinite);
+        }
+
+        private static readonly PingRequest pingRequest = new PingRequest();
+
+        private void CheckPing()
+        {
+            LogicalConnection savedConnection = _droppableLogicalConnection;
+
+            if (savedConnection == null)
+            {
+                return;
+            }
+
+            try
+            {
+                Task task = savedConnection.SendRequestWithEmptyResponse(pingRequest);
+                if (!task.Wait(connectionTimeout) || task.Status != TaskStatus.RanToCompletion)
+                {
+                    savedConnection.Dispose();
+                }
+            }
+            catch (AggregateException ae)
+            {
+                savedConnection.Dispose();
+            }
+            finally
+            {
+                if (_disposing == 0)
+                {
+                    _timer?.Change(connectionPingInterval, Timeout.Infinite);
+                }
+            }
         }
 
         public bool IsConnected()

src/tarantool.client/Model/Requests/PingRequest.cs

@@ -0,0 +1,9 @@
+using ProGaudi.Tarantool.Client.Model.Enums;
+
+namespace ProGaudi.Tarantool.Client.Model.Requests
+{
+    public class PingRequest : IRequest
+    {
+        public CommandCode Code => CommandCode.Ping;
+    }
+}

src/tarantool.client/NetworkStreamPhysicalConnection.cs

@@ -42,7 +42,8 @@ public async Task Connect(ClientOptions options)
 
             _socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
             await ConnectAsync(_socket, singleNode.Uri.Host, singleNode.Uri.Port);
-            SetKeepAlive(true, 1000, 100);
+            // unfortunately does not worl under linux
+            // SetKeepAlive(true, 1000, 100);
             _stream = new NetworkStream(_socket, true);
             options.LogWriter?.WriteLine("Socket connection established.");
         }

src/tarantool.client/TarantoolConvertersRegistrator.cs

@@ -45,6 +45,7 @@ public static void Register(MsgPackContext context)
             context.RegisterGenericConverter(typeof(InsertReplacePacketConverter<>));
             context.RegisterGenericConverter(typeof(SelectPacketConverter<>));
             context.RegisterGenericConverter(typeof(UpsertPacketConverter<>));
+            context.RegisterConverter(new PingPacketConverter());
 
             context.RegisterGenericConverter(typeof(TupleConverter<>));
             context.RegisterGenericConverter(typeof(TupleConverter<,>));