Merge pull request #141 from rabbitmq/rabbitmq-dotnet-client-140

michaelklishin · michaelklishin · commit 4b8dd2f02017 · 2015-12-01T16:52:05.000+03:00
Introduce ConnectionFactory.SocketReadTimeout and SocketWriteTimeout
diff --git a/projects/client/RabbitMQ.Client.WinRT/src/client/impl/SocketFrameHandler.cs b/projects/client/RabbitMQ.Client.WinRT/src/client/impl/SocketFrameHandler.cs
@@ -71,12 +71,14 @@ public class SocketFrameHandler : IFrameHandler
 
         public SocketFrameHandler(AmqpTcpEndpoint endpoint,
             Func<StreamSocket> socketFactory,
-            int timeout)
+            int connectionTimeout,
+            int _readTimeout,
+            int _writeTimeout)
         {
             Endpoint = endpoint;
 
             m_socket = socketFactory();
-            Connect(m_socket, endpoint, timeout);
+            Connect(m_socket, endpoint, connectionTimeout);
 
             if (endpoint.Ssl.Enabled)
             {
@@ -131,14 +133,23 @@ public int RemotePort
             }
         }
 
-        public int Timeout
+        public int ReadTimeout
         {
             set
             {
                 // Ignored, timeouts over streams on WinRT
                 // are much trickier to get right.
                 //
-                // Heartbeat implementation is in Connection.
+                // See heartbeats implementation is in Connection.
+            }
+        }
+
+        public int WriteTimeout
+        {
+            set
+            {
+                // Ignored, timeouts over streams on WinRT
+                // are much trickier to get right.
             }
         }
 
diff --git a/projects/client/RabbitMQ.Client/src/client/api/ConnectionFactory.cs b/projects/client/RabbitMQ.Client/src/client/api/ConnectionFactory.cs
@@ -202,7 +202,7 @@ public TimeSpan ContinuationTimeout
         public int Port = AmqpTcpEndpoint.UseDefaultPort;
 
         /// <summary>
-        /// The AMQP protocol to be used. Currently 0-9-1.
+        /// Protocol used, only AMQP 0-9-1 is supported in modern versions.
         /// </summary>
         public IProtocol Protocol = Protocols.DefaultProtocol;
 
@@ -211,6 +211,16 @@ public TimeSpan ContinuationTimeout
         /// </summary>
         public int RequestedConnectionTimeout = DefaultConnectionTimeout;
 
+        /// <summary>
+        /// Timeout setting for socket read operations (in milliseconds).
+        /// </summary>
+        public int SocketReadTimeout = DefaultConnectionTimeout;
+
+        /// <summary>
+        /// Timeout setting for socket write operations (in milliseconds).
+        /// </summary>
+        public int SocketWriteTimeout = DefaultConnectionTimeout;
+
         /// <summary>
         /// Ssl options setting.
         /// </summary>
@@ -380,12 +390,25 @@ public IConnection CreateConnection(IList<string> hostnames)
 
         public IFrameHandler CreateFrameHandler()
         {
-            return Protocols.DefaultProtocol.CreateFrameHandler(Endpoint, SocketFactory, RequestedConnectionTimeout);
+            var fh = Protocols.DefaultProtocol.CreateFrameHandler(Endpoint, SocketFactory,
+                RequestedConnectionTimeout, SocketReadTimeout, SocketWriteTimeout);
+            return ConfigureFrameHandler(fh);
         }
 
         public IFrameHandler CreateFrameHandler(AmqpTcpEndpoint endpoint)
         {
-            return Protocols.DefaultProtocol.CreateFrameHandler(endpoint, SocketFactory, RequestedConnectionTimeout);
+            var fh = Protocols.DefaultProtocol.CreateFrameHandler(endpoint, SocketFactory,
+                RequestedConnectionTimeout, SocketReadTimeout, SocketWriteTimeout);
+            return ConfigureFrameHandler(fh);
+        }
+
+        private IFrameHandler ConfigureFrameHandler(IFrameHandler fh)
+        {
+            // make sure socket timeouts are higher than heartbeat
+            fh.ReadTimeout  = Math.Max(SocketReadTimeout,  RequestedHeartbeat * 1000);
+            fh.WriteTimeout = Math.Max(SocketWriteTimeout, RequestedHeartbeat * 1000);
+            // TODO: add user-provided configurator, like in the Java client
+            return fh;
         }
 
         private void SetUri(Uri uri)
diff --git a/projects/client/RabbitMQ.Client/src/client/api/IProtocol.cs b/projects/client/RabbitMQ.Client/src/client/api/IProtocol.cs
@@ -99,7 +99,7 @@ public interface IProtocol
         ///  Construct a frame handler for a given endpoint.
         ///  </summary>
         /// <param name="socketFactory">Socket factory method.</param>
-        /// <param name="timeout">Timeout in milliseconds.</param>
+        /// <param name="connectionTimeout">Timeout in milliseconds.</param>
         /// <param name="endpoint">Represents a TCP-addressable AMQP peer: a host name and port number.</param>
         IFrameHandler CreateFrameHandler(
             AmqpTcpEndpoint endpoint, 
@@ -108,7 +108,9 @@ IFrameHandler CreateFrameHandler(
 #else
             Func<StreamSocket> socketFactory,
 #endif
-            int timeout);
+            int connectionTimeout,
+            int readTimeout,
+            int writeTimeout);
         /// <summary>
         /// Construct a protocol model atop a given session.
         /// </summary>
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/Connection.cs b/projects/client/RabbitMQ.Client/src/client/impl/Connection.cs
@@ -263,7 +263,7 @@ public ushort Heartbeat
                 // timers fire at slightly below half the interval to avoid race
                 // conditions
                 m_heartbeatTimeSpan = TimeSpan.FromMilliseconds((value * 1000) / 4);
-                m_frameHandler.Timeout = value * 1000 * 2;
+                m_frameHandler.ReadTimeout = value * 1000 * 2;
             }
         }
 
@@ -441,7 +441,7 @@ public void ClosingLoop()
         {
             try
             {
-                m_frameHandler.Timeout = 0;
+                m_frameHandler.ReadTimeout = 0;
                 // Wait for response/socket closure or timeout
                 while (!m_closed)
                 {
@@ -1247,7 +1247,7 @@ protected void StartAndTune()
             var connectionStartCell = new BlockingCell();
             m_model0.m_connectionStartCell = connectionStartCell;
             m_model0.HandshakeContinuationTimeout = m_factory.HandshakeContinuationTimeout;
-            m_frameHandler.Timeout = m_factory.HandshakeContinuationTimeout.Milliseconds;
+            m_frameHandler.ReadTimeout = m_factory.HandshakeContinuationTimeout.Milliseconds;
             m_frameHandler.SendHeader();
 
             var connectionStart = (ConnectionStartDetails)
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/IFrameHandler.cs b/projects/client/RabbitMQ.Client/src/client/impl/IFrameHandler.cs
@@ -61,7 +61,10 @@ public interface IFrameHandler
         int RemotePort { get; }
 
         ///<summary>Socket read timeout, in milliseconds. Zero signals "infinity".</summary>
-        int Timeout { set; }
+        int ReadTimeout { set; }
+
+        ///<summary>Socket write timeout, in milliseconds. Zero signals "infinity".</summary>
+        int WriteTimeout { set; }
 
         void Close();
 
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/ProtocolBase.cs b/projects/client/RabbitMQ.Client/src/client/impl/ProtocolBase.cs
@@ -48,7 +48,6 @@
 #endif
 
 using RabbitMQ.Client.Impl;
-using RabbitMQ.Client;
 using RabbitMQ.Util;
 
 namespace RabbitMQ.Client.Framing.Impl
@@ -151,9 +150,12 @@ public IFrameHandler CreateFrameHandler(
 #else
             Func<StreamSocket> socketFactory,
 #endif
-            int timeout)
+            int connectionTimeout,
+            int readTimeout,
+            int writeTimeout)
         {
-            return new SocketFrameHandler(endpoint, socketFactory, timeout);
+            return new SocketFrameHandler(endpoint, socketFactory,
+                connectionTimeout, readTimeout, writeTimeout);
         }
 
         public IModel CreateModel(ISession session)
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/SocketFrameHandler.cs b/projects/client/RabbitMQ.Client/src/client/impl/SocketFrameHandler.cs
@@ -66,7 +66,7 @@ public class SocketFrameHandler : IFrameHandler
 
         public SocketFrameHandler(AmqpTcpEndpoint endpoint,
             Func<AddressFamily, ITcpClient> socketFactory,
-            int timeout)
+            int connectionTimeout, int readTimeout, int writeTimeout)
         {
             Endpoint = endpoint;
             m_socket = null;
@@ -75,7 +75,7 @@ public SocketFrameHandler(AmqpTcpEndpoint endpoint,
                 try
                 {
                     m_socket = socketFactory(AddressFamily.InterNetworkV6);
-                    Connect(m_socket, endpoint, timeout);
+                    Connect(m_socket, endpoint, connectionTimeout);
                 }
                 catch (ConnectFailureException) // could not connect using IPv6
                 {
@@ -91,13 +91,12 @@ public SocketFrameHandler(AmqpTcpEndpoint endpoint,
             if (m_socket == null)
             {
                 m_socket = socketFactory(AddressFamily.InterNetwork);
-                Connect(m_socket, endpoint, timeout);
+                Connect(m_socket, endpoint, connectionTimeout);
             }
 
             Stream netstream = m_socket.GetStream();
-            // make sure the socket timeout is greater than heartbeat timeout
-            netstream.ReadTimeout = timeout;
-            netstream.WriteTimeout = timeout;
+            netstream.ReadTimeout  = readTimeout;
+            netstream.WriteTimeout = writeTimeout;
 
             if (endpoint.Ssl.Enabled)
             {
@@ -113,6 +112,8 @@ public SocketFrameHandler(AmqpTcpEndpoint endpoint,
             }
             m_reader = new NetworkBinaryReader(new BufferedStream(netstream));
             m_writer = new NetworkBinaryWriter(new BufferedStream(netstream));
+
+            m_writeableStateTimeout = writeTimeout;
         }
 
         public AmqpTcpEndpoint Endpoint { get; set; }
@@ -137,17 +138,15 @@ public int RemotePort
             get { return ((IPEndPoint)LocalEndPoint).Port; }
         }
 
-        public int Timeout
+        public int ReadTimeout
         {
             set
             {
                 try
                 {
                     if (m_socket.Connected)
-                    {
-                        // make sure the socket timeout is greater than heartbeat interval
-                        m_socket.ReceiveTimeout = value * 4;
-                        m_writeableStateTimeout = value * 4;
+                    {                        
+                        m_socket.ReceiveTimeout = value;
                     }
                 }
 #pragma warning disable 0168
@@ -159,6 +158,15 @@ public int Timeout
             }
         }
 
+        public int WriteTimeout
+        {
+            set
+            {
+                m_writeableStateTimeout = value;
+                m_socket.Client.SendTimeout = value;
+            }
+        }
+
         public void Close()
         {
             lock (_semaphore)

Original file line number	Diff line number	Diff line change
`@@ -71,12 +71,14 @@ public class SocketFrameHandler : IFrameHandler`
`71`	`71`
`72`	`72`	`public SocketFrameHandler(AmqpTcpEndpoint endpoint,`
`73`	`73`	`Func<StreamSocket> socketFactory,`
`74`		`- int timeout)`
	`74`	`+ int connectionTimeout,`
	`75`	`+ int _readTimeout,`
	`76`	`+ int _writeTimeout)`
`75`	`77`	`{`
`76`	`78`	`Endpoint = endpoint;`
`77`	`79`
`78`	`80`	`m_socket = socketFactory();`
`79`		`- Connect(m_socket, endpoint, timeout);`
	`81`	`+ Connect(m_socket, endpoint, connectionTimeout);`
`80`	`82`
`81`	`83`	`if (endpoint.Ssl.Enabled)`
`82`	`84`	`{`
`@@ -131,14 +133,23 @@ public int RemotePort`
`131`	`133`	`}`
`132`	`134`	`}`
`133`	`135`
`134`		`- public int Timeout`
	`136`	`+ public int ReadTimeout`
`135`	`137`	`{`
`136`	`138`	`set`
`137`	`139`	`{`
`138`	`140`	`// Ignored, timeouts over streams on WinRT`
`139`	`141`	`// are much trickier to get right.`
`140`	`142`	`//`
`141`		`- // Heartbeat implementation is in Connection.`
	`143`	`+ // See heartbeats implementation is in Connection.`
	`144`	`+ }`
	`145`	`+ }`
	`146`	`+`
	`147`	`+ public int WriteTimeout`
	`148`	`+ {`
	`149`	`+ set`
	`150`	`+ {`
	`151`	`+ // Ignored, timeouts over streams on WinRT`
	`152`	`+ // are much trickier to get right.`
`142`	`153`	`}`
`143`	`154`	`}`
`144`	`155`
Original file line number	Diff line number	Diff line change
`@@ -263,7 +263,7 @@ public ushort Heartbeat`
`263`	`263`	`// timers fire at slightly below half the interval to avoid race`
`264`	`264`	`// conditions`
`265`	`265`	`m_heartbeatTimeSpan = TimeSpan.FromMilliseconds((value * 1000) / 4);`
`266`		`- m_frameHandler.Timeout = value * 1000 * 2;`
	`266`	`+ m_frameHandler.ReadTimeout = value * 1000 * 2;`
`267`	`267`	`}`
`268`	`268`	`}`
`269`	`269`
`@@ -441,7 +441,7 @@ public void ClosingLoop()`
`441`	`441`	`{`
`442`	`442`	`try`
`443`	`443`	`{`
`444`		`- m_frameHandler.Timeout = 0;`
	`444`	`+ m_frameHandler.ReadTimeout = 0;`
`445`	`445`	`// Wait for response/socket closure or timeout`
`446`	`446`	`while (!m_closed)`
`447`	`447`	`{`
`@@ -1247,7 +1247,7 @@ protected void StartAndTune()`
`1247`	`1247`	`var connectionStartCell = new BlockingCell();`
`1248`	`1248`	`m_model0.m_connectionStartCell = connectionStartCell;`
`1249`	`1249`	`m_model0.HandshakeContinuationTimeout = m_factory.HandshakeContinuationTimeout;`
`1250`		`- m_frameHandler.Timeout = m_factory.HandshakeContinuationTimeout.Milliseconds;`
	`1250`	`+ m_frameHandler.ReadTimeout = m_factory.HandshakeContinuationTimeout.Milliseconds;`
`1251`	`1251`	`m_frameHandler.SendHeader();`
`1252`	`1252`
`1253`	`1253`	`var connectionStart = (ConnectionStartDetails)`
Original file line number	Diff line number	Diff line change
`@@ -66,7 +66,7 @@ public class SocketFrameHandler : IFrameHandler`
`66`	`66`
`67`	`67`	`public SocketFrameHandler(AmqpTcpEndpoint endpoint,`
`68`	`68`	`Func<AddressFamily, ITcpClient> socketFactory,`
`69`		`- int timeout)`
	`69`	`+ int connectionTimeout, int readTimeout, int writeTimeout)`
`70`	`70`	`{`
`71`	`71`	`Endpoint = endpoint;`
`72`	`72`	`m_socket = null;`
`@@ -75,7 +75,7 @@ public SocketFrameHandler(AmqpTcpEndpoint endpoint,`
`75`	`75`	`try`
`76`	`76`	`{`
`77`	`77`	`m_socket = socketFactory(AddressFamily.InterNetworkV6);`
`78`		`- Connect(m_socket, endpoint, timeout);`
	`78`	`+ Connect(m_socket, endpoint, connectionTimeout);`
`79`	`79`	`}`
`80`	`80`	`catch (ConnectFailureException) // could not connect using IPv6`
`81`	`81`	`{`
`@@ -91,13 +91,12 @@ public SocketFrameHandler(AmqpTcpEndpoint endpoint,`
`91`	`91`	`if (m_socket == null)`
`92`	`92`	`{`
`93`	`93`	`m_socket = socketFactory(AddressFamily.InterNetwork);`
`94`		`- Connect(m_socket, endpoint, timeout);`
	`94`	`+ Connect(m_socket, endpoint, connectionTimeout);`
`95`	`95`	`}`
`96`	`96`
`97`	`97`	`Stream netstream = m_socket.GetStream();`
`98`		`- // make sure the socket timeout is greater than heartbeat timeout`
`99`		`- netstream.ReadTimeout = timeout;`
`100`		`- netstream.WriteTimeout = timeout;`
	`98`	`+ netstream.ReadTimeout = readTimeout;`
	`99`	`+ netstream.WriteTimeout = writeTimeout;`
`101`	`100`
`102`	`101`	`if (endpoint.Ssl.Enabled)`
`103`	`102`	`{`
`@@ -113,6 +112,8 @@ public SocketFrameHandler(AmqpTcpEndpoint endpoint,`
`113`	`112`	`}`
`114`	`113`	`m_reader = new NetworkBinaryReader(new BufferedStream(netstream));`
`115`	`114`	`m_writer = new NetworkBinaryWriter(new BufferedStream(netstream));`
	`115`	`+`
	`116`	`+ m_writeableStateTimeout = writeTimeout;`
`116`	`117`	`}`
`117`	`118`
`118`	`119`	`public AmqpTcpEndpoint Endpoint { get; set; }`
`@@ -137,17 +138,15 @@ public int RemotePort`
`137`	`138`	`get { return ((IPEndPoint)LocalEndPoint).Port; }`
`138`	`139`	`}`
`139`	`140`
`140`		`- public int Timeout`
	`141`	`+ public int ReadTimeout`
`141`	`142`	`{`
`142`	`143`	`set`
`143`	`144`	`{`
`144`	`145`	`try`
`145`	`146`	`{`
`146`	`147`	`if (m_socket.Connected)`
`147`		`- {`
`148`		`- // make sure the socket timeout is greater than heartbeat interval`
`149`		`- m_socket.ReceiveTimeout = value * 4;`
`150`		`- m_writeableStateTimeout = value * 4;`
	`148`	`+ {`
	`149`	`+ m_socket.ReceiveTimeout = value;`
`151`	`150`	`}`
`152`	`151`	`}`
`153`	`152`	`#pragma warning disable 0168`
`@@ -159,6 +158,15 @@ public int Timeout`
`159`	`158`	`}`
`160`	`159`	`}`
`161`	`160`
	`161`	`+ public int WriteTimeout`
	`162`	`+ {`
	`163`	`+ set`
	`164`	`+ {`
	`165`	`+ m_writeableStateTimeout = value;`
	`166`	`+ m_socket.Client.SendTimeout = value;`
	`167`	`+ }`
	`168`	`+ }`
	`169`	`+`
`162`	`170`	`public void Close()`
`163`	`171`	`{`
`164`	`172`	`lock (_semaphore)`