Merge lp:~elachuni/txamqp/heartbeat into lp:txamqp

[Anthony Lenton]

[Anthony Lenton (elachuni)]

Added handling for heartbeat frames.
The heartbeat interval is configurable by a AMQClient constructor parameter.
Added a brief test to check that it works.

It's only tested on RabbitMQ, but it should work on qpid too.

[Thomas Herve (therve)]

Thanks a lot for this code Anthony! It's going to be pretty useful. Here are some comments:

[1] There is a small conflict with current trunk.

[2] The connectionLostEvent doesn't seem to be used anywhere, it can probably be removed.

[3] The test is a bit bad: I understand you want to test it live, but a test running for 8 seconds is too long. Also, setTimeout is deprecated.

Thanks!

review: Needs Fixing

[lp:~elachuni/txamqp/heartbeat]

20. By Anthony Lenton on 2009-06-11

- Merged changes from trunk
- Removed unused connectionLostEvent
- Fixed test.

[Anthony Lenton (elachuni)]

Hi Thomas,
>
> [1] There is a small conflict with current trunk.
Merged and resolved.

>
> [2] The connectionLostEvent doesn't seem to be used anywhere, it can probably
> be removed.
Right, it was being used by client code here to easily access connectionLost events, but definitely shouldn't be part of the same patch. Removed.

>
> [3] The test is a bit bad: I understand you want to test it live, but a test
> running for 8 seconds is too long. Also, setTimeout is deprecated.
I've reduced it to 3 seconds. If that's still too long we'll need to test it some other way, as you can't ask for a heartbeat interval of less than one second. And it's not calling setTimeout now.

>
> Thanks!
Thank you! :)

[Thomas Herve (therve)]

OK, some last comments:

1) There is a conflict again :). Be careful here, because connectionLost is now defined in trunk, so you want to move your clean code.

2) The heartbeat test class is called TxTests, it should be named HeartbeatTests. The test file doesn't need the ASF license header too.

Thanks, +1!

review: Approve

[lp:~elachuni/txamqp/heartbeat]

21. By Anthony Lenton on 2009-06-30

Merged changes from trunk.

[Esteve Fernandez (esteve)]

Sorry for the slow response. I think adding heartbeat is a very useful feature, thanks for taking the time for implementing it. However, there's a couple of issues:

1 - The virtual host was changed to "/", instead of "localhost". We really need to implement "profiles" or whatever, so we can select some sane defaults for every broker. Until then, I'm not comfortable with introducing a change that's not intrinsic to the problem your branch solves.

2 - Although the code works and is clear, I'd rather use a LoopingCall (http://twistedmatrix.com/documents/8.2.0/api/twisted.internet.task.LoopingCall.html). It's already built into Twisted, so you don't need to implement things like lastSent, lastReceived, etc. I think it would be a bit easier to do something like this (it's just pseudo-python, can't guarantee that it works):

class AMQClient(...):
    # Max unreceived heartbeat frames. The AMQP standard says it's 3.
    MAX_UNSEEN_HEARTBEAT = 3

    def __init__(self, ...):
        ...
        if self.heartbeatInterval > 0:
            self.sendHB = LoopingCall(self.sendHeartbeat)
            self.checkHB = LoopingCall(self.checkHeartbeat)
            d = self.started.wait()
            d.addCallback(lambda _: self.sendHB.start(self.heartbeatInterval, now=False))
            d.addCallback(lambda _: self.checkHB.start(
                self.heartbeatInterval * self.MAX_UNSEEN_HEARTBEAT, now=False))

    def sendHeartbeat(self):
        self.sendFrame(Frame(0, Heartbeat()))

    def checkHeartbeat(self):
        self.checkHB.stop()
        self.transport.loseConnection()

    def processFrame(self, frame):
        ...
        if frame.payload.type != Frame.HEARTBEAT:
            self.sendHB.stop()
            self.sendHB.start(self.heartbeatInterval, now=False)
            ch.dispatch(frame, self.work)
        self.checkHB.stop()
        self.checkHb.start(self.heartbeatInterval * self.MAX_UNSEEN_HEARTBEAT, now=False)

What do you think guys? The changes are minimal and I think they are a bit easier to read.

[Esteve Fernandez (esteve)]

I just realized that self.checkHB doesn't need to be a LoopingCall, reactor.callLater is enough.

[lp:~elachuni/txamqp/heartbeat]

22. By Anthony Lenton on 2009-07-09

Putting back 'localhost' vhost.

[Anthony Lenton (elachuni)]

> 1 - The virtual host was changed to "/", instead of "localhost".

Yikes, that was unintentionally left in from running the tests :-/
I've put the 'localhost' vhost back in rev.22, sorry for that.

> 2 - Although the code works and is clear, I'd rather use a LoopingCall (http:/
> /twistedmatrix.com/documents/8.2.0/api/twisted.internet.task.LoopingCall.html)
> . It's already built into Twisted, so you don't need to implement things like
> lastSent, lastReceived, etc. I think it would be a bit easier to do something
> like this (it's just pseudo-python, can't guarantee that it works):
>
> (...code...)
>
> What do you think guys? The changes are minimal and I think they are a bit
> easier to read.

I gave your code a try and it works well, with a couple of changes. However I don't really see much benefit in using LoopingCall in this case:
 * checkHB could use a regular callLater call, as you've pointed out.
 * The rescheduling of sendHB should actually be done in sendFrame, not in processFrame, as it's sending
frames (not receiving them) that postpones the need to send a HB frame.
 * After changing that it's practically the same to be using a regular callLater instead of a LoopingCall for sendHB; if in sendFrame you reschedule sendHB whatever the payload type (instead of filtering out HBs) it effectively behaves like a LoopingCall.
 * We do get rid of lastSent and lastReceived, but instead we'd be using two callbacks instead of one (checkHeartbeat and sendHeartbeat vs. heartbeatHandler), and two 'callback handles' (checkHB and sendHB vs. pendingHeartbeat).
 * The free bonus you get with lastSent and lastReceived is a bit of accountability about how long ago you sent and received frames (including heartbeat frames), which makes it easier to test.

I haven't pushed these changes into this branch, but if you want I can push a working version using LoopingCalls on some other branch so we can compare.

[lp:~elachuni/txamqp/heartbeat]

23. By Anthony Lenton on 2009-07-15

Removed License header and fixed class name, per therve's review.

=== modified file 'src/txamqp/client.py'
--- src/txamqp/client.py	2009-05-04 11:12:36 +0000
+++ src/txamqp/client.py	2009-05-05 15:41:08 +0000
@@ -29,7 +29,8 @@
 
     def connection_tune(self, ch, msg):
         self.client.MAX_LENGTH = msg.frame_max
-        ch.connection_tune_ok(*msg.fields)
+        args = msg.channel_max, msg.frame_max, self.client.heartbeatInterval
+        ch.connection_tune_ok(*args)
         self.client.started.set()
 
     @defer.inlineCallbacks
 
=== modified file 'src/txamqp/connection.py'
--- src/txamqp/connection.py	2008-10-29 18:31:04 +0000
+++ src/txamqp/connection.py	2009-04-27 17:00:08 +0000
@@ -196,3 +196,15 @@
 
   def __str__(self):
     return "Body(%r)" % self.content
+
+class Heartbeat(Payload):
+  type = Frame.HEARTBEAT
+  def __str__(self):
+    return "Heartbeat()"
+
+  def encode(self, enc):
+    enc.encode_long(0)
+
+  def decode(spec, dec):
+    dec.decode_long()
+    return Heartbeat()
 
=== modified file 'src/txamqp/protocol.py'
--- src/txamqp/protocol.py	2009-05-04 11:12:36 +0000
+++ src/txamqp/protocol.py	2009-06-04 14:42:57 +0000
@@ -1,16 +1,17 @@
 # coding: utf-8
 from twisted.python import log
-from twisted.internet import defer, protocol
+from twisted.internet import defer, protocol, reactor
 from twisted.protocols import basic
 from txamqp import spec
 from txamqp.codec import Codec, EOF
-from txamqp.connection import Header, Frame, Method, Body
+from txamqp.connection import Header, Frame, Method, Body, Heartbeat
 from txamqp.message import Message
 from txamqp.content import Content
 from txamqp.queue import TimeoutDeferredQueue, Empty, Closed as QueueClosed
 from txamqp.client import TwistedEvent, TwistedDelegate, Closed
 from cStringIO import StringIO
 import struct
+from time import time
 
 class GarbageException(Exception):
     pass
@@ -200,7 +201,7 @@
 
     channelClass = AMQChannel
 
-    def __init__(self, delegate, vhost, *args, **kwargs):
+    def __init__(self, delegate, vhost, heartbeat=0, *args, **kwargs):
         FrameReceiver.__init__(self, *args, **kwargs)
         self.delegate = delegate
 
@@ -218,6 +219,9 @@
         self.work = defer.DeferredQueue()
 
         self.started = TwistedEvent()
+        self.connectionLostEvent = TwistedEvent()
+        self.lastSent = time()
+        self.lastReceived = time()
 
         self.queueLock = defer.DeferredLock()
 
@@ -225,6 +229,10 @@
 
         self.outgoing.get().addCallback(self.writer)
         self.work.get().addCallback(self.worker)
+        self.heartbeatInterval = heartbeat
+        self.pendingHeartbeat = None
+        if self.heartbeatInterval > 0:
+            self.started.wait().addCallback(self.heartbeatHandler)
 
     def check_0_8(self):
         return (self.spec.minor, self.spec.major) == (0, 8)
@@ -293,13 +301,25 @@
         self.sendInitString()
         self.setFrameMode()
 
+    def connectionLost(self, reason):
+        if self.pendingHeartbeat is not None and self.pendingHeartbeat.active():
+            self.pendingHeartbeat.cancel()
+            self.pendingHeartbeat = None
+        self.connectionLostEvent.set()
+
     def frameReceived(self, frame):
         self.processFrame(frame)
 
+    def sendFrame(self, frame):
+        self.lastSent = time()
+        FrameReceiver.sendFrame(self, frame)
+
     @defer.inlineCallbacks
     def processFrame(self, frame):
+        self.lastReceived = time()
         ch = yield self.channel(frame.channel)
-        ch.dispatch(frame, self.work)
+        if frame.payload.type != Frame.HEARTBEAT:
+            ch.dispatch(frame, self.work)
 
     @defer.inlineCallbacks
     def authenticate(self, username, password, mechanism='AMQPLAIN', locale='en_US'):
@@ -320,3 +340,15 @@
 
         channel0 = yield self.channel(0)
         yield channel0.connection_open(self.vhost)
+
+    def heartbeatHandler (self, dummy=None):
+        now = time()
+        if self.lastSent + self.heartbeatInterval < now:
+            self.sendFrame(Frame(0, Heartbeat()))
+        if self.lastReceived + self.heartbeatInterval * 3 < now:
+            self.transport.loseConnection()
+        tple = None
+        if self.transport.connected:
+            tple = reactor.callLater(self.heartbeatInterval, self.heartbeatHandler)
+        self.pendingHeartbeat = tple
+            
 
=== added file 'src/txamqp/test/test_heartbeat.py'
--- src/txamqp/test/test_heartbeat.py	1970-01-01 00:00:00 +0000
+++ src/txamqp/test/test_heartbeat.py	2009-06-04 16:26:58 +0000
@@ -0,0 +1,57 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+# 
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+from time import time
+#from txamqp.client import Closed
+#from txamqp.queue import Empty
+#from txamqp.content import Content
+from txamqp.testlib import TestBase
+
+#from twisted.internet.defer import, returnValue,
+from twisted.internet.defer import Deferred, inlineCallbacks
+
+class TxTests(TestBase):
+    """
+    Tests handling of heartbeat frames
+    """
+
+    
+    @inlineCallbacks
+    def setUp(self):
+        """ Set up a heartbeat frame per second """
+        self.client = yield self.connect(heartbeat=1)
+
+        self.channel = yield self.client.channel(1)
+        yield self.channel.channel_open()
+
+
+    @inlineCallbacks
+    def test_heartbeat(self):
+        """
+        Test that heartbeat frames are sent and received
+        """
+        d = Deferred()
+        d.setTimeout(8)
+        d.addBoth(lambda x:True)
+        yield d
+        t = time()
+        self.assertTrue(self.client.lastSent > t - 3,
+                        "A heartbeat frame was recently sent")
+        self.assertTrue(self.client.lastReceived > t - 3,
+                        "A heartbeat frame was recently received")
 
=== modified file 'src/txamqp/testlib.py'
--- src/txamqp/testlib.py	2009-02-11 22:28:45 +0000
+++ src/txamqp/testlib.py	2009-06-04 16:26:58 +0000
@@ -39,22 +39,26 @@
         self.user = 'guest'
         self.password = 'guest'
         self.vhost = 'localhost'
+        self.heartbeat = 0
         self.queues = []
         self.exchanges = []
         self.connectors = []
 
     @inlineCallbacks
-    def connect(self, host=None, port=None, spec=None, user=None, password=None, vhost=None):
+    def connect(self, host=None, port=None, spec=None, user=None,
+                           password=None, vhost=None, heartbeat=None):
         host = host or self.host
         port = port or self.port
         spec = spec or self.spec
         user = user or self.user
         password = password or self.password
         vhost = vhost or self.vhost
+        heartbeat = heartbeat or self.heartbeat
 
         delegate = TwistedDelegate()
         onConn = Deferred()
-        f = protocol._InstanceFactory(reactor, AMQClient(delegate, vhost, txamqp.spec.load(spec)), onConn)
+        p = AMQClient(delegate, vhost, heartbeat=heartbeat, spec=txamqp.spec.load(spec))
+        f = protocol._InstanceFactory(reactor, p, onConn)
         c = reactor.connectTCP(host, port, f)
         self.connectors.append(c)
         client = yield onConn