moved heartbeating into a separate thread

Author: Rob Harrop

src/com/rabbitmq/client/impl/AMQConnection.java

@@ -37,6 +37,9 @@
 import java.util.Map;
 import java.util.HashMap;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Address;
@@ -122,6 +125,9 @@ public static Map<String, Object> defaultClientProperties() {
     /** Flag indicating whether the client received Connection.Close message from the broker */
     private boolean _brokerInitiatedShutdown = false;
 
+    /** Manages heartbeats for this connection */
+    private final Heartbeater heartbeater;
+
     /**
      * Protected API - respond, in the driver thread, to a ShutdownSignal.
      * @param channel the channel to disconnect
@@ -211,6 +217,7 @@ public AMQConnection(ConnectionFactory factory,
         _clientProperties = new HashMap<String, Object>(factory.getClientProperties());
 
         this.factory = factory;
+        this.heartbeater = new Heartbeater(frameHandler);
         _frameHandler = frameHandler;
         _running = true;
         _frameMax = 0;
@@ -349,10 +356,11 @@ public int getHeartbeat() {
      */
     public void setHeartbeat(int heartbeat) {
         try {
+            this.heartbeater.setHeartbeat(heartbeat);
+
             // Divide by four to make the maximum unwanted delay in
             // sending a timeout be less than a quarter of the
             // timeout setting.
-            _heartbeat = heartbeat;
             _frameHandler.setTimeout(heartbeat * 1000 / 4);
         } catch (SocketException se) {
             // should do more here?
@@ -416,7 +424,7 @@ private class MainLoop extends Thread {
             try {
                 while (_running) {
                     Frame frame = readFrame();
-                    maybeSendHeartbeat();
+
                     if (frame != null) {
                         _missedHeartbeats = 0;
                         if (frame.type == AMQP.FRAME_HEARTBEAT) {
@@ -459,25 +467,6 @@ private class MainLoop extends Thread {
         }
     }
 
-    private static final long NANOS_IN_SECOND = 1000 * 1000 * 1000;
-
-    /**
-     * Private API - Checks lastActivityTime and heartbeat, sending a
-     * heartbeat frame if conditions are right.
-     */
-    public void maybeSendHeartbeat() throws IOException {
-        if (_heartbeat == 0) {
-            // No heartbeating.
-            return;
-        }
-
-        long now = System.nanoTime();
-        if (now > (_lastActivityTime + (_heartbeat * NANOS_IN_SECOND))) {
-            _lastActivityTime = now;
-            writeFrame(new Frame(AMQP.FRAME_HEARTBEAT, 0));
-        }
-    }
-
     /**
      * Private API - Called when a frame-read operation times out. Checks to
      * see if too many heartbeats have been missed, and if so, throws
@@ -607,6 +596,10 @@ public ShutdownSignalException shutdown(Object reason,
             if (isOpen())
                 _shutdownCause = sse;
         }
+
+        // stop any heartbeating
+        this.heartbeater.shutdown();
+
         _channel0.processShutdownSignal(sse, !initiatedByApplication, notifyRpc);
         _channelManager.handleSignal(sse);
         return sse;

src/com/rabbitmq/client/impl/Heartbeater.java

@@ -0,0 +1,132 @@
+//   The contents of this file are subject to the Mozilla Public License
+//   Version 1.1 (the "License"); you may not use this file except in
+//   compliance with the License. You may obtain a copy of the License at
+//   http://www.mozilla.org/MPL/
+//
+//   Software distributed under the License is distributed on an "AS IS"
+//   basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+//   License for the specific language governing rights and limitations
+//   under the License.
+//
+//   The Original Code is RabbitMQ.
+//
+//   The Initial Developers of the Original Code are LShift Ltd,
+//   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+//   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+//   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+//   Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+//   Ltd. Portions created by Cohesive Financial Technologies LLC are
+//   Copyright (C) 2007-2010 Cohesive Financial Technologies
+//   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+//   (C) 2007-2010 Rabbit Technologies Ltd.
+//
+//   All Rights Reserved.
+//
+//   Contributor(s): ______________________________________.
+//
+
+package com.rabbitmq.client.impl;
+
+import com.rabbitmq.client.AMQP;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ScheduledFuture;
+import java.io.IOException;
+
+/**
+ * Manages heartbeats for a {@link AMQConnection}.
+ * <p/>
+ * Heartbeats are sent in a dedicated thread that is separate
+ * from the main loop thread used for the connection.
+ */
+final class Heartbeater {
+
+    private final Object monitor = new Object();
+
+    private final FrameHandler frameHandler;
+
+    private ScheduledExecutorService executor;
+
+    private ScheduledFuture<?> future;
+
+    Heartbeater(FrameHandler frameHandler) {
+        this.frameHandler = frameHandler;
+    }
+
+    /**
+     * Sets the heartbeat in seconds.
+     */
+    public void setHeartbeat(int heartbeatSeconds) {
+        ScheduledFuture<?> previousFuture = null;
+        synchronized (this.monitor) {
+            previousFuture = this.future;
+            this.future = null;
+        }
+
+        if (previousFuture != null) {
+            previousFuture.cancel(true);
+        }
+
+
+        if (heartbeatSeconds > 0) {
+            ScheduledExecutorService executor = createExecutorIfNecessary();
+            ScheduledFuture<?> newFuture = executor.scheduleAtFixedRate(
+                    new HeartbeatRunnable(), heartbeatSeconds,
+                    heartbeatSeconds, TimeUnit.SECONDS);
+
+            synchronized (this.monitor) {
+                this.future = newFuture;
+            }
+        }
+
+    }
+
+    private ScheduledExecutorService createExecutorIfNecessary() {
+        synchronized (this.monitor) {
+            if (this.executor == null) {
+                this.executor = Executors.newSingleThreadScheduledExecutor();
+            }
+            return this.executor;
+        }
+    }
+
+    /**
+     * Shutdown the heartbeat process, if any.
+     */
+    public void shutdown() {
+        ScheduledFuture<?> future;
+        ScheduledExecutorService executor;
+
+        synchronized (this.monitor) {
+            future = this.future;
+            executor = this.executor;
+            this.future = null;
+            this.executor = null;
+        }
+
+        if (future != null) {
+            future.cancel(true);
+        }
+
+        if (executor != null) {
+            executor.shutdown();
+        }
+    }
+
+    private class HeartbeatRunnable implements Runnable {
+
+        public void run() {
+            try {
+                frameHandler.writeFrame(new Frame(AMQP.FRAME_HEARTBEAT, 0));
+            } catch (IOException e) {
+                // ignore
+            }
+        }
+    }
+}