reinstated the lastActivityTime check

Author: Rob Harrop

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

@@ -144,12 +144,6 @@ public void ensureIsOpen()
         }
     }
 
-    /**
-     * Timestamp of last time we wrote a frame - used for deciding when to
-     * send a heartbeat
-     */
-    private volatile long _lastActivityTime = Long.MAX_VALUE;
-
     /**
      * Count of socket-timeouts that have happened without any incoming frames
      */
@@ -403,7 +397,7 @@ public Frame readFrame() throws IOException {
      */
     public void writeFrame(Frame f) throws IOException {
         _frameHandler.writeFrame(f);
-        _lastActivityTime = System.nanoTime();
+        this.heartbeater.signalActivity();
     }
 
     private static int negotiatedMaxValue(int clientValue, int serverValue) {

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

@@ -55,10 +55,16 @@ final class Heartbeater {
 
     private ScheduledFuture<?> future;
 
+    private volatile long lastActivityTime;
+
     Heartbeater(FrameHandler frameHandler) {
         this.frameHandler = frameHandler;
     }
 
+    public void signalActivity() {
+        this.lastActivityTime = System.nanoTime();
+    }
+
     /**
      * Sets the heartbeat in seconds.
      */
@@ -77,7 +83,7 @@ public void setHeartbeat(int heartbeatSeconds) {
         if (heartbeatSeconds > 0) {
             ScheduledExecutorService executor = createExecutorIfNecessary();
             ScheduledFuture<?> newFuture = executor.scheduleAtFixedRate(
-                    new HeartbeatRunnable(), heartbeatSeconds,
+                    new HeartbeatRunnable(heartbeatSeconds), heartbeatSeconds,
                     heartbeatSeconds, TimeUnit.SECONDS);
 
             synchronized (this.monitor) {
@@ -121,9 +127,21 @@ public void shutdown() {
 
     private class HeartbeatRunnable implements Runnable {
 
+        private static final long NANOS_IN_SECOND = 1000 * 1000 * 1000;
+
+        private final long heartbeatNanos;
+
+        private HeartbeatRunnable(int heartbeatSeconds) {
+            this.heartbeatNanos = NANOS_IN_SECOND * heartbeatSeconds;
+        }
+
         public void run() {
             try {
-                frameHandler.writeFrame(new Frame(AMQP.FRAME_HEARTBEAT, 0));
+                long now = System.nanoTime();
+
+                if(now > (lastActivityTime + this.heartbeatNanos)) {
+                    frameHandler.writeFrame(new Frame(AMQP.FRAME_HEARTBEAT, 0));
+                }
             } catch (IOException e) {
                 // ignore
             }