Connection thread dies when broker shuts down

[magnusbaeck]

Description

When the RabbitMQ broker is shut down the connection thread crashes in the _run() method:

Traceback (most recent call last):
  File "/home/magnus/src/eiffel-pythonlib/eiffellib/subscribers/rabbitmq_subscriber.py", line 240, in _run
    self.connection.sleep(0.1)
  File "/home/magnus/src/eiffel-pythonlib/.env/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 852, in sleep
    self.process_data_events(time_limit)
  File "/home/magnus/src/eiffel-pythonlib/.env/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 829, in process_data_events
    self._flush_output(timer.is_ready, common_terminator)
  File "/home/magnus/src/eiffel-pythonlib/.env/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 523, in _flush_output
    raise self._closed_result.value.error
pika.exceptions.ConnectionClosedByBroker: (320, "CONNECTION_FORCED - broker forced connection closure with reason 'shutdown'")
Attempting reconnection in 5s
Exception in thread Thread-1:
Traceback (most recent call last):
  File "/home/magnus/src/eiffel-pythonlib/eiffellib/subscribers/rabbitmq_subscriber.py", line 240, in _run
    self.connection.sleep(0.1)
  File "/home/magnus/src/eiffel-pythonlib/.env/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 852, in sleep
    self.process_data_events(time_limit)
  File "/home/magnus/src/eiffel-pythonlib/.env/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 829, in process_data_events
    self._flush_output(timer.is_ready, common_terminator)
  File "/home/magnus/src/eiffel-pythonlib/.env/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 523, in _flush_output
    raise self._closed_result.value.error
pika.exceptions.ConnectionClosedByBroker: (320, "CONNECTION_FORCED - broker forced connection closure with reason 'shutdown'")

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/lib/python3.6/threading.py", line 916, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.6/threading.py", line 864, in run
    self._target(*self._args, **self._kwargs)
  File "/home/magnus/src/eiffel-pythonlib/eiffellib/subscribers/rabbitmq_subscriber.py", line 249, in _run
    self.connection.sleep(5)
  File "/home/magnus/src/eiffel-pythonlib/.env/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 852, in sleep
    self.process_data_events(time_limit)
  File "/home/magnus/src/eiffel-pythonlib/.env/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 828, in process_data_events
    with _IoloopTimerContext(time_limit, self._impl) as timer:
  File "/home/magnus/src/eiffel-pythonlib/.env/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 191, in __enter__
    self._duration, self._callback_result.signal_once)
  File "/home/magnus/src/eiffel-pythonlib/.env/lib/python3.6/site-packages/pika/adapters/base_connection.py", line 205, in _adapter_call_later
    return self._nbio.call_later(delay, callback)
  File "/home/magnus/src/eiffel-pythonlib/.env/lib/python3.6/site-packages/pika/adapters/utils/selector_ioloop_adapter.py", line 222, in call_later
    return _TimerHandle(self._loop.call_later(delay, callback), self._loop)
  File "/home/magnus/src/eiffel-pythonlib/.env/lib/python3.6/site-packages/pika/adapters/select_connection.py", line 444, in call_later
    return self._timer.call_later(delay, callback)
  File "/home/magnus/src/eiffel-pythonlib/.env/lib/python3.6/site-packages/pika/adapters/select_connection.py", line 258, in call_later
    heapq.heappush(self._timeout_heap, timeout)
TypeError: heap argument must be a list

https://stackoverflow.com/q/55570743/414355 suggests that this could be caused by attempting to access the same Pika connection from multiple threads, but I'm not sure that's the case here. Not starting the delegator thread doesn't make a difference, and with an empty queue it seems all connection/channel operations are done from the connection thread.

This happens reliably with the example program from README.rst, RabbitMQ 3.8.2, Python 3.6.9, and the following Python state:

$ pip freeze
attrs==19.3.0
-e git+git@github.com:magnusbaeck/eiffel-pythonlib@7b4eb839ae39ba9d165bd19df86f3eb6cbe3233f#egg=eiffellib
jsonschema==3.0.0a2
pika==1.0.1
pkg-resources==0.0.0
pyrsistent==0.15.6
six==1.13.0

Motivation

Broker restarts are routine and should obviously be handled gracefully by all clients.

[magnusbaeck]

Apart from fixing this bug, it would be a good idea if there was a standard way for clients to wait for the connection thread to die (which under normal circumstances should never happen). That way the client has the opportunity to restart the subscriber or just terminate to allow for a supervisor process like systemd or Docker to respawn the program.

[t-persson]

@magnusbaeck I have pushed a test branch to "issue_16_test_branch" where I've attempted to fix your issues.
This is a rewrite of the subscribers and publishers so that they are more stable.
It will now attempt to reconnect if the broker shuts down.
You can now configure the number of threads the subscriber can start at any given time.
You can now wait for the subscribers and publishers to start or stop.

The interfaces should be the same as before, so please test them out. I know there is missing documentation and that there should be a base connection handler, this is just for you to test to make sure that I'm delivering the fix you need :)

from eiffellib.subscribers import RabbitMQSubscriber

SUBSCRIBER = RabbitMQSubscriber(host="127.0.0.1", port=5672, exchange="amq.fanout", ssl=False, queue="eiffel")
SUBSCRIBER.start()  # Will automatically wait for the subscriber to start. Disable by passing 'False'

def receiver(event, _):
    print(event)

SUBSCRIBER.subscribe("EiffelActivityTriggeredEvent", receiver)
SUBSCRIBER.subscribe("EiffelActivityStartedEvent", receiver)
SUBSCRIBER.subscribe("EiffelActivityFinishedEvent", receiver)
SUBSCRIBER.wait_close()  # Wait until the subscriber is closed manually or by a process killer.

import time
from eiffellib.publishers import RabbitMQPublisher
from eiffellib.events import (EiffelActivityTriggeredEvent,
                                            EiffelActivityStartedEvent,
                                            EiffelActivityFinishedEvent)

PUBLISHER = RabbitMQPublisher(host="127.0.0.1", port=5672,
                              exchange="amq.fanout", ssl=False)

PUBLISHER.start()    # Will automatically wait for the subscriber to start. Disable by passing 'False'

try:
    while True:
        ACTIVITY = EiffelActivityTriggeredEvent()
        ACTIVITY.data.add("name", "Hello!")
        PUBLISHER.send_event(ACTIVITY)

        STARTED = EiffelActivityStartedEvent()
        STARTED.links.add("ACTIVITY_EXECUTION", ACTIVITY)
        PUBLISHER.send_event(STARTED)

        FINISHED = EiffelActivityFinishedEvent()
        FINISHED.data.add("outcome", {"conclusion": "SUCCESSFUL"})
        FINISHED.links.add("ACTIVITY_EXECUTION", ACTIVITY)
        PUBLISHER.send_event(FINISHED)
        time.sleep(0.1)
finally:
    PUBLISHER.close()
    PUBLISHER.wait_close()

Now you should be able to shutdown your broker and the subscribers and publishers should reconnect.

Please test this out, and I'll continue to test a bit and make sure it's stable.