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

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.

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!

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! :)

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!

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.

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

> 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.