Start heartbeat timers after connection.tune

Also simplify socket timeout configuration and
flush the socket after every heartbeat sent.

Author: michaelklishin

projects/client/RabbitMQ.Client/src/client/impl/Connection.cs

@@ -122,7 +122,6 @@ public Connection(IConnectionFactory factory, bool insist, IFrameHandler frameHa
             StartMainLoop(factory.UseBackgroundThreadsForIO);
             Open(insist);
 
-            StartHeartbeatTimers();
             AppDomain.CurrentDomain.DomainUnload += HandleDomainUnload;
         }
 
@@ -242,10 +241,10 @@ public ushort Heartbeat
             set
             {
                 m_heartbeat = value;
-                // timers fire at half the interval to avoid race
+                // timers fire at slightly below half the interval to avoid race
                 // conditions
-                m_heartbeatTimeSpan = TimeSpan.FromMilliseconds((value * 1000) / 2.0);
-                m_frameHandler.Timeout = (value * 1000) / 2;
+                m_heartbeatTimeSpan = TimeSpan.FromMilliseconds((value * 1000) / 4);
+                m_frameHandler.Timeout = value * 1000 * 2;
             }
         }
 
@@ -477,7 +476,7 @@ public void FinishClose()
             m_heartbeatRead.Set();
             m_heartbeatWrite.Set();
             m_closed = true;
-            StopHeartbeatTimers();
+            MaybeStopHeartbeatTimers();
 
             m_frameHandler.Close();
             m_model0.SetCloseReason(m_closeReason);
@@ -896,15 +895,15 @@ public bool SetCloseReason(ShutdownEventArgs reason)
             }
         }
 
-        public void StartHeartbeatTimers()
+        public void MaybeStartHeartbeatTimers()
         {
             if (Heartbeat != 0)
             {
                 _heartbeatWriteTimer = new Timer(HeartbeatWriteTimerCallback);
-                _heartbeatWriteTimer.Change(TimeSpan.FromMilliseconds(0), TimeSpan.FromMilliseconds(-1));
+                _heartbeatWriteTimer.Change(TimeSpan.FromMilliseconds(200), m_heartbeatTimeSpan);
 
                 _heartbeatReadTimer = new Timer(HeartbeatReadTimerCallback);
-                _heartbeatReadTimer.Change(TimeSpan.FromMilliseconds(0), TimeSpan.FromMilliseconds(-1));
+                _heartbeatReadTimer.Change(TimeSpan.FromMilliseconds(200), m_heartbeatTimeSpan);
             }
         }
 
@@ -930,8 +929,11 @@ public void HeartbeatReadTimerCallback(object state)
                     m_missedHeartbeats = 0;
                 }
 
-                // Has to miss two full heartbeats to force socket close
-                if (m_missedHeartbeats > 1)
+                // We check against 8 = 2 * 4 because we need to wait for at
+                // least two complete heartbeat setting intervals before
+                // complaining, and we've set the socket timeout to a quarter
+                // of the heartbeat setting in setHeartbeat above.
+                if (m_missedHeartbeats > 2 * 4)
                 {
                     String description = String.Format("Heartbeat missing with heartbeat == {0} seconds", m_heartbeat);
                     var eose = new EndOfStreamException(description);
@@ -960,10 +962,8 @@ public void HeartbeatWriteTimerCallback(object state)
             {
                 if (!m_closed)
                 {
-                    if (!m_heartbeatWrite.WaitOne(0, false))
-                    {
-                        WriteFrame(m_heartbeatFrame);
-                    }
+                    WriteFrame(m_heartbeatFrame);
+                    m_frameHandler.Flush();
                 }
             }
             catch (Exception e)
@@ -981,13 +981,9 @@ public void HeartbeatWriteTimerCallback(object state)
                 TerminateMainloop();
                 FinishClose();
             }
-            else
-            {
-                _heartbeatWriteTimer.Change(Heartbeat * 1000, Timeout.Infinite);
-            }
         }
 
-        protected void StopHeartbeatTimers()
+        protected void MaybeStopHeartbeatTimers()
         {
             if(_heartbeatReadTimer != null)
             {
@@ -1004,7 +1000,7 @@ protected void StopHeartbeatTimers()
         ///</remarks>
         public void TerminateMainloop()
         {
-            StopHeartbeatTimers();
+            MaybeStopHeartbeatTimers();
             m_running = false;
         }
 
@@ -1089,7 +1085,7 @@ public void HandleConnectionUnblocked()
 
         void IDisposable.Dispose()
         {
-            StopHeartbeatTimers();
+            MaybeStopHeartbeatTimers();
             Abort();
             if (ShutdownReport.Count > 0)
             {
@@ -1219,6 +1215,8 @@ protected void StartAndTune()
                 heartbeat);
 
             m_inConnectionNegotiation = false;
+            // now we can start heartbeat timers
+            MaybeStartHeartbeatTimers();
         }
 
         private static uint NegotiatedMaxValue(uint clientValue, uint serverValue)

projects/client/RabbitMQ.Client/src/client/impl/IFrameHandler.cs

@@ -68,5 +68,7 @@ public interface IFrameHandler
         void SendHeader();
 
         void WriteFrame(Frame frame);
+
+        void Flush();
     }
 }

projects/client/RabbitMQ.Client/src/client/impl/SocketFrameHandler.cs

@@ -95,8 +95,8 @@ public SocketFrameHandler(AmqpTcpEndpoint endpoint,
 
             Stream netstream = m_socket.GetStream();
             // make sure the socket timeout is greater than heartbeat timeout
-            netstream.ReadTimeout = timeout * 4;
-            netstream.WriteTimeout = timeout * 4;
+            netstream.ReadTimeout = timeout;
+            netstream.WriteTimeout = timeout;
 
             if (endpoint.Ssl.Enabled)
             {
@@ -226,6 +226,14 @@ public void WriteFrame(Frame frame)
             }
         }
 
+        public void Flush()
+        {
+            lock (m_writer)
+            {
+                m_writer.Flush();
+            }
+        }
+
         private void Connect(TcpClient socket, AmqpTcpEndpoint endpoint, int timeout)
         {
             IAsyncResult ar = null;