Merge pull request #1299 from sekogan/sekogan/fix_1298_case_3

Fix case 3 from #1298

pika/adapters/twisted_connection.py

@@ -134,11 +134,15 @@ def _on_channel_closed(self, _channel, reason):
         # errback all pending calls
         for d in self._calls:
             d.errback(self._closed)
+        # errback all pending deliveries
+        for d in self._deliveries.values():
+            d.errback(self._closed)
         # close all open queues
         for consumer in self._consumers.values():
             consumer.close(self._closed)
         # release references to stored objects
         self._calls = set()
+        self._deliveries = {}
         self._consumers = {}
         self.on_closed.callback(self._closed)
 

tests/acceptance/twisted_adapter_tests.py

@@ -610,6 +610,23 @@ def check_response(results):
         self.pika_channel.confirm_delivery.call_args[1]["callback"](None)
         return d
 
+    @deferred(timeout=5.0)
+    def test_delivery_confirmation_errback_on_close(self):
+        # Verify deliveries that haven't had their callback invoked errback when
+        # the channel closes.
+        d = self.channel.confirm_delivery()
+        # Simulate Confirm.SelectOk
+        self.pika_channel.confirm_delivery.call_args[1]["callback"](None)
+
+        def send_message_and_close_channel(_result):
+            d = self.channel.basic_publish("testexch", "testrk", "testbody")
+            self.channel._on_channel_closed(None, RuntimeError("testing"))
+            self.assertEqual(len(self.channel._deliveries), 0)
+            return d
+
+        d.addCallback(send_message_and_close_channel)
+        return self.assertFailure(d, RuntimeError)
+
 
 class TwistedProtocolConnectionTestCase(TestCase):