Merge ~cjwatson/launchpad:remove-rabbitmq-sessions into launchpad:master

[Colin Watson]

Remove RabbitMQ session infrastructure

This was only used by the longpoll mechanism that was removed in commit df2fe07c8766a80affa8467c5491af647f923840.

I started to try to modify this code to support multiple RabbitMQ broker URLs, then realized that would be a lot easier if I removed a big chunk of it first.

[Jürgen Gmach (jugmac00)]

review: Approve

diff --git a/lib/lp/services/configure.zcml b/lib/lp/services/configure.zcml
index 1461879..77dac94 100644
--- a/lib/lp/services/configure.zcml
+++ b/lib/lp/services/configure.zcml
@@ -21,7 +21,6 @@
   <include package=".mail" />
   <include package=".memcache" />
   <include package=".messages" />
-  <include package=".messaging" />
   <include package=".oauth" />
   <include package=".openid" />
   <include package=".profile" />
diff --git a/lib/lp/services/messaging/configure.zcml b/lib/lp/services/messaging/configure.zcml
deleted file mode 100644
index f8d23bd..0000000
--- a/lib/lp/services/messaging/configure.zcml
+++ /dev/null
@@ -1,18 +0,0 @@
-<!-- Copyright 2011 Canonical Ltd.  This software is licensed under the
-     GNU Affero General Public License version 3 (see the file LICENSE).
--->
-<configure
-    xmlns="http://namespaces.zope.org/zope"
-    xmlns:browser="http://namespaces.zope.org/browser"
-    xmlns:i18n="http://namespaces.zope.org/i18n"
-    i18n_domain="launchpad">
-  <utility
-      provides=".interfaces.IMessageSession"
-      component=".rabbit.unreliable_session" />
-  <subscriber
-      for="lp.services.webapp.interfaces.IFinishReadOnlyRequestEvent"
-      handler=".rabbit.session_finish_handler" />
-  <subscriber
-      for="lp.services.webapp.interfaces.IFinishReadOnlyRequestEvent"
-      handler=".rabbit.unreliable_session_finish_handler" />
-</configure>
diff --git a/lib/lp/services/messaging/interfaces.py b/lib/lp/services/messaging/interfaces.py
index c2c5dee..c73efac 100644
--- a/lib/lp/services/messaging/interfaces.py
+++ b/lib/lp/services/messaging/interfaces.py
@@ -4,96 +4,14 @@
 """Messaging interfaces."""
 
 __all__ = [
-    "IMessageConsumer",
-    "IMessageProducer",
-    "IMessageSession",
     "MessagingException",
     "MessagingUnavailable",
-    "QueueEmpty",
-    "QueueNotFound",
 ]
 
 
-from zope.interface import Interface
-from zope.schema import Bool
-
-
 class MessagingException(Exception):
     """Failure in messaging."""
 
 
 class MessagingUnavailable(MessagingException):
     """Messaging systems are not available."""
-
-
-class QueueNotFound(MessagingException):
-    """Raised if the queue was not found."""
-
-
-class QueueEmpty(MessagingException):
-    """Raised if there are no queued messages on a non-blocking read."""
-
-
-class IMessageSession(Interface):
-
-    is_connected = Bool(
-        "Whether the session is connected to the messaging system."
-    )
-
-    def connect():
-        """Connect to the messaging system.
-
-        If the session is already connected this should be a no-op.
-        """
-
-    def disconnect():
-        """Disconnect from the messaging system.
-
-        If the session is already disconnected this should be a no-op.
-        """
-
-    def flush():
-        """Run deferred tasks."""
-
-    def finish():
-        """Flush the session and reset."""
-
-    def reset():
-        """Reset the session."""
-
-    def defer(func, *args, **kwargs):
-        """Schedule something to happen when this session is finished."""
-
-    def getProducer(name):
-        """Get a `IMessageProducer` associated with this session."""
-
-    def getConsumer(name):
-        """Get a `IMessageConsumer` associated with this session."""
-
-
-class IMessageConsumer(Interface):
-    def receive(blocking=True):
-        """Receive data from the queue.
-
-        :raises EmptyQueue: If non-blocking and the queue is empty.
-        """
-
-
-class IMessageProducer(Interface):
-    def send(data):
-        """Serialize `data` into JSON and send it to the queue on commit."""
-
-    def sendNow(data):
-        """Serialize `data` into JSON and send it to the queue immediately."""
-
-    def associateConsumer(consumer):
-        """Make the consumer receive messages from this producer on commit.
-
-        :param consumer: An `IMessageConsumer`
-        """
-
-    def associateConsumerNow(consumer):
-        """Make the consumer receive messages from this producer.
-
-        :param consumer: An `IMessageConsumer`
-        """
diff --git a/lib/lp/services/messaging/rabbit.py b/lib/lp/services/messaging/rabbit.py
index 46b3b18..6c8bd34 100644
--- a/lib/lp/services/messaging/rabbit.py
+++ b/lib/lp/services/messaging/rabbit.py
@@ -6,51 +6,12 @@
 __all__ = [
     "connect",
     "is_configured",
-    "session",
-    "unreliable_session",
 ]
 
-import json
-import sys
-import threading
-import time
-from collections import deque
-from functools import partial
-
 import amqp
-import transaction
-from transaction._transaction import Status as TransactionStatus
-from zope.interface import implementer
 
 from lp.services.config import config
-from lp.services.messaging.interfaces import (
-    IMessageConsumer,
-    IMessageProducer,
-    IMessageSession,
-    MessagingUnavailable,
-    QueueEmpty,
-    QueueNotFound,
-)
-
-LAUNCHPAD_EXCHANGE = "launchpad-exchange"
-
-
-@implementer(transaction.interfaces.ISynchronizer)
-class RabbitSessionTransactionSync:
-    def __init__(self, session):
-        self.session = session
-
-    def newTransaction(self, txn):
-        pass
-
-    def beforeCompletion(self, txn):
-        pass
-
-    def afterCompletion(self, txn):
-        if txn.status == TransactionStatus.COMMITTED:
-            self.session.finish()
-        else:
-            self.session.reset()
+from lp.services.messaging.interfaces import MessagingUnavailable
 
 
 def is_configured():
@@ -78,217 +39,3 @@ def connect():
     )
     connection.connect()
     return connection
-
-
-@implementer(IMessageSession)
-class RabbitSession(threading.local):
-
-    exchange = LAUNCHPAD_EXCHANGE
-
-    def __init__(self):
-        super().__init__()
-        self._connection = None
-        self._deferred = deque()
-        # Maintain sessions according to transaction boundaries. Keep a strong
-        # reference to the sync because the transaction manager does not. We
-        # need one per thread (definining it here is enough to ensure that).
-        self._sync = RabbitSessionTransactionSync(self)
-        transaction.manager.registerSynch(self._sync)
-
-    @property
-    def is_connected(self):
-        """See `IMessageSession`."""
-        return self._connection is not None and self._connection.connected
-
-    def connect(self):
-        """See `IMessageSession`.
-
-        Open a connection for this thread if necessary. Connections cannot be
-        shared between threads.
-        """
-        if self._connection is None or not self._connection.connected:
-            self._connection = connect()
-        return self._connection
-
-    def disconnect(self):
-        """See `IMessageSession`."""
-        if self._connection is not None:
-            try:
-                self._connection.close()
-            except OSError:
-                # Socket error is fine; the connection is still closed.
-                pass
-            finally:
-                self._connection = None
-
-    def flush(self):
-        """See `IMessageSession`."""
-        tasks = self._deferred
-        while len(tasks) != 0:
-            tasks.popleft()()
-
-    def finish(self):
-        """See `IMessageSession`."""
-        try:
-            self.flush()
-        finally:
-            self.reset()
-
-    def reset(self):
-        """See `IMessageSession`."""
-        self._deferred.clear()
-        self.disconnect()
-
-    def defer(self, func, *args, **kwargs):
-        """See `IMessageSession`."""
-        self._deferred.append(partial(func, *args, **kwargs))
-
-    def getProducer(self, name):
-        """See `IMessageSession`."""
-        return RabbitRoutingKey(self, name)
-
-    def getConsumer(self, name):
-        """See `IMessageSession`."""
-        return RabbitQueue(self, name)
-
-
-# Per-thread sessions.
-session = RabbitSession()
-session_finish_handler = lambda event: session.finish()
-
-
-class RabbitUnreliableSession(RabbitSession):
-    """An "unreliable" `RabbitSession`.
-
-    Unreliable in this case means that certain errors in deferred tasks are
-    silently suppressed. This means that services can continue to function
-    even in the absence of a running and fully functional message queue.
-
-    Other types of errors are also caught because we don't want this
-    subsystem to destabilise other parts of Launchpad but we nonetheless
-    record OOPses for these.
-
-    XXX: We only suppress MessagingUnavailable for now because we want to
-    monitor this closely before we add more exceptions to the
-    suppressed_errors list. Potential candidates are `MessagingException`,
-    `IOError` or `amqp.AMQPException`.
-    """
-
-    suppressed_errors = (MessagingUnavailable,)
-
-    def finish(self):
-        """See `IMessageSession`.
-
-        Suppresses errors listed in `suppressed_errors`. Also suppresses
-        other errors but files an oops report for these.
-        """
-        try:
-            super().finish()
-        except self.suppressed_errors:
-            pass
-        except Exception:
-            from lp.services.webapp import errorlog
-
-            errorlog.globalErrorUtility.raising(sys.exc_info())
-
-
-# Per-thread "unreliable" sessions.
-unreliable_session = RabbitUnreliableSession()
-unreliable_session_finish_handler = lambda event: unreliable_session.finish()
-
-
-class RabbitMessageBase:
-    """Base class for all RabbitMQ messaging."""
-
-    def __init__(self, session):
-        self.session = IMessageSession(session)
-        self._channel = None
-
-    @property
-    def channel(self):
-        if self._channel is None or not self._channel.is_open:
-            connection = self.session.connect()
-            self._channel = connection.channel()
-            self._channel.exchange_declare(
-                self.session.exchange,
-                "direct",
-                durable=False,
-                auto_delete=False,
-                nowait=False,
-            )
-        return self._channel
-
-
-@implementer(IMessageProducer)
-class RabbitRoutingKey(RabbitMessageBase):
-    """A RabbitMQ data origination point."""
-
-    def __init__(self, session, routing_key):
-        super().__init__(session)
-        self.key = routing_key
-
-    def associateConsumer(self, consumer):
-        """Only receive messages for requested routing key."""
-        self.session.defer(self.associateConsumerNow, consumer)
-
-    def associateConsumerNow(self, consumer):
-        """Only receive messages for requested routing key."""
-        # The queue will be auto-deleted 5 minutes after its last use.
-        # http://www.rabbitmq.com/extensions.html#queue-leases
-        self.channel.queue_declare(
-            consumer.name,
-            nowait=False,
-            auto_delete=False,
-            arguments={"x-expires": 300000},
-        )  # 5 minutes.
-        self.channel.queue_bind(
-            queue=consumer.name,
-            exchange=self.session.exchange,
-            routing_key=self.key,
-            nowait=False,
-        )
-
-    def send(self, data):
-        """See `IMessageProducer`."""
-        self.session.defer(self.sendNow, data)
-
-    def sendNow(self, data):
-        """Immediately send a message to the broker."""
-        json_data = json.dumps(data)
-        msg = amqp.Message(json_data)
-        self.channel.basic_publish(
-            exchange=self.session.exchange, routing_key=self.key, msg=msg
-        )
-
-
-@implementer(IMessageConsumer)
-class RabbitQueue(RabbitMessageBase):
-    """A RabbitMQ Queue."""
-
-    def __init__(self, session, name):
-        super().__init__(session)
-        self.name = name
-
-    def receive(self, timeout=0.0):
-        """Pull a message from the queue.
-
-        :param timeout: Wait a maximum of `timeout` seconds before giving up,
-            trying at least once.
-        :raises QueueEmpty: if the timeout passes.
-        """
-        endtime = time.time() + timeout
-        while True:
-            try:
-                message = self.channel.basic_get(self.name)
-                if message is None:
-                    if time.time() > endtime:
-                        raise QueueEmpty()
-                    time.sleep(0.1)
-                else:
-                    self.channel.basic_ack(message.delivery_tag)
-                    return json.loads(message.body)
-            except amqp.ChannelError as error:
-                if error.reply_code == 404:
-                    raise QueueNotFound()
-                else:
-                    raise
diff --git a/lib/lp/services/messaging/tests/__init__.py b/lib/lp/services/messaging/tests/__init__.py
deleted file mode 100644
index e69de29..0000000
--- a/lib/lp/services/messaging/tests/__init__.py
+++ /dev/null
diff --git a/lib/lp/services/messaging/tests/test_rabbit.py b/lib/lp/services/messaging/tests/test_rabbit.py
deleted file mode 100644
index f82a0a1..0000000
--- a/lib/lp/services/messaging/tests/test_rabbit.py
+++ /dev/null
@@ -1,447 +0,0 @@
-# Copyright 2011 Canonical Ltd.  This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-"""Messaging utility tests."""
-
-from functools import partial
-from itertools import count
-
-import transaction
-from testtools.testcase import ExpectedException
-from transaction._transaction import Status as TransactionStatus
-from zope.component import getUtility
-from zope.event import notify
-
-from lp.services.messaging.interfaces import (
-    IMessageConsumer,
-    IMessageProducer,
-    IMessageSession,
-    MessagingUnavailable,
-    QueueEmpty,
-    QueueNotFound,
-)
-from lp.services.messaging.rabbit import (
-    RabbitMessageBase,
-    RabbitQueue,
-    RabbitRoutingKey,
-    RabbitSession,
-    RabbitSessionTransactionSync,
-    RabbitUnreliableSession,
-)
-from lp.services.messaging.rabbit import session as global_session
-from lp.services.messaging.rabbit import (
-    unreliable_session as global_unreliable_session,
-)
-from lp.services.webapp.interfaces import FinishReadOnlyRequestEvent
-from lp.testing import TestCase, monkey_patch
-from lp.testing.fakemethod import FakeMethod
-from lp.testing.faketransaction import FakeTransaction
-from lp.testing.layers import LaunchpadFunctionalLayer, RabbitMQLayer
-from lp.testing.matchers import Provides
-
-# RabbitMQ is not (yet) torn down or reset between tests, so here are sources
-# of distinct names.
-queue_names = ("queue.%d" % num for num in count(1))
-key_names = ("key.%d" % num for num in count(1))
-
-
-class FakeRabbitSession:
-    def __init__(self):
-        self.log = []
-
-    def finish(self):
-        self.log.append("finish")
-
-    def reset(self):
-        self.log.append("reset")
-
-
-class TestRabbitSessionTransactionSync(TestCase):
-    def test_interface(self):
-        self.assertThat(
-            RabbitSessionTransactionSync(None),
-            Provides(transaction.interfaces.ISynchronizer),
-        )
-
-    def test_afterCompletion_COMMITTED(self):
-        txn = FakeTransaction()
-        txn.status = TransactionStatus.COMMITTED
-        fake_session = FakeRabbitSession()
-        sync = RabbitSessionTransactionSync(fake_session)
-        sync.afterCompletion(txn)
-        self.assertEqual(["finish"], fake_session.log)
-
-    def test_afterCompletion_ACTIVE(self):
-        txn = FakeTransaction()
-        txn.status = TransactionStatus.ACTIVE
-        fake_session = FakeRabbitSession()
-        sync = RabbitSessionTransactionSync(fake_session)
-        sync.afterCompletion(txn)
-        self.assertEqual(["reset"], fake_session.log)
-
-
-class RabbitTestCase(TestCase):
-
-    layer = RabbitMQLayer
-
-    def tearDown(self):
-        super().tearDown()
-        global_session.reset()
-        global_unreliable_session.reset()
-
-
-class TestRabbitSession(RabbitTestCase):
-
-    session_factory = RabbitSession
-
-    def test_interface(self):
-        session = self.session_factory()
-        self.assertThat(session, Provides(IMessageSession))
-
-    def test_connect(self):
-        session = self.session_factory()
-        self.assertFalse(session.is_connected)
-        connection = session.connect()
-        self.assertTrue(session.is_connected)
-        self.assertIs(connection, session._connection)
-
-    def test_connect_with_incomplete_configuration(self):
-        self.pushConfig("rabbitmq", host="none")
-        session = self.session_factory()
-        with ExpectedException(
-            MessagingUnavailable, "Incomplete configuration"
-        ):
-            session.connect()
-
-    def test_disconnect(self):
-        session = self.session_factory()
-        session.connect()
-        session.disconnect()
-        self.assertFalse(session.is_connected)
-
-    def test_disconnect_with_error(self):
-        session = self.session_factory()
-        session.connect()
-        old_close = session._connection.close
-
-        def new_close(*args, **kwargs):
-            old_close(*args, **kwargs)
-            raise OSError
-
-        with monkey_patch(session._connection, close=new_close):
-            session.disconnect()
-            self.assertFalse(session.is_connected)
-
-    def test_is_connected(self):
-        # is_connected is False once a connection has been closed.
-        session = self.session_factory()
-        session.connect()
-        # Close the connection without using disconnect().
-        session._connection.close()
-        self.assertFalse(session.is_connected)
-
-    def test_defer(self):
-        task = lambda foo, bar: None
-        session = self.session_factory()
-        session.defer(task, "foo", bar="baz")
-        self.assertEqual(1, len(session._deferred))
-        [deferred_task] = session._deferred
-        self.assertIsInstance(deferred_task, partial)
-        self.assertIs(task, deferred_task.func)
-        self.assertEqual(("foo",), deferred_task.args)
-        self.assertEqual({"bar": "baz"}, deferred_task.keywords)
-
-    def test_flush(self):
-        # RabbitSession.flush() runs deferred tasks.
-        log = []
-        task = lambda: log.append("task")
-        session = self.session_factory()
-        session.defer(task)
-        session.connect()
-        session.flush()
-        self.assertEqual(["task"], log)
-        self.assertEqual([], list(session._deferred))
-        self.assertTrue(session.is_connected)
-
-    def test_reset(self):
-        # RabbitSession.reset() resets session variables and does not run
-        # deferred tasks.
-        log = []
-        task = lambda: log.append("task")
-        session = self.session_factory()
-        session.defer(task)
-        session.connect()
-        session.reset()
-        self.assertEqual([], log)
-        self.assertEqual([], list(session._deferred))
-        self.assertFalse(session.is_connected)
-
-    def test_finish(self):
-        # RabbitSession.finish() resets session variables after running
-        # deferred tasks.
-        log = []
-        task = lambda: log.append("task")
-        session = self.session_factory()
-        session.defer(task)
-        session.connect()
-        session.finish()
-        self.assertEqual(["task"], log)
-        self.assertEqual([], list(session._deferred))
-        self.assertFalse(session.is_connected)
-
-    def test_getProducer(self):
-        session = self.session_factory()
-        producer = session.getProducer("foo")
-        self.assertIsInstance(producer, RabbitRoutingKey)
-        self.assertIs(session, producer.session)
-        self.assertEqual("foo", producer.key)
-
-    def test_getConsumer(self):
-        session = self.session_factory()
-        consumer = session.getConsumer("foo")
-        self.assertIsInstance(consumer, RabbitQueue)
-        self.assertIs(session, consumer.session)
-        self.assertEqual("foo", consumer.name)
-
-
-class TestRabbitUnreliableSession(TestRabbitSession):
-
-    session_factory = RabbitUnreliableSession
-    layer = RabbitMQLayer
-
-    def setUp(self):
-        super().setUp()
-        self.prev_oops = self.getOops()
-
-    def getOops(self):
-        try:
-            self.oops_capture.sync()
-            return self.oopses[-1]
-        except IndexError:
-            return None
-
-    def assertNoOops(self):
-        oops_report = self.getOops()
-        self.assertEqual(repr(self.prev_oops), repr(oops_report))
-
-    def assertOops(self, text_in_oops):
-        oops_report = self.getOops()
-        self.assertNotEqual(
-            repr(self.prev_oops), repr(oops_report), "No OOPS reported!"
-        )
-        self.assertIn(text_in_oops, str(oops_report))
-
-    def _test_finish_suppresses_exception(self, exception):
-        # Simple helper to test that the given exception is suppressed
-        # when raised by finish().
-        session = self.session_factory()
-        session.defer(FakeMethod(failure=exception))
-        session.finish()  # Look, no exceptions!
-
-    def test_finish_suppresses_MessagingUnavailable(self):
-        self._test_finish_suppresses_exception(
-            MessagingUnavailable("Messaging borked.")
-        )
-        self.assertNoOops()
-
-    def test_finish_suppresses_other_errors_with_oopses(self):
-        exception = Exception("That hent worked.")
-        self._test_finish_suppresses_exception(exception)
-        self.assertOops(str(exception))
-
-
-class TestRabbitMessageBase(RabbitTestCase):
-    def test_session(self):
-        base = RabbitMessageBase(global_session)
-        self.assertIs(global_session, base.session)
-
-    def test_channel(self):
-        # Referencing the channel property causes the session to connect.
-        base = RabbitMessageBase(global_session)
-        self.assertFalse(base.session.is_connected)
-        channel = base.channel
-        self.assertTrue(base.session.is_connected)
-        self.assertIsNot(None, channel)
-        # The same channel is returned every time.
-        self.assertIs(channel, base.channel)
-
-    def test_channel_session_closed(self):
-        # When the session is disconnected the channel is thrown away too.
-        base = RabbitMessageBase(global_session)
-        channel1 = base.channel
-        base.session.disconnect()
-        channel2 = base.channel
-        self.assertNotEqual(channel1, channel2)
-
-
-class TestRabbitRoutingKey(RabbitTestCase):
-    def test_interface(self):
-        routing_key = RabbitRoutingKey(global_session, next(key_names))
-        self.assertThat(routing_key, Provides(IMessageProducer))
-
-    def test_associateConsumer(self):
-        # associateConsumer() only associates the consumer at transaction
-        # commit time. However, order is preserved.
-        consumer = RabbitQueue(global_session, next(queue_names))
-        routing_key = RabbitRoutingKey(global_session, next(key_names))
-        routing_key.associateConsumer(consumer)
-        # The session is still not connected.
-        self.assertFalse(global_session.is_connected)
-        routing_key.sendNow("now")
-        routing_key.send("later")
-        # The queue is not found because the consumer has not yet been
-        # associated with the routing key and the queue declared.
-        self.assertRaises(QueueNotFound, consumer.receive, timeout=2)
-        transaction.commit()
-        # Now that the transaction has been committed, the consumer is
-        # associated, and receives the deferred message.
-        self.assertEqual("later", consumer.receive(timeout=2))
-
-    def test_associateConsumerNow(self):
-        # associateConsumerNow() associates the consumer right away.
-        consumer = RabbitQueue(global_session, next(queue_names))
-        routing_key = RabbitRoutingKey(global_session, next(key_names))
-        routing_key.associateConsumerNow(consumer)
-        routing_key.sendNow("now")
-        routing_key.send("later")
-        # There is already something in the queue.
-        self.assertEqual("now", consumer.receive(timeout=2))
-        transaction.commit()
-        # Now that the transaction has been committed there is another item in
-        # the queue.
-        self.assertEqual("later", consumer.receive(timeout=2))
-
-    def test_send(self):
-        consumer = RabbitQueue(global_session, next(queue_names))
-        routing_key = RabbitRoutingKey(global_session, next(key_names))
-        routing_key.associateConsumerNow(consumer)
-
-        for data in range(90, 100):
-            routing_key.send(data)
-
-        routing_key.sendNow("sync")
-        # There is nothing in the queue except the sync we just sent.
-        self.assertEqual("sync", consumer.receive(timeout=2))
-
-        # Messages get sent on commit
-        transaction.commit()
-        for data in range(90, 100):
-            self.assertEqual(data, consumer.receive())
-
-        # There are no more messages. They have all been consumed.
-        routing_key.sendNow("sync")
-        self.assertEqual("sync", consumer.receive(timeout=2))
-
-    def test_sendNow(self):
-        consumer = RabbitQueue(global_session, next(queue_names))
-        routing_key = RabbitRoutingKey(global_session, next(key_names))
-        routing_key.associateConsumerNow(consumer)
-
-        for data in range(50, 60):
-            routing_key.sendNow(data)
-            received_data = consumer.receive(timeout=2)
-            self.assertEqual(data, received_data)
-
-    def test_does_not_connect_session_immediately(self):
-        # RabbitRoutingKey does not connect the session until necessary.
-        RabbitRoutingKey(global_session, next(key_names))
-        self.assertFalse(global_session.is_connected)
-
-
-class TestRabbitQueue(RabbitTestCase):
-    def test_interface(self):
-        consumer = RabbitQueue(global_session, next(queue_names))
-        self.assertThat(consumer, Provides(IMessageConsumer))
-
-    def test_receive(self):
-        consumer = RabbitQueue(global_session, next(queue_names))
-        routing_key = RabbitRoutingKey(global_session, next(key_names))
-        routing_key.associateConsumerNow(consumer)
-
-        for data in range(55, 65):
-            routing_key.sendNow(data)
-            self.assertEqual(data, consumer.receive(timeout=2))
-
-        # All the messages received were consumed.
-        self.assertRaises(QueueEmpty, consumer.receive, timeout=2)
-
-        # New connections to the queue see an empty queue too.
-        consumer.session.disconnect()
-        consumer = RabbitQueue(global_session, next(queue_names))
-        routing_key = RabbitRoutingKey(global_session, next(key_names))
-        routing_key.associateConsumerNow(consumer)
-        self.assertRaises(QueueEmpty, consumer.receive, timeout=2)
-
-    def test_does_not_connect_session_immediately(self):
-        # RabbitQueue does not connect the session until necessary.
-        RabbitQueue(global_session, next(queue_names))
-        self.assertFalse(global_session.is_connected)
-
-
-class TestRabbit(RabbitTestCase):
-    """Integration-like tests for the RabbitMQ messaging abstractions."""
-
-    def get_synced_sessions(self):
-        try:
-            syncs_set = transaction.manager.manager._synchs
-        except KeyError:
-            return set()
-        else:
-            return {
-                sync.session
-                for sync in syncs_set.data.values()
-                if isinstance(sync, RabbitSessionTransactionSync)
-            }
-
-    def test_global_session(self):
-        self.assertIsInstance(global_session, RabbitSession)
-        self.assertIn(global_session, self.get_synced_sessions())
-
-    def test_global_unreliable_session(self):
-        self.assertIsInstance(
-            global_unreliable_session, RabbitUnreliableSession
-        )
-        self.assertIn(global_unreliable_session, self.get_synced_sessions())
-
-    def test_abort(self):
-        consumer = RabbitQueue(global_session, next(queue_names))
-        routing_key = RabbitRoutingKey(global_session, next(key_names))
-        routing_key.associateConsumerNow(consumer)
-
-        for data in range(90, 100):
-            routing_key.send(data)
-
-        # Messages sent using send() are forgotten on abort.
-        transaction.abort()
-        self.assertRaises(QueueEmpty, consumer.receive, timeout=2)
-
-
-class TestRabbitWithLaunchpad(RabbitTestCase):
-    """Integration-like tests for the RabbitMQ messaging abstractions."""
-
-    layer = LaunchpadFunctionalLayer
-
-    def test_utility(self):
-        # The unreliable session is registered as the default IMessageSession
-        # utility.
-        self.assertIs(global_unreliable_session, getUtility(IMessageSession))
-
-    def _test_session_finish_read_only_request(self, session):
-        # When a read-only request ends the session is also finished.
-        log = []
-        task = lambda: log.append("task")
-        session.defer(task)
-        session.connect()
-        notify(FinishReadOnlyRequestEvent(None, None))
-        self.assertEqual(["task"], log)
-        self.assertEqual([], list(session._deferred))
-        self.assertFalse(session.is_connected)
-
-    def test_global_session_finish_read_only_request(self):
-        # When a read-only request ends the global_session is finished too.
-        self._test_session_finish_read_only_request(global_session)
-
-    def test_global_unreliable_session_finish_read_only_request(self):
-        # When a read-only request ends the global_unreliable_session is
-        # finished too.
-        self._test_session_finish_read_only_request(global_unreliable_session)
diff --git a/lib/lp/services/webapp/doc/webapp-publication.rst b/lib/lp/services/webapp/doc/webapp-publication.rst
index 0fbbbe6..c478f92 100644
--- a/lib/lp/services/webapp/doc/webapp-publication.rst
+++ b/lib/lp/services/webapp/doc/webapp-publication.rst
@@ -1001,7 +1001,7 @@ afterCall().  For example, this publication subclass will simply print
 some string in its finishReadOnlyRequest().
 
     >>> class MyPublication(LaunchpadBrowserPublication):
-    ...     def finishReadOnlyRequest(self, request, ob, txn):
+    ...     def finishReadOnlyRequest(self, txn):
     ...         print("booo!")
     ...
 
diff --git a/lib/lp/services/webapp/interfaces.py b/lib/lp/services/webapp/interfaces.py
index aa53686..c6ef41a 100644
--- a/lib/lp/services/webapp/interfaces.py
+++ b/lib/lp/services/webapp/interfaces.py
@@ -818,23 +818,6 @@ class ICheckBoxWidgetLayout(IAlwaysSubmittedWidget):
     """A widget that is displayed like a check box with label to the right."""
 
 
-class IFinishReadOnlyRequestEvent(Interface):
-    """An event which gets sent when the publication is ended"""
-
-    object = Attribute("The object to which this request pertains.")
-
-    request = Attribute("The active request.")
-
-
-@implementer(IFinishReadOnlyRequestEvent)
-class FinishReadOnlyRequestEvent:
-    """An event which gets sent when the publication is ended"""
-
-    def __init__(self, ob, request):
-        self.object = ob
-        self.request = request
-
-
 class StormRangeFactoryError(Exception):
     """Raised when a Storm result set cannot be used for slicing by a
     StormRangeFactory.
diff --git a/lib/lp/services/webapp/publication.py b/lib/lp/services/webapp/publication.py
index bbe8b83..1557a46 100644
--- a/lib/lp/services/webapp/publication.py
+++ b/lib/lp/services/webapp/publication.py
@@ -54,7 +54,6 @@ from lp.services.features.flags import NullFeatureController
 from lp.services.oauth.interfaces import IOAuthSignedRequest
 from lp.services.statsd.interfaces.statsd_client import IStatsdClient
 from lp.services.webapp.interfaces import (
-    FinishReadOnlyRequestEvent,
     ILaunchpadRoot,
     IOpenLaunchBag,
     IPlacelessAuthUtility,
@@ -506,7 +505,7 @@ class LaunchpadBrowserPublication(
         # Abort the transaction on a read-only request.
         # NOTHING AFTER THIS SHOULD CAUSE A RETRY.
         if request.method in ["GET", "HEAD"]:
-            self.finishReadOnlyRequest(request, ob, txn)
+            self.finishReadOnlyRequest(txn)
         elif txn.isDoomed():
             # The following sends an abort to the database, even though the
             # transaction is still doomed.
@@ -528,13 +527,12 @@ class LaunchpadBrowserPublication(
             # calling beforeTraversal or doing proper cleanup.
             pass
 
-    def finishReadOnlyRequest(self, request, ob, txn):
+    def finishReadOnlyRequest(self, txn):
         """Hook called at the end of a read-only request.
 
         By default it abort()s the transaction, but subclasses may need to
         commit it instead, so they must overwrite this.
         """
-        notify(FinishReadOnlyRequestEvent(ob, request))
         txn.abort()
 
     def callTraversalHooks(self, request, ob):
diff --git a/lib/lp/services/webapp/tests/test_servers.py b/lib/lp/services/webapp/tests/test_servers.py
index e0fa7e6..14820e0 100644
--- a/lib/lp/services/webapp/tests/test_servers.py
+++ b/lib/lp/services/webapp/tests/test_servers.py
@@ -32,8 +32,6 @@ from lp.services.auth.enums import AccessTokenScope
 from lp.services.identity.interfaces.account import AccountStatus
 from lp.services.oauth.interfaces import TokenException
 from lp.services.webapp.interaction import get_interaction_extras
-from lp.services.webapp.interfaces import IFinishReadOnlyRequestEvent
-from lp.services.webapp.publication import LaunchpadBrowserPublication
 from lp.services.webapp.servers import (
     ApplicationServerSettingRequestFactory,
     FeedsBrowserRequest,
@@ -49,7 +47,7 @@ from lp.services.webapp.servers import (
     WebServiceTestRequest,
     web_service_request_to_browser_request,
 )
-from lp.testing import EventRecorder, TestCase, TestCaseWithFactory, logout
+from lp.testing import TestCase, TestCaseWithFactory, logout
 from lp.testing.layers import DatabaseFunctionalLayer, FunctionalLayer
 from lp.testing.publication import get_request_and_publication
 
@@ -782,55 +780,6 @@ class LoggingTransaction:
         self.log.append("ABORT")
 
 
-class TestFinishReadOnlyRequest(TestCase):
-    # Publications that have a finishReadOnlyRequest() method are obliged to
-    # fire an IFinishReadOnlyRequestEvent.
-
-    def _test_publication(self, publication, expected_transaction_log):
-        # publication.finishReadOnlyRequest() issues an
-        # IFinishReadOnlyRequestEvent and alters the transaction.
-        fake_request = object()
-        fake_object = object()
-        fake_transaction = LoggingTransaction()
-
-        with EventRecorder() as event_recorder:
-            publication.finishReadOnlyRequest(
-                fake_request, fake_object, fake_transaction
-            )
-
-        self.assertEqual(expected_transaction_log, fake_transaction.log)
-
-        finish_events = [
-            event
-            for event in event_recorder.events
-            if IFinishReadOnlyRequestEvent.providedBy(event)
-        ]
-        self.assertEqual(
-            1,
-            len(finish_events),
-            (
-                "Expected only one IFinishReadOnlyRequestEvent, but "
-                "got: %r" % finish_events
-            ),
-        )
-
-        [finish_event] = finish_events
-        self.assertIs(fake_request, finish_event.request)
-        self.assertIs(fake_object, finish_event.object)
-
-    def test_WebServicePub_fires_FinishReadOnlyRequestEvent(self):
-        # WebServicePublication.finishReadOnlyRequest() issues an
-        # IFinishReadOnlyRequestEvent and aborts the transaction.
-        publication = WebServicePublication(None)
-        self._test_publication(publication, ["ABORT"])
-
-    def test_LaunchpadBrowserPub_fires_FinishReadOnlyRequestEvent(self):
-        # LaunchpadBrowserPublication.finishReadOnlyRequest() issues an
-        # IFinishReadOnlyRequestEvent and aborts the transaction.
-        publication = LaunchpadBrowserPublication(None)
-        self._test_publication(publication, ["ABORT"])
-
-
 class TestWebServiceAccessTokens(TestCaseWithFactory):
     """Test personal access tokens for the webservice.