Merge lp:~me-smira/txamqp/patches into lp:txamqp

[Andrey Smirnov]

[Andrey Smirnov (me-smira)]

Some small patches I had to apply to txAMQP to keep it working.

I'm using it heavily at qik.com, it receives good load (5M+ messages per day pushed through queues per day).

One of the changes fixes bug with channel-flow messages not being implemented. Other are related to scenario when channel is closed by server due to some error (like "queue doesn't exist") and client should reopen the channel to continue operations.

Thanks for great library!

[Esteve Fernandez (esteve)]

review: Approve

=== modified file 'src/txamqp/client.py'
--- src/txamqp/client.py	2009-08-08 17:54:09 +0000
+++ src/txamqp/client.py	2010-02-17 06:06:16 +0000
@@ -40,7 +40,11 @@
     def basic_return_(self, ch, msg):
         self.client.basic_return_queue.put(msg)
 
+    def channel_flow(self, ch, msg):
+        ch.channel_flow_ok(active=msg.active)
+
     def channel_close(self, ch, msg):
+        ch.channel_close_ok()
         ch.close(msg)
 
     def connection_close(self, ch, msg):
 
=== modified file 'src/txamqp/protocol.py'
--- src/txamqp/protocol.py	2009-11-13 18:50:13 +0000
+++ src/txamqp/protocol.py	2010-02-17 06:06:16 +0000
@@ -284,6 +284,24 @@
             self.queueLock.release()
         defer.returnValue(q)
 
+    @defer.inlineCallbacks
+    def closeChannel(self, channel):
+        yield self.channelLock.acquire()
+        try:
+            channel.close(None)
+            del self.channels[channel.id]
+        finally:
+            self.channelLock.release()
+ 
+    @defer.inlineCallbacks
+    def closeQueue(self, key, queue):
+         yield self.queueLock.acquire()
+         try:
+            queue.close()
+            del self.queues[key]
+         finally:
+            self.queueLock.release()
+ 
     def close(self, reason):
         for ch in self.channels.values():
             ch.close(reason)