Merge lp:~lifeless/python-oops-amqp/0.0.2 into lp:python-oops-amqp

[Robert Collins]

0.0.2
-----

* Fix documentation warts from initial release, updated to 0.0.2 and prepare
  for making the receiver deal with interrupted services.
  (Robert Collins, #875976, #875984)

* Fix Receiver.run_forever to actually run forever. (Robert Collins)

* Change API for constructing Receiver to take a connection factory rather than
  a channel. This will permit handling transient faults internally rather than
  forcing a restart. (Robert Collins)

* Implement resiliency for the Receiver: automatically reconnect if a socket
  error is received from rabbit, for up to two minutes of downtime.
  (Robert Collins)

[Robert Collins (lifeless)]

I forgot to add improvements from the work on LP, moved to WIP

[lp:~lifeless/python-oops-amqp/0.0.2]

10. By Robert Collins on 2011-10-18

Permit supplying the oops id when publishing.

[William Grant (wgrant)]

183 - def __init__(self, config, channel, queue_name):
184 + def __init__(self, config, connection_factory, queue_name):
185 """Create a Receiver.
186
187 :param config: An oops.Config to republish the OOPS reports.
188 - :param channel: An amqplib Channel to listen for reports on.
189 + :param connection: An amqplib connection factory, used to make the
190 + initial connection and to reconnect if that connection is
191 + interrupted.

docstring says the param is "connection", but it's actually "connection_factory".

227 + while not self.stopping and time.time() < self.connection_start + 120:

This is going to stop retrying immediately if the connection dies after more than two minutes.

321 + connection = self.connection_factory()
322 + self.addCleanup(connection.close)
323 + channel = connection.channel()
324 + self.addCleanup(channel.close)
325 + queue = self.useFixture(QueueFixture(channel, self.getUniqueString))

This is now roughly quadruplicated. Put it in a helper or setUp?

318 + def test_publish_inherit_id(self):

Are test_publish and test_publish_inherit_id around the wrong way?

304 + # Publication returns the oops ID allocated.

s/allocated/provided/?

371 - def test_receive_one(self):

Where'd that go?

398 + connection = self.connection_factory()
399 + self.addCleanup(connection.close)
400 + channel = connection.channel()
401 + self.addCleanup(channel.close)
402 + queue = self.useFixture(QueueFixture(channel, self.getUniqueString))

More duplication.

410 + receiver.sentinel = sentinel
411 + receiver.run_forever()
412 + self.assertEqual([expected_report], reports)

May be worth a comment around run_forever to say that it will return quickly since you've already sent the sentinel.

422 + connection = self.connection_factory()
423 + self.addCleanup(connection.close)
424 + channel = connection.channel()
425 + self.addCleanup(channel.close)
426 + queue = self.useFixture(QueueFixture(channel, self.getUniqueString))

Guess what...

499 + # publisher tests is what le aks through when rabbit is shutdown).

You've leaked some whitespace there.

507 + connection = self.connection_factory()
508 + self.addCleanup(connection.close)
509 + channel = connection.channel()
510 + self.addCleanup(channel.close)
511 + queue = self.useFixture(QueueFixture(channel, self.getUniqueString))

Ahem.

review: Needs Fixing (code)

[Robert Collins (lifeless)]

receive_one got renamed stop_via_stopping

I see an fixtures_amqp coming on sometime soon.

[lp:~lifeless/python-oops-amqp/0.0.2]

11. By Robert Collins on 2011-10-18

Review feedback (fix retry, docstring).

12. By Robert Collins on 2011-10-18

Refactor test support.

[William Grant (wgrant)]

review: Approve (code)

=== modified file 'NEWS'
--- NEWS	2011-10-10 21:49:50 +0000
+++ NEWS	2011-10-18 03:36:23 +0000
@@ -6,6 +6,33 @@
 NEXT
 ----
 
+0.0.2
+-----
+
+* Fix documentation warts from initial release, updated to 0.0.2 and prepare
+  for making the receiver deal with interrupted services.
+  (Robert Collins, #875976, #875984)
+
+* Fix Receiver.run_forever to actually run forever. (Robert Collins)
+
+* Change API for constructing Receiver to take a connection factory rather than
+  a channel. This will permit handling transient faults internally rather than
+  forcing a restart. (Robert Collins)
+
+* Implement resiliency for the Receiver: automatically reconnect if a socket
+  error is received from rabbit, for up to two minutes of downtime.
+  (Robert Collins)
+
+* The publisher can honour any existing oops id if desired (set
+  inherit_id=True). This is useful if using a chain of publishers where
+  AMQP is the last stage rather than the primary mechanism.
+  (Robert Collins)
+
+* Receiver.sentinel may be set and used to cause the run_forever loop to exit
+  when a crafted message is received. This is useful for testing, and defaults
+  to None (so cannot be used to shutdown a normally running service).
+  (Robert Collins)
+
 0.0.1
 -----
 

=== modified file 'README'
--- README	2011-10-10 21:49:50 +0000
+++ README	2011-10-18 03:36:23 +0000
@@ -98,17 +98,16 @@
 AMQP. To use it you need to configure a local config to publish the received
 reports. A full config is used because that includes support for filtering
 (which can be useful if you need to throttle volume, for instance).
-Additionally you need an amqp channel object and a queue name to receive from.
+Additionally you need an amqp connection factory (to handle the amqp server
+being restarted) and a queue name to receive from.
 
-This example uses the OOPSDateDirRepo publisher, telling it to accept whatever
+This example uses the DateDirRepo publisher, telling it to accept whatever
 id was assigned by the process publishing to AMQP::
 
-  >>> publisher = oops_datedir_repo.OOPSDateDirRepo('.', inherit_id=True)
+  >>> publisher = oops_datedir_repo.DateDirRepo('.', inherit_id=True)
   >>> config = oops.Config()
   >>> config.publishers.append(publisher.publish)
-  >>> connection = amqp.Connection(host="localhost:5672",
-  ...     userid="guest", password="guest", virtual_host="/", insist=False)
-  >>> receiver = oops_amqp.Receiver(config, connection, "my queue")
+  >>> receiver = oops_amqp.Receiver(config, factory, "my queue")
   >>> receiver.run_forever()
 
 For more information see pydoc oops_amqp.

=== modified file 'oops_amqp/__init__.py'
--- oops_amqp/__init__.py	2011-10-10 21:49:50 +0000
+++ oops_amqp/__init__.py	2011-10-18 03:36:23 +0000
@@ -73,14 +73,16 @@
 AMQP. To use it you need to configure a local config to publish the received
 reports. A full config is used because that includes support for filtering
 (which can be useful if you need to throttle volume, for instance).
+Additionally you need an amqp connection factory (to handle the amqp server
+being restarted) and a queue name to receive from.
 
-This example uses the OOPSDateDirRepo publisher, telling it to accept whatever
+This example uses the DateDirRepo publisher, telling it to accept whatever
 id was assigned by the process publishing to AMQP::
 
-  >>> publisher = oops_datedir_repo.OOPSDateDirRepo('.', inherit_id=True)
+  >>> publisher = oops_datedir_repo.DateDirRepo('.', inherit_id=True)
   >>> config = oops.Config()
   >>> config.publishers.append(publisher.publish)
-  >>> receiver = oops_amqp.Receiver(config)
+  >>> receiver = oops_amqp.Receiver(config, factory, "my queue")
   >>> receiver.run_forever()
 """
 
@@ -95,7 +97,7 @@
 # established at this point, and setup.py will use a version of next-$(revno).
 # If the releaselevel is 'final', then the tarball will be major.minor.micro.
 # Otherwise it is major.minor.micro~$(revno).
-__version__ = (0, 0, 1, 'beta', 0)
+__version__ = (0, 0, 2, 'beta', 0)
 
 __all__ = [
     'Publisher',

=== modified file 'oops_amqp/publisher.py'
--- oops_amqp/publisher.py	2011-10-10 21:51:53 +0000
+++ oops_amqp/publisher.py	2011-10-18 03:36:23 +0000
@@ -36,7 +36,8 @@
     supplied exchange + routing key.
     """
 
-    def __init__(self, connection_factory, exchange_name, routing_key):
+    def __init__(self, connection_factory, exchange_name, routing_key,
+            inherit_id=False):
         """Create a publisher.
 
         :param connection_factory: A callable which creates an amqplib
@@ -46,11 +47,15 @@
             across threads.
         :param exchange_name: The name of the exchange to publish to.
         :param routing_key: The routing key for messages.
+        :param inherit_id: If True any 'True' 'id' in an OOPS report is
+            preserved. Handy if an id that has already been shown to a user is
+            being published (but uniqueness cannot be guaranteed).
         """
         self.connection_factory = connection_factory
         self.exchange_name = exchange_name
         self.routing_key = routing_key
         self.channels = local()
+        self.inherit_id = inherit_id
 
     def get_channel(self):
         if getattr(self.channels, 'channel', None) is None:
@@ -64,13 +69,14 @@
     def __call__(self, report):
         # Don't mess with the passed in report.
         report = dict(report)
-        # Discard any existing id.
-        report.pop('id', None)
-        # Hash it, to make an ID
-        oops_id = "OOPS-%s" % md5(dumps(report)).hexdigest()
-        # Store the id in what we send on the wire, so that the recipient has
-        # it.
-        report['id'] = oops_id
+        if not self.inherit_id or not report.get('id'):
+            # Discard any existing id.
+            original_id = report.pop('id', None)
+            # Hash it, to make an ID
+            oops_id = "OOPS-%s" % md5(dumps(report)).hexdigest()
+            # Store the id in what we send on the wire, so that the recipient
+            # has it.
+            report['id'] = oops_id
         message = amqp.Message(dumps(report))
         # We don't want to drop OOPS on the floor if rabbit is restarted.
         message.properties["delivery_mode"] = 2
@@ -83,4 +89,4 @@
         except socket.error:
             self.channels.channel = None
             return None
-        return oops_id
+        return report['id']

=== modified file 'oops_amqp/receiver.py'
--- oops_amqp/receiver.py	2011-10-10 21:49:50 +0000
+++ oops_amqp/receiver.py	2011-10-18 03:36:23 +0000
@@ -18,8 +18,13 @@
 
 __metaclass__ = type
 
+import socket
+import time
+
 import bson
 
+from utils import close_ignoring_EPIPE
+
 __all__ = [
     'Receiver',
     ]
@@ -28,23 +33,32 @@
     """Republish OOPS reports from AMQP to a local oops.Config.
     
     :ivar stopping: When True will cause Receiver to break out of run_forever.
+        Calls to run_forever reset this to False.
+    :ivar sentinel: If a message identical to the sentinel is received,
+        handle_report will set stopping to True.
     """
 
-    def __init__(self, config, channel, queue_name):
+    def __init__(self, config, connection_factory, queue_name):
         """Create a Receiver.
 
         :param config: An oops.Config to republish the OOPS reports.
-        :param channel: An amqplib Channel to listen for reports on.
+        :param connection_factory: An amqplib connection factory, used to make
+            the initial connection and to reconnect if that connection is
+            interrupted.
         :param queue_name: The queue to listen for reports on.
         """
         self.config = config
-        self.channel = channel
+        self.connection = None
+        self.channel = None
+        self.connection_factory = connection_factory
         self.queue_name = queue_name
-        self.stopping = False
+        self.sentinel = None
 
     def handle_report(self, message):
-        if self.stopping:
-            self.channel.basic_cancel(self.consume_tag)
+        if message.body == self.sentinel:
+            self.stopping = True
+            self.channel.basic_ack(message.delivery_tag)
+            return
         try:
             report = bson.loads(message.body)
         except KeyError:
@@ -56,6 +70,41 @@
         self.channel.basic_ack(message.delivery_tag)
     
     def run_forever(self):
-        self.consume_tag = self.channel.basic_consume(
-            self.queue_name, callback=self.handle_report)
-        self.channel.wait()
+        """Run in a loop handling messages.
+
+        If the amqp server is down or uncontactable for > 120 seconds, error
+        out.
+        """
+        self.stopping = False
+        self.went_bad = None
+        while (not self.stopping and
+               (not self.went_bad or time.time() < self.went_bad + 120)):
+            try:
+                self._run_forever()
+            except socket.error:
+                if not self.went_bad:
+                    self.went_bad = time.time()
+                # Don't probe immediately, give the network/process time to
+                # come back.
+                time.sleep(0.1)
+
+    def _run_forever(self):
+        self.connection = self.connection_factory()
+        # A successful connection: record this so run_forever won't bail early.
+        self.went_bad = None
+        try:
+            self.channel = self.connection.channel()
+            try:
+                self.consume_tag = self.channel.basic_consume(
+                    self.queue_name, callback=self.handle_report)
+                try:
+                    while True:
+                        self.channel.wait()
+                        if self.stopping:
+                            break
+                finally:
+                    self.channel.basic_cancel(self.consume_tag)
+            finally:
+                close_ignoring_EPIPE(self.channel)
+        finally:
+            close_ignoring_EPIPE(self.connection)

=== modified file 'oops_amqp/tests/__init__.py'
--- oops_amqp/tests/__init__.py	2011-10-10 21:49:50 +0000
+++ oops_amqp/tests/__init__.py	2011-10-18 03:36:23 +0000
@@ -16,8 +16,6 @@
 
 """Tests for oops_amqp."""
 
-import errno
-import socket
 from unittest import TestLoader
 
 from amqplib import client_0_8 as amqp
@@ -30,7 +28,10 @@
     ResourcedTestCase,
     )
 
+from oops_amqp.utils import close_ignoring_EPIPE
+
 __all__ = [
+    'ChannelFixture',
     'QueueFixture',
     'test_suite',
     'TestCase',
@@ -80,6 +81,25 @@
         self.channel.exchange_delete(self.exchange_name)
 
 
+class ChannelFixture(Fixture):
+    """Create an AMQP connection and channel for tests.
+    
+    :ivar connection: an amqplib connection.
+    :ivar channel: an amqplib channel
+    """
+
+    def __init__(self, connection_factory):
+        super(ChannelFixture, self).__init__()
+        self.connection_factory = connection_factory
+
+    def setUp(self):
+        super(ChannelFixture, self).setUp()
+        self.connection = self.connection_factory()
+        self.addCleanup(close_ignoring_EPIPE, self.connection)
+        self.channel = self.connection.channel()
+        self.addCleanup(close_ignoring_EPIPE, self.channel)
+
+
 class TestCase(testtools.TestCase, ResourcedTestCase):
     """Subclass to mix in testresources ResourcedTestCase."""
 
@@ -91,13 +111,6 @@
             self.rabbit.config.port), userid="guest", password="guest",
             virtual_host="/")
 
-    def close_ignoring_EPIPE(self, closable):
-        try:
-            closable.close()
-        except socket.error, e:
-            if e.errno != errno.EPIPE:
-                raise
-
 
 def test_suite():
     test_mod_names = [

=== modified file 'oops_amqp/tests/test_publisher.py'
--- oops_amqp/tests/test_publisher.py	2011-10-10 21:49:50 +0000
+++ oops_amqp/tests/test_publisher.py	2011-10-18 03:36:23 +0000
@@ -21,7 +21,9 @@
 import bson
 
 from oops_amqp import Publisher
+from oops_amqp.utils import close_ignoring_EPIPE
 from oops_amqp.tests import (
+    ChannelFixture,
     QueueFixture,
     TestCase,
     )
@@ -29,13 +31,36 @@
 
 class TestPublisher(TestCase):
 
+    def test_publish_inherit_id(self):
+        # OOPS id's can be set outside of Publisher().
+        channel = self.useFixture(
+            ChannelFixture(self.connection_factory)).channel
+        queue = self.useFixture(QueueFixture(channel, self.getUniqueString))
+        publisher = Publisher(self.connection_factory, queue.exchange_name, "",
+            inherit_id=True)
+        reference_oops = {'id': 'kept', 'akey': 'avalue'}
+        oops = dict(reference_oops)
+        expected_id = 'kept'
+        oops_id = publisher(oops)
+        # Publication returns the oops ID allocated.
+        self.assertEqual(expected_id, oops_id)
+        # The oops should not be altered by publication.
+        self.assertEqual(reference_oops, oops)
+        # The received OOPS should have the ID embedded and be a bson dict.
+        def check_oops(msg):
+            self.assertEqual(reference_oops, bson.loads(msg.body))
+            channel.basic_ack(msg.delivery_tag)
+            channel.basic_cancel(queue.queue_name)
+        channel.basic_consume(
+            queue.queue_name, callback=check_oops,
+            consumer_tag=queue.queue_name)
+        channel.wait()
+
     def test_publish(self):
         # Publishing an oops sends it to the exchange, making a connection as
         # it goes.
-        connection = self.connection_factory()
-        self.addCleanup(connection.close)
-        channel = connection.channel()
-        self.addCleanup(channel.close)
+        channel = self.useFixture(
+            ChannelFixture(self.connection_factory)).channel
         queue = self.useFixture(QueueFixture(channel, self.getUniqueString))
         publisher = Publisher(self.connection_factory, queue.exchange_name, "")
         reference_oops = {'akey': 'avalue'}
@@ -63,10 +88,8 @@
         # If amqp is down when a connection is attempted, None is returned to
         # indicate that publication failed - and publishing after it comes back
         # works.
-        connection = self.connection_factory()
-        self.addCleanup(self.close_ignoring_EPIPE, connection)
-        channel = connection.channel()
-        self.addCleanup(self.close_ignoring_EPIPE, channel)
+        channel = self.useFixture(
+            ChannelFixture(self.connection_factory)).channel
         queue = self.useFixture(QueueFixture(channel, self.getUniqueString))
         # The private method use and the restart of rabbit before it gets torn
         # down are bugs in rabbitfixture that will be fixed in a future
@@ -86,11 +109,10 @@
         # If amqp goes down after its been successfully used, None is returned
         # to indicate that publication failed - and publishing after it comes
         # back works.
-        connection = self.connection_factory()
-        self.addCleanup(self.close_ignoring_EPIPE, connection)
-        channel = connection.channel()
-        self.addCleanup(self.close_ignoring_EPIPE, channel)
-        queue = self.useFixture(QueueFixture(channel, self.getUniqueString))
+        channel = self.useFixture(
+            ChannelFixture(self.connection_factory)).channel
+        queue = self.useFixture(
+            QueueFixture(channel, self.getUniqueString))
         publisher = Publisher(self.connection_factory, queue.exchange_name, "")
         oops = {'akey': 42}
         self.assertNotEqual(None, publisher(oops))

=== modified file 'oops_amqp/tests/test_receiver.py'
--- oops_amqp/tests/test_receiver.py	2011-10-10 21:49:50 +0000
+++ oops_amqp/tests/test_receiver.py	2011-10-18 03:36:23 +0000
@@ -16,12 +16,16 @@
 
 """Tests for AMQP receiving."""
 
+import errno
+import socket
+
 from amqplib import client_0_8 as amqp
 import bson
 from oops import Config
 
 from oops_amqp import Receiver
 from oops_amqp.tests import (
+    ChannelFixture,
     QueueFixture,
     TestCase,
     )
@@ -29,49 +33,144 @@
 
 class TestReceiver(TestCase):
 
-    def test_receive_one(self):
-        # Receiving a message from AMQP should republish it unaltered.
-        reports = []
-        def capture(report):
-            reports.append(report)
-            return report['id']
-        expected_report = {'id': 'foo', 'otherkey': 42}
-        message = amqp.Message(bson.dumps(expected_report))
-        connection = self.connection_factory()
-        self.addCleanup(connection.close)
-        channel = connection.channel()
-        self.addCleanup(channel.close)
-        queue = self.useFixture(QueueFixture(channel, self.getUniqueString))
-        channel.basic_publish(message, queue.exchange_name, routing_key="")
-        config = Config()
-        config.publishers.append(capture)
-        receiver_channel = connection.channel()
-        self.addCleanup(receiver_channel.close)
-        receiver = Receiver(config, receiver_channel, queue.queue_name)
+    def test_stop_on_sentinel(self):
+        # A sentinel can be used to stop the receiver (useful for testing).
+        reports = []
+        def capture(report):
+            reports.append(report)
+            return report['id']
+        expected_report = {'id': 'foo', 'otherkey': 42}
+        message = amqp.Message(bson.dumps(expected_report))
+        channel = self.useFixture(
+            ChannelFixture(self.connection_factory)).channel
+        queue = self.useFixture(QueueFixture(channel, self.getUniqueString))
+        channel.basic_publish(
+            message, queue.exchange_name, routing_key="")
+        sentinel = "xxx"
+        channel.basic_publish(
+            amqp.Message(sentinel), queue.exchange_name, routing_key="")
+        config = Config()
+        config.publishers.append(capture)
+        receiver = Receiver(config, self.connection_factory, queue.queue_name)
+        receiver.sentinel = sentinel
+        receiver.run_forever()
+        self.assertEqual([expected_report], reports)
+
+    def test_stop_via_stopping(self):
+        # Setting the stopping field should stop the run_forever loop.
+        reports = []
+        def capture(report):
+            reports.append(report)
+            return report['id']
+        expected_report = {'id': 'foo', 'otherkey': 42}
+        message = amqp.Message(bson.dumps(expected_report))
+        channel = self.useFixture(
+            ChannelFixture(self.connection_factory)).channel
+        queue = self.useFixture(QueueFixture(channel, self.getUniqueString))
+        channel.basic_publish(
+            message, queue.exchange_name, routing_key="")
+        config = Config()
+        config.publishers.append(capture)
         # We don't want to loop forever: patch the channel so that after one
         # call to wait (which will get our injected message) the loop will shut
-        # down. This also checks we use the consume_tag correctly.
-        old_wait = receiver_channel.wait
-        def new_wait(allowed_methods=None):
-            receiver.stopping = True
-            return old_wait(allowed_methods=allowed_methods)
-        receiver_channel.wait = new_wait
+        # down.
+        def patching_factory():
+            connection = self.connection_factory()
+            old_channel = connection.channel
+            def new_channel():
+                result = old_channel()
+                old_wait = result.wait
+                def new_wait(allowed_methods=None):
+                    receiver.stopping = True
+                    return old_wait(allowed_methods=allowed_methods)
+                result.wait = new_wait
+                return result
+            connection.channel = new_channel
+            return connection
+        receiver = Receiver(config, patching_factory, queue.queue_name)
         receiver.run_forever()
         self.assertEqual([expected_report], reports)
 
     def test_run_forever(self):
-        # run_forever subscribes and then calls wait.
+        # run_forever subscribes and then calls wait in a loop.
         config = None
+        calls = []
         class FakeChannel:
-            def __init__(self):
-                self.calls = []
+            def __init__(self, calls):
+                self.calls = calls
             def basic_consume(self, queue_name, callback=None):
                 self.calls.append(('basic_consume', queue_name, callback))
+                return 'tag'
             def wait(self):
                 self.calls.append(('wait',))
-        channel = FakeChannel()
-        receiver = Receiver(None, channel, 'foo')
+                if len(self.calls) > 2:
+                    receiver.stopping = True
+            def basic_cancel(self, tag):
+                self.calls.append(('basic_cancel', tag))
+            def close(self):
+                pass
+        class FakeConnection:
+            def channel(self):
+                return FakeChannel(calls)
+            def close(self):
+                pass
+        receiver = Receiver(None, FakeConnection, 'foo')
         receiver.run_forever()
         self.assertEqual(
-            [('basic_consume', 'foo', receiver.handle_report), ('wait',)],
-            channel.calls)
+            [('basic_consume', 'foo', receiver.handle_report),
+             ('wait',),
+             ('wait',),
+             ('basic_cancel', 'tag')],
+            calls)
+
+    def test_tolerates_amqp_trouble(self):
+        # If the AMQP server is unavailable for a short period, the receiver
+        # will automatically reconnect.
+        # Break a connection to raise socket.error (which we know from the 
+        # publisher tests is what leaks through when rabbit is shutdown).
+        # We raise it the first time on each amqp method call.
+        reports = []
+        def capture(report):
+            reports.append(report)
+            return report['id']
+        expected_report = {'id': 'foo', 'otherkey': 42}
+        message = amqp.Message(bson.dumps(expected_report))
+        channel = self.useFixture(
+            ChannelFixture(self.connection_factory)).channel
+        queue = self.useFixture(QueueFixture(channel, self.getUniqueString))
+        channel.basic_publish(message, queue.exchange_name, routing_key="")
+        config = Config()
+        config.publishers.append(capture)
+        state = {}
+        def error_once(func):
+            def wrapped(*args, **kwargs):
+                func_ref = func.func_code
+                if func_ref in state:
+                    return func(*args, **kwargs)
+                else:
+                    state[func_ref] = True
+                    # Use EPIPE because the close() code checks that (though
+                    # the rest doesn't)
+                    raise socket.error(errno.EPIPE, "booyah")
+            return wrapped
+        @error_once
+        def patching_factory():
+            connection = self.connection_factory()
+            old_channel = connection.channel
+            @error_once
+            def new_channel():
+                result = old_channel()
+                result.wait = error_once(result.wait)
+                result.basic_consume = error_once(result.basic_consume)
+                result.basic_cancel = error_once(result.basic_cancel)
+                result.close = error_once(result.close)
+                return result
+            connection.channel = new_channel
+            connection.close = error_once(connection.close)
+            return connection
+        receiver = Receiver(config, patching_factory, queue.queue_name)
+        receiver.sentinel = "arhh"
+        channel.basic_publish(
+            amqp.Message("arhh"), queue.exchange_name, routing_key="")
+        receiver.run_forever()
+        self.assertEqual([expected_report], reports)

=== added file 'oops_amqp/utils.py'
--- oops_amqp/utils.py	1970-01-01 00:00:00 +0000
+++ oops_amqp/utils.py	2011-10-18 03:36:23 +0000
@@ -0,0 +1,32 @@
+# Copyright (c) 2011, Canonical Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Utility functions for oops_amqp."""
+
+import errno
+import socket
+
+__all__ = [
+    'close_ignoring_EPIPE',
+    ]
+
+
+def close_ignoring_EPIPE(closable):
+    try:
+        return closable.close()
+    except socket.error, e:
+        if e.errno != errno.EPIPE:
+            raise

=== modified file 'setup.py'
--- setup.py	2011-10-10 21:49:50 +0000
+++ setup.py	2011-10-18 03:36:23 +0000
@@ -23,7 +23,7 @@
         os.path.join(os.path.dirname(__file__), 'README'), 'rb').read()
 
 setup(name="oops_amqp",
-      version="0.0.1",
+      version="0.0.2",
       description=\
               "OOPS AMQP transport.",
       long_description=description,