Merge lp:~txamqpteam/txamqp/support-basic-return into lp:txamqp

[Terry Jones]

[Terry Jones (terrycojones)]

This is work done with Esteve to add support for the AMQP basic return message. It adds a mandatory argument to outgoing messages, resulting in a basic return if the message is not routable.

There is an outstanding issue with what, if anything, to do with Thrift oneway methods. They have no client side deferred whose errback can be called if they are unroutable.

[Esteve Fernandez (esteve)]

review: Approve

[Esteve Fernandez (esteve)]

Dan, given that you're the only one I know using Thrift and AMQP together (apart from Terry and me), could you review this branch? Thanks!

[lp:~txamqpteam/txamqp/support-basic-return]

22. By Terry Jones on 2009-07-07

Cleaned up handling of KeyError n txamqp/contrib/thrift/client.py (which had a silly logic error - of mine). Added a __repr__ to the Content class in txamqp/content.py. And I now have a reproducible example of 'headers' not being in msg.content when trying to get the thriftClientName in txamqp/contrib/thrift/client.py though haven't tried to dig into why it happens.

[Dan Di Spaltro (dan-dispaltro)]

+1

Ran the code works well, very handy to boot, now our services are more reliable. The log stuff could probably be cleaned up a bit and lines shortened but that is just cosmetic.

review: Approve (code)

[Esteve Fernandez (esteve)]

> +1
>
> Ran the code works well, very handy to boot, now our services are more
> reliable. The log stuff could probably be cleaned up a bit and lines
> shortened but that is just cosmetic.

Thanks! Just merged it.

[Terry Jones (terrycojones)]

Hi Dan

>>>>> "Dan" == Dan Di Spaltro <email address hidden> writes:
Dan> Ran the code works well, very handy to boot, now our services are more
Dan> reliable. The log stuff could probably be cleaned up a bit and lines
Dan> shortened but that is just cosmetic.

The logging in src/txamqp/contrib/thrift/client.py (if that's what you were
referring to) is there due to very the occasional absence of 'headers' in
msg.content. I've seen it happen a couple of times, and had it in
reproducible form (though in the complex setup that is FluidDB, so not easy
to reduce and post for others). But I've not had time to look at it. So I
left the logging there in the hope that someone else might see this pop up
and feel like digging into it :-) I'm sure I will at some point.

It may be benign / harmless. Esteve, I think, came up with a scenario in
which this might happen, but I forget the details.

Terry

=== modified file 'src/examples/client.py'
--- src/examples/client.py	2009-02-11 22:28:45 +0000
+++ src/examples/client.py	2009-06-02 20:33:36 +0000
@@ -6,7 +6,7 @@
 sys.path.insert(0, os.path.join(os.path.abspath(os.path.split(sys.argv[0])[0]), 'gen-py'))
 import tutorial.Calculator
 from tutorial.ttypes import *
-from thrift.transport import TTwisted
+from thrift.transport import TTwisted, TTransport
 from thrift.protocol import TBinaryProtocol
 
 from twisted.internet import reactor, defer
@@ -34,9 +34,15 @@
     print results
 
 def gotCalculateErrors(error):
-    print "Got an error"
+    error.trap(InvalidOperation)
+    print "Got a calculator error"
     print error.value.why
 
+def gotTransportError(error):
+    error.trap(TTransport.TTransportException)
+    print "Got an AMQP unroutable message error:"
+    print error.value.message
+
 @defer.inlineCallbacks
 def prepareClient(client, username, password):
     yield client.authenticate(username, password)
@@ -48,6 +54,11 @@
     yield channel.exchange_declare(exchange=responsesExchange, type="direct")
 
     pfactory = TBinaryProtocol.TBinaryProtocolFactory()
+
+    # To trigger an unroutable message error (caught in the above
+    # gotTransportError errback), change the routing key (i.e.,
+    # calculatorKey) in the following to be something invalid, like
+    # calculatorKey + 'xxx'.
     thriftClient = yield client.createThriftClient(responsesExchange,
         servicesExchange, calculatorKey, tutorial.Calculator.Client,
         iprot_factory=pfactory, oprot_factory=pfactory)
@@ -55,35 +66,35 @@
     defer.returnValue(thriftClient)
 
 def gotClient(client):
-    d1 = client.ping().addCallback(gotPing)
+    d1 = client.ping().addCallback(gotPing).addErrback(gotTransportError)
 
-    d2 = client.add(1, 2).addCallback(gotAddResults)
+    d2 = client.add(1, 2).addCallback(gotAddResults).addErrback(gotTransportError)
 
     w = Work(num1=2, num2=3, op=Operation.ADD)
 
     d3 = client.calculate(1, w).addCallbacks(gotCalculateResults,
-        gotCalculateErrors)
+        gotCalculateErrors).addErrback(gotTransportError)
 
     w = Work(num1=2, num2=3, op=Operation.SUBTRACT)
 
     d4 = client.calculate(2, w).addCallbacks(gotCalculateResults,
-        gotCalculateErrors)
+        gotCalculateErrors).addErrback(gotTransportError)
 
     w = Work(num1=2, num2=3, op=Operation.MULTIPLY)
 
     d5 = client.calculate(3, w).addCallbacks(gotCalculateResults,
-        gotCalculateErrors)
+        gotCalculateErrors).addErrback(gotTransportError)
 
     w = Work(num1=2, num2=3, op=Operation.DIVIDE)
 
     d6 = client.calculate(4, w).addCallbacks(gotCalculateResults,
-        gotCalculateErrors)
+        gotCalculateErrors).addErrback(gotTransportError)
 
     # This will fire an errback
     w = Work(num1=2, num2=0, op=Operation.DIVIDE)
 
     d7 = client.calculate(5, w).addCallbacks(gotCalculateResults,
-        gotCalculateErrors)
+        gotCalculateErrors).addErrback(gotTransportError)
 
     d8 = client.zip()
 
 
=== modified file 'src/txamqp/client.py'
--- src/txamqp/client.py	2009-05-28 09:46:03 +0000
+++ src/txamqp/client.py	2009-06-02 20:33:36 +0000
@@ -36,6 +36,9 @@
     def basic_deliver(self, ch, msg):
         (yield self.client.queue(msg.consumer_tag)).put(msg)
 
+    def basic_return_(self, ch, msg):
+        self.client.basic_return_queue.put(msg)
+
     def channel_close(self, ch, msg):
         ch.close(msg)
 
 
=== added file 'src/txamqp/contrib/thrift/client.py'
--- src/txamqp/contrib/thrift/client.py	1970-01-01 00:00:00 +0000
+++ src/txamqp/contrib/thrift/client.py	2009-06-09 13:29:13 +0000
@@ -0,0 +1,21 @@
+from twisted.internet import defer
+
+from txamqp.client import TwistedDelegate
+
+
+class ThriftTwistedDelegate(TwistedDelegate):
+
+    @defer.inlineCallbacks
+    def basic_return_(self, ch, msg):
+        try:
+            thriftClientName = msg.content['headers']['thriftClientName']
+        except KeyError:
+            from twisted.python import log
+            if 'headers' in msg.content:
+                log.msg("'headers' not in msg.content: %r" % msg.content)
+            else:
+                log.msg("'thriftClientName' not in msg.content headers: %r" %
+                        msg.content['headers'])
+        else:
+            (yield self.client.thriftBasicReturnQueue(thriftClientName))\
+                   .put(msg)
 
=== modified file 'src/txamqp/contrib/thrift/protocol.py'
--- src/txamqp/contrib/thrift/protocol.py	2009-05-29 00:20:49 +0000
+++ src/txamqp/contrib/thrift/protocol.py	2009-06-09 13:35:48 +0000
@@ -2,6 +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 twisted.internet import defer
 from twisted.python import log
@@ -19,6 +20,22 @@
         else:
             self.replyToField = "reply-to"
 
+        self.thriftBasicReturnQueueLock = defer.DeferredLock()
+        self.thriftBasicReturnQueues = {}
+
+    @defer.inlineCallbacks
+    def thriftBasicReturnQueue(self, key):
+        yield self.thriftBasicReturnQueueLock.acquire()
+        try:
+            try:
+                q = self.thriftBasicReturnQueues[key]
+            except KeyError:
+                q = TimeoutDeferredQueue()
+                self.thriftBasicReturnQueues[key] = q
+        finally:
+            self.thriftBasicReturnQueueLock.release()
+        defer.returnValue(q)
+
     @defer.inlineCallbacks
     def createThriftClient(self, responsesExchange, serviceExchange,
         routingKey, clientClass, channel=1, responseQueue=None, iprot_factory=None,
@@ -39,8 +56,11 @@
 
         log.msg("Consuming messages on queue: %s" % responseQueue)
 
-        amqpTransport = TwistedAMQPTransport(channel, serviceExchange,
-            routingKey, replyTo=responseQueue, replyToField=self.replyToField)
+        thriftClientName = clientClass.__name__ + routingKey
+        
+        amqpTransport = TwistedAMQPTransport(
+            channel, serviceExchange, routingKey, clientName=thriftClientName,
+            replyTo=responseQueue, replyToField=self.replyToField)
 
         if iprot_factory is None:
             iprot_factory = self.factory.iprot_factory
@@ -54,10 +74,16 @@
         queue.get().addCallback(self.parseClientMessage, channel, queue,
             thriftClient, iprot_factory=iprot_factory)
 
+        basicReturnQueue = yield self.thriftBasicReturnQueue(thriftClientName)
+
+        basicReturnQueue.get().addCallback(
+            self.parseClientUnrouteableMessage, channel, basicReturnQueue,
+            thriftClient, iprot_factory=iprot_factory)
+
         defer.returnValue(thriftClient)
 
     def parseClientMessage(self, msg, channel, queue, thriftClient,
-        iprot_factory=None):
+                           iprot_factory=None):
         deliveryTag = msg.delivery_tag
         tr = TTransport.TMemoryBuffer(msg.content.body)
         if iprot_factory is None:
@@ -66,6 +92,12 @@
             iprot = iprot_factory.getProtocol(tr)
         (fname, mtype, rseqid) = iprot.readMessageBegin()
 
+        if rseqid in thriftClient._reqs:
+            # log.msg('Got reply: fname = %r, rseqid = %s, mtype = %r, routing key = %r, client = %r, msg.content.body = %r' % (fname, rseqid, mtype, msg.routing_key, thriftClient, msg.content.body))
+            pass
+        else:
+            log.msg('Missing rseqid! fname = %r, rseqid = %s, mtype = %r, routing key = %r, client = %r, msg.content.body = %r' % (fname, rseqid, mtype, msg.routing_key, thriftClient, msg.content.body))
+            
         method = getattr(thriftClient, 'recv_' + fname)
         method(iprot, mtype, rseqid)
 
@@ -73,6 +105,34 @@
         queue.get().addCallback(self.parseClientMessage, channel, queue,
             thriftClient, iprot_factory=iprot_factory)
 
+    def parseClientUnrouteableMessage(self, msg, channel, queue, thriftClient,
+                                      iprot_factory=None):
+        tr = TTransport.TMemoryBuffer(msg.content.body)
+        if iprot_factory is None:
+            iprot = self.factory.iprot_factory.getProtocol(tr)
+        else:
+            iprot = iprot_factory.getProtocol(tr)
+        (fname, mtype, rseqid) = iprot.readMessageBegin()
+
+        # log.msg('Got unroutable. fname = %r, rseqid = %s, mtype = %r, routing key = %r, client = %r, msg.content.body = %r' % (fname, rseqid, mtype, msg.routing_key, thriftClient, msg.content.body))
+
+        try:
+            d = thriftClient._reqs.pop(rseqid)
+        except KeyError:
+            # KeyError will occur if the remote Thrift method is oneway,
+            # since there is no outstanding local request deferred for
+            # oneway calls.
+            pass
+        else:
+            d.errback(TTransport.TTransportException(
+                type=TTransport.TTransportException.NOT_OPEN,
+                message='Unrouteable message, routing key = %r calling function %r'
+                % (msg.routing_key, fname)))
+
+        queue.get().addCallback(
+            self.parseClientUnrouteableMessage, channel, queue,
+            thriftClient, iprot_factory=iprot_factory)
+
     def parseServerMessage(self, msg, channel, exchange, queue, processor,
         iprot_factory=None, oprot_factory=None):
         deliveryTag = msg.delivery_tag
 
=== modified file 'src/txamqp/contrib/thrift/transport.py'
--- src/txamqp/contrib/thrift/transport.py	2009-02-11 22:28:45 +0000
+++ src/txamqp/contrib/thrift/transport.py	2009-06-05 13:34:08 +0000
@@ -2,18 +2,26 @@
 from thrift.transport import TTwisted
 
 class TwistedAMQPTransport(TTwisted.TMessageSenderTransport):
-    def __init__(self, channel, exchange, routingKey, replyTo=None, replyToField=None):
+    def __init__(self, channel, exchange, routingKey, clientName=None,
+                 replyTo=None, replyToField=None):
         TTwisted.TMessageSenderTransport.__init__(self)
         self.channel = channel
         self.exchange = exchange
         self.routingKey = routingKey
+        # clientName is the name of the client class we are trying to get
+        # the message through to. We need to send it seeing as the message
+        # may be unroutable and we need a basic return that will tell us
+        # who were trying to reach.
+        self.clientName = clientName
         self.replyTo = replyTo
         self.replyToField = replyToField
 
     def sendMessage(self, message):
         content = Content(body=message)
+        if self.clientName:
+            content['headers'] = { 'thriftClientName' : self.clientName }
         if self.replyTo:
             content[self.replyToField] = self.replyTo
 
         self.channel.basic_publish(exchange=self.exchange,
-            routing_key=self.routingKey, content=content)
+            routing_key=self.routingKey, content=content, mandatory=True)
 
=== modified file 'src/txamqp/protocol.py'
--- src/txamqp/protocol.py	2009-06-01 16:35:03 +0000
+++ src/txamqp/protocol.py	2009-06-09 13:35:48 +0000
@@ -220,6 +220,7 @@
         self.started = TwistedEvent()
 
         self.queueLock = defer.DeferredLock()
+        self.basic_return_queue = TimeoutDeferredQueue()
 
         self.queues = {}
 
@@ -254,7 +255,7 @@
         finally:
             self.queueLock.release()
         defer.returnValue(q)
- 
+
     def close(self, reason):
         for ch in self.channels.values():
             ch.close(reason)