Connection pool mechanism

Directory.Packages.props

@@ -18,5 +18,6 @@
       <PackageVersion Include="xunit.runner.visualstudio" Version="2.4.0" />
       <PackageVersion Include="JetBrains.Annotations" Version="2018.2.1" />
       <PackageVersion Include="Shouldly" Version="3.0.1" />
+      <PackageVersion Include="Moq" Version="4.20.70" />
     </ItemGroup>
   </Project>

src/progaudi.tarantool/Box.cs

@@ -1,5 +1,7 @@
-using System.Collections.Generic;
+using System;
+using System.Collections.Generic;
 using System.Threading.Tasks;
+using ProGaudi.Tarantool.Client.Core;
 using ProGaudi.Tarantool.Client.Model;
 using ProGaudi.Tarantool.Client.Model.Requests;
 using ProGaudi.Tarantool.Client.Model.Responses;
@@ -22,7 +24,10 @@ public Box(ClientOptions options)
             _clientOptions = options;
             TarantoolConvertersRegistrator.Register(options.MsgPackContext);
 
-            _logicalConnection = new LogicalConnectionManager(options);
+            var lcm = new LogicalConnectionManager(options);
+            lcm.ConnectionGoesDown += ConnectionWentDownHandler;
+            _logicalConnection = lcm;
+
             Metrics = new Metrics(_logicalConnection);
             Schema = new Schema(_logicalConnection);
         }
@@ -42,6 +47,8 @@ private set
                 _sqlReady = value.IsSqlAvailable();
             }
         }
+
+        public event EventHandler<ConnectionWentDownEventArgs> ConnectionGoesDown;
 
         public async Task Connect()
         {
@@ -163,5 +170,26 @@ public Task<DataResponse<TResponse[]>> ExecuteSql<TResponse>(string query, param
 
             return _logicalConnection.SendRequest<ExecuteSqlRequest, TResponse>(new ExecuteSqlRequest(query, parameters));
         }
+
+        public Task Do<TRequest>(TRequest request) where TRequest : IRequest
+        {
+            return _logicalConnection.SendRequest(request);
+        }
+
+        public Task<DataResponse<TResponse[]>> Do<TRequest, TResponse>(TRequest request) where TRequest : IRequest
+        {
+            return _logicalConnection.SendRequest<TRequest, TResponse>(request);
+        }
+
+        protected virtual void OnConnectionGoesDown(ConnectionWentDownEventArgs e)
+        {
+            var handler = ConnectionGoesDown;
+            handler?.Invoke(this, e);
+        }
+
+        private void ConnectionWentDownHandler(object sender, ConnectionWentDownEventArgs e)
+        {
+            OnConnectionGoesDown(e);
+        }
     }
 }

src/progaudi.tarantool/Core/ConnectionTimeoutThresholdReachedEventArgs.cs

@@ -0,0 +1,9 @@
+using System;
+
+namespace ProGaudi.Tarantool.Client.Core
+{
+    public class ConnectionTimeoutThresholdReachedEventArgs : EventArgs
+    {
+        public uint TimeoutCount { get; set; }
+    }
+}

src/progaudi.tarantool/Core/ConnectionWentDownEventArgs.cs

@@ -0,0 +1,21 @@
+using System;
+using System.Linq;
+using ProGaudi.Tarantool.Client.Model;
+
+namespace ProGaudi.Tarantool.Client.Core
+{
+    public class ConnectionWentDownEventArgs : EventArgs
+    {
+        public ConnectionWentDownEventArgs(TarantoolNode node)
+        {
+            Node = node;
+        }
+
+        public ConnectionWentDownEventArgs(ClientOptions clientOptions)
+        {
+            Node = clientOptions.ConnectionOptions.Nodes.FirstOrDefault();
+        }
+
+        public TarantoolNode Node { get; private set; }
+    }
+}

src/progaudi.tarantool/Core/ConnectionWentUpEventArgs.cs

@@ -0,0 +1,10 @@
+using System;
+
+namespace ProGaudi.Tarantool.Client.Core
+{
+    public class ConnectionWentUpEventArgs : EventArgs
+    {
+        public IBox Box { get; set; }
+        public string ConnectionKey { get; set; }
+    }
+}

src/progaudi.tarantool/ILogicalConnection.cs → src/progaudi.tarantool/Core/ILogicalConnection.cs

@@ -1,10 +1,9 @@
 using System;
 using System.Threading.Tasks;
-
 using ProGaudi.Tarantool.Client.Model.Requests;
 using ProGaudi.Tarantool.Client.Model.Responses;
 
-namespace ProGaudi.Tarantool.Client
+namespace ProGaudi.Tarantool.Client.Core
 {
     public interface ILogicalConnection : IDisposable
     {

src/progaudi.tarantool/IPhysicalConnection.cs → src/progaudi.tarantool/Core/IPhysicalConnection.cs

@@ -2,7 +2,7 @@
 using System.Threading.Tasks;
 using ProGaudi.Tarantool.Client.Model;
 
-namespace ProGaudi.Tarantool.Client
+namespace ProGaudi.Tarantool.Client.Core
 {
     public interface IPhysicalConnection : IDisposable
     {

src/progaudi.tarantool/IRequestWriter.cs → src/progaudi.tarantool/Core/IRequestWriter.cs

@@ -1,7 +1,6 @@
 using System;
-using System.Threading.Tasks;
 
-namespace ProGaudi.Tarantool.Client
+namespace ProGaudi.Tarantool.Client.Core
 {
     internal interface IRequestWriter : IDisposable
     {

src/progaudi.tarantool/IResponseReader.cs → src/progaudi.tarantool/Core/IResponseReader.cs

@@ -1,10 +1,9 @@
 using System;
 using System.IO;
 using System.Threading.Tasks;
-
 using ProGaudi.Tarantool.Client.Model;
 
-namespace ProGaudi.Tarantool.Client
+namespace ProGaudi.Tarantool.Client.Core
 {
     public interface IResponseReader : IDisposable
     {

src/progaudi.tarantool/LogicalConnection.cs → src/progaudi.tarantool/Core/LogicalConnection.cs

@@ -3,16 +3,14 @@
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
-
 using ProGaudi.MsgPack.Light;
-
 using ProGaudi.Tarantool.Client.Model;
 using ProGaudi.Tarantool.Client.Model.Headers;
 using ProGaudi.Tarantool.Client.Model.Requests;
 using ProGaudi.Tarantool.Client.Model.Responses;
 using ProGaudi.Tarantool.Client.Utils;
 
-namespace ProGaudi.Tarantool.Client
+namespace ProGaudi.Tarantool.Client.Core
 {
     internal class LogicalConnection : ILogicalConnection
     {
@@ -43,11 +41,7 @@ public LogicalConnection(ClientOptions options, RequestIdCounter requestIdCounte
             _requestWriter = new RequestWriter(_clientOptions, _physicalConnection);
         }
 
-        public uint PingsFailedByTimeoutCount
-        {
-            get;
-            private set;
-        }
+        public uint PingsFailedByTimeoutCount { get; private set; }
 
         public void Dispose()
         {
@@ -123,6 +117,13 @@ public async Task<byte[]> SendRawRequest<TRequest>(TRequest request, TimeSpan? t
         {
             return (await SendRequestImpl(request, timeout).ConfigureAwait(false)).ToArray();
         }
+        public event EventHandler<ConnectionTimeoutThresholdReachedEventArgs> ConnectionTimeoutThresholdReached;
+
+        protected virtual void OnConnectionTimeoutThresholdReached(ConnectionTimeoutThresholdReachedEventArgs e)
+        {
+            var handler = ConnectionTimeoutThresholdReached;
+            handler?.Invoke(this, e);
+        }
 
         private async Task LoginIfNotGuest(GreetingsResponse greetings)
         {
@@ -193,6 +194,11 @@ private async Task<MemoryStream> SendRequestImpl<TRequest>(TRequest request, Tim
             catch (TimeoutException)
             {
                 PingsFailedByTimeoutCount++;
+                if (PingsFailedByTimeoutCount > 5)
+                {
+                    OnConnectionTimeoutThresholdReached(new ConnectionTimeoutThresholdReachedEventArgs
+                        { TimeoutCount = PingsFailedByTimeoutCount });
+                }
                 throw;
             }
         }

src/progaudi.tarantool/LogicalConnectionManager.cs → src/progaudi.tarantool/Core/LogicalConnectionManager.cs

@@ -1,13 +1,12 @@
 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
+namespace ProGaudi.Tarantool.Client.Core
 {
     public class LogicalConnectionManager : ILogicalConnection
     {
@@ -46,9 +45,11 @@ public LogicalConnectionManager(ClientOptions options)
 
             _pingTimeout = _clientOptions.ConnectionOptions.PingCheckTimeout;
         }
-
+        
         public uint PingsFailedByTimeoutCount => _droppableLogicalConnection?.PingsFailedByTimeoutCount ?? 0;
 
+        public event EventHandler<ConnectionWentDownEventArgs> ConnectionGoesDown;
+
         public void Dispose()
         {
             if (Interlocked.Exchange(ref _disposing, 1) > 0)
@@ -84,6 +85,7 @@ public async Task Connect()
                 _clientOptions.LogWriter?.WriteLine($"{nameof(LogicalConnectionManager)}: Connecting...");
 
                 var newConnection = new LogicalConnection(_clientOptions, _requestIdCounter);
+                newConnection.ConnectionTimeoutThresholdReached += ConnectionTimeoutThresholdReached;
                 await newConnection.Connect().ConfigureAwait(false);;
                 Interlocked.Exchange(ref _droppableLogicalConnection, newConnection)?.Dispose();
 
@@ -102,6 +104,20 @@ public async Task Connect()
             }
         }
 
+        private void ConnectionTimeoutThresholdReached(object sender, ConnectionTimeoutThresholdReachedEventArgs e)
+        {
+            _clientOptions.LogWriter?.WriteLine($"{nameof(LogicalConnectionManager)}: Connection timeout threshold reached. Dropping current connection.");
+            var connection = (LogicalConnection)sender;
+            connection?.Dispose();
+            OnConnectionGoesDown(new ConnectionWentDownEventArgs(_clientOptions));
+        }
+
+        protected virtual void OnConnectionGoesDown(ConnectionWentDownEventArgs e)
+        {
+            var handler = ConnectionGoesDown;
+            handler?.Invoke(this, e);
+        }
+
         private static readonly PingRequest _pingRequest = new PingRequest();
 
         private void CheckPing()
@@ -119,6 +135,7 @@ private void CheckPing()
             {
                 _clientOptions.LogWriter?.WriteLine($"{nameof(LogicalConnectionManager)}: Ping failed with exception: {e.Message}. Dropping current connection.");
                 _droppableLogicalConnection?.Dispose();
+                OnConnectionGoesDown(new ConnectionWentDownEventArgs(_clientOptions));
             }
             finally
             {

src/progaudi.tarantool/NetworkStreamPhysicalConnection.cs → src/progaudi.tarantool/Core/NetworkStreamPhysicalConnection.cs

@@ -3,15 +3,13 @@
 using System.Linq;
 using System.Net.Sockets;
 using System.Threading.Tasks;
-
 using ProGaudi.Tarantool.Client.Model;
 using ProGaudi.Tarantool.Client.Utils;
-
 #if PROGAUDI_NETCORE
 using System.Net;
 #endif
 
-namespace ProGaudi.Tarantool.Client
+namespace ProGaudi.Tarantool.Client.Core
 {
     internal class NetworkStreamPhysicalConnection : IPhysicalConnection
     {

src/progaudi.tarantool/RequestIdCounter.cs → src/progaudi.tarantool/Core/RequestIdCounter.cs

@@ -1,8 +1,7 @@
 using System.Threading;
-
 using ProGaudi.Tarantool.Client.Model;
 
-namespace ProGaudi.Tarantool.Client
+namespace ProGaudi.Tarantool.Client.Core
 {
     public class RequestIdCounter
     {

src/progaudi.tarantool/RequestWriter.cs → src/progaudi.tarantool/Core/RequestWriter.cs

@@ -1,12 +1,9 @@
 using System;
 using System.Collections.Generic;
-using System.Diagnostics;
-using System.IO;
 using System.Threading;
-using System.Threading.Tasks;
 using ProGaudi.Tarantool.Client.Model;
 
-namespace ProGaudi.Tarantool.Client
+namespace ProGaudi.Tarantool.Client.Core
 {
     internal class RequestWriter : IRequestWriter
     {

src/progaudi.tarantool/ResponseReader.cs → src/progaudi.tarantool/Core/ResponseReader.cs

@@ -4,17 +4,15 @@
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
-
 using JetBrains.Annotations;
-
 using ProGaudi.MsgPack.Light;
 using ProGaudi.Tarantool.Client.Model;
 using ProGaudi.Tarantool.Client.Model.Enums;
 using ProGaudi.Tarantool.Client.Model.Headers;
 using ProGaudi.Tarantool.Client.Model.Responses;
 using ProGaudi.Tarantool.Client.Utils;
 
-namespace ProGaudi.Tarantool.Client
+namespace ProGaudi.Tarantool.Client.Core
 {
     internal class ResponseReader : IResponseReader
     {

src/progaudi.tarantool/IBox.cs

@@ -1,6 +1,8 @@
 using System;
 using System.Threading.Tasks;
+using ProGaudi.Tarantool.Client.Core;
 using ProGaudi.Tarantool.Client.Model;
+using ProGaudi.Tarantool.Client.Model.Requests;
 using ProGaudi.Tarantool.Client.Model.Responses;
 
 namespace ProGaudi.Tarantool.Client
@@ -47,5 +49,11 @@ public interface IBox : IDisposable
         Task<DataResponse<TResponse[]>> ExecuteSql<TResponse>(string query, params SqlParameter[] parameters);
 
         Task<DataResponse> ExecuteSql(string query, params SqlParameter[] parameters);
+
+        Task Do<TRequest>(TRequest request) where TRequest : IRequest;
+
+        Task<DataResponse<TResponse[]>> Do<TRequest, TResponse>(TRequest request) where TRequest : IRequest;
+
+        event EventHandler<ConnectionWentDownEventArgs> ConnectionGoesDown;
     }
 }

src/progaudi.tarantool/IIndex.cs

@@ -1,5 +1,6 @@
 using System.Collections.Generic;
 using System.Threading.Tasks;
+using ProGaudi.Tarantool.Client.Core;
 using ProGaudi.Tarantool.Client.Model;
 using ProGaudi.Tarantool.Client.Model.Enums;
 using ProGaudi.Tarantool.Client.Model.Responses;

src/progaudi.tarantool/ISpace.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Threading.Tasks;
+using ProGaudi.Tarantool.Client.Core;
 using ProGaudi.Tarantool.Client.Model;
 using ProGaudi.Tarantool.Client.Model.Enums;
 using ProGaudi.Tarantool.Client.Model.Responses;

src/progaudi.tarantool/Index.cs

@@ -2,7 +2,7 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
-
+using ProGaudi.Tarantool.Client.Core;
 using ProGaudi.Tarantool.Client.Model;
 using ProGaudi.Tarantool.Client.Model.Enums;
 using ProGaudi.Tarantool.Client.Model.Requests;

src/progaudi.tarantool/Model/BoxInfo.cs

@@ -11,9 +11,9 @@ public class BoxInfo
 
         public long Pid { get; private set; }
 
-        public bool ReadOnly { get; private set; }
+        public virtual bool ReadOnly { get; private set; }
 
-        public Guid Uuid { get; private set; }
+        public virtual Guid Uuid { get; private set; }
 
         public TarantoolVersion Version { get; private set; }