Merge lp:~jonathan-stoppani/txamqp/catch-closed-queues into lp:txamqp

[Jonathan Stoppani]

[Jonathan Stoppani (jonathan-stoppani)]

Solves the incorrect shutdown of the sample client and allows correct channel closing.

The problem was solved by adding a errback to the TimeoutDeferreQueue instance used by ThriftAMQClient which only traps a "Closed()" failure.

As I don't know well the internals of txAMQP could be that some more cleanup code should be added to the errback, but the client is shutting down properly, thus I assumed all worked correctly.

Greets

[Esteve Fernandez (esteve)]

Sorry for the delay, thanks Jonathan for your patch. It looks great, I just made some small modifications (lp:~esteve/txamqp/catch-closed-queues) to it:

- separated handling of client's and server's errbacks
- added two empty functions (handleClientQueueError and handleClosedServerQueue), which can be overridden by subclasses to provide more fine-grained error handling

Let me know what you think of them. Thanks!

[Jonathan Stoppani (jonathan-stoppani)]

Hi Esteve,
I just took a look at your changes and I have no issues ;-)

Thanks for reviewing and no problems for the delay...

=== modified file 'src/txamqp/contrib/thrift/protocol.py'
--- src/txamqp/contrib/thrift/protocol.py	2009-06-09 13:35:48 +0000
+++ src/txamqp/contrib/thrift/protocol.py	2009-08-06 08:30:08 +0000
@@ -2,7 +2,7 @@
 from txamqp.protocol import AMQClient
 from txamqp.contrib.thrift.transport import TwistedAMQPTransport
 from txamqp.content import Content
-from txamqp.queue import TimeoutDeferredQueue
+from txamqp.queue import TimeoutDeferredQueue, Closed
 
 from twisted.internet import defer
 from twisted.python import log
@@ -71,15 +71,18 @@
         thriftClient = clientClass(amqpTransport, oprot_factory)
 
         queue = yield self.queue(reply.consumer_tag)
-        queue.get().addCallback(self.parseClientMessage, channel, queue,
-            thriftClient, iprot_factory=iprot_factory)
+        d = queue.get()
+        d.addCallback(self.parseClientMessage, channel, queue, thriftClient,
+            iprot_factory=iprot_factory)
+        d.addErrback(self.handleClosedQueue)
 
         basicReturnQueue = yield self.thriftBasicReturnQueue(thriftClientName)
-
-        basicReturnQueue.get().addCallback(
-            self.parseClientUnrouteableMessage, channel, basicReturnQueue,
-            thriftClient, iprot_factory=iprot_factory)
-
+        
+        d = basicReturnQueue.get()
+        d.addCallback(self.parseClientUnrouteableMessage, channel,
+            basicReturnQueue, thriftClient, iprot_factory=iprot_factory)
+        d.addErrback(self.handleClosedQueue)
+        
         defer.returnValue(thriftClient)
 
     def parseClientMessage(self, msg, channel, queue, thriftClient,
@@ -102,8 +105,11 @@
         method(iprot, mtype, rseqid)
 
         channel.basic_ack(deliveryTag, True)
-        queue.get().addCallback(self.parseClientMessage, channel, queue,
-            thriftClient, iprot_factory=iprot_factory)
+        
+        d = queue.get()
+        d.addCallback(self.parseClientMessage, channel, queue, thriftClient,
+            iprot_factory=iprot_factory)
+        d.addErrback(self.handleClosedQueue)
 
     def parseClientUnrouteableMessage(self, msg, channel, queue, thriftClient,
                                       iprot_factory=None):
@@ -129,9 +135,10 @@
                 message='Unrouteable message, routing key = %r calling function %r'
                 % (msg.routing_key, fname)))
 
-        queue.get().addCallback(
-            self.parseClientUnrouteableMessage, channel, queue,
+        d = queue.get()
+        d.addCallback(self.parseClientUnrouteableMessage, channel, queue,
             thriftClient, iprot_factory=iprot_factory)
+        d.addErrback(self.handleClosedQueue)
 
     def parseServerMessage(self, msg, channel, exchange, queue, processor,
         iprot_factory=None, oprot_factory=None):
@@ -157,9 +164,14 @@
         d = processor.process(iprot, oprot)
         channel.basic_ack(deliveryTag, True)
 
-        queue.get().addCallback(self.parseServerMessage, channel,
-            exchange, queue, processor, iprot_factory, oprot_factory)
+        d = queue.get()
+        d.addCallback(self.parseServerMessage, channel, exchange, queue,
+            processor, iprot_factory, oprot_factory)
+        d.addErrback(self.handleClosedQueue)
 
+    def handleClosedQueue(self, failure):
+        # The queue is closed. Catch the exception and cleanup as needed.
+        failure.trap(Closed)
 
 class IThriftAMQClientFactory(Interface):