Reconnect the connection thread if broker shuts down (#18)

* Reconnect the connection thread if broker shuts down

Make the subscriber and publisher interfaces more clear and stable.
Reconnect if the rabbitmq broker shuts down.
Configurable parameters for how many threads the subscriber is
allowed to create.
Keep track of published messages in order to requeue them if broker
goes down.
Create a base RabbitMQ class for shared methods.

* Updates according to review comments.

Added default value of '#' to routing key.
Changed '%s' to %r

Author: t-persson

eiffellib/lib/__init__.py

@@ -0,0 +1,15 @@
+# Copyright 2020 Axis Communications AB.
+#
+# For a full list of individual contributors, please see the commit history.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

eiffellib/lib/base_rabbitmq.py

@@ -0,0 +1,252 @@
+# Copyright 2020 Axis Communications AB.
+#
+# For a full list of individual contributors, please see the commit history.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""RabbitMQ base connection."""
+import time
+import logging
+import threading
+import ssl as _ssl
+import pika
+
+
+_LOG = logging.getLogger(__name__)
+
+
+# pylint:disable=too-many-instance-attributes
+class BaseRabbitMQ:
+    """Base RabbitMQ connection object."""
+
+    _connection = None
+    _channel = None
+    _was_active = False
+    _closing = False
+
+    should_reconnect = False
+    active = False
+    connection_thread = None
+    running = False
+
+    # pylint:disable=too-many-arguments
+    def __init__(self, host, port, username, password, vhost, ssl):
+        """Set connection parameters."""
+        parameters = {"port": port}
+        if ssl is True:
+            context = _ssl.create_default_context()
+            ssl_options = pika.SSLOptions(context, host)
+            parameters["ssl_options"] = ssl_options
+        if username and password:
+            parameters["credentials"] = pika.PlainCredentials(username, password)
+        if vhost:
+            parameters["virtual_host"] = vhost
+
+        self.parameters = pika.ConnectionParameters(host, **parameters)
+
+    def reset_parameters(self):
+        """Reset parameters to default."""
+        self.should_reconnect = False
+        self.active = False
+        self._closing = False
+        self._was_active = False
+
+    def _setup(self, channel):
+        """Setup channel. Called after channel is opened.
+
+        :param channel: RabbitMQ channel.
+        :type channel: :obj:`pika.channel.Channel`
+        """
+        raise NotImplementedError
+
+    def _start(self, *args, **kwargs):
+        """Start up the RabbitMQ connection. Call when connection is ready."""
+        raise NotImplementedError
+
+    def _cancel(self):
+        """Cancel the RabbitMQ connection."""
+        raise NotImplementedError
+
+    def _on_start(self):
+        """Called just before starting the connection."""
+
+    def _on_connect(self):
+        """Called just before connecting."""
+
+    def reconnect(self):
+        """Reconnect RabbitMQ connection."""
+        self.should_reconnect = True
+        self.stop()
+
+    def connect(self):
+        """Connect to RabbitMQ."""
+        self._connection = pika.SelectConnection(self.parameters,
+                                                 on_open_callback=self._connection_open,
+                                                 on_close_callback=self._connection_close,
+                                                 on_open_error_callback=self._connection_open_error)
+
+    def close_connection(self):
+        """Close the RabbitMQ connection."""
+        self.active = False
+        if self._connection.is_closing or self._connection.is_closed:
+            _LOG.info("Connection is closing or already closed.")
+        else:
+            _LOG.info("Closing connection")
+            self._connection.close()
+
+    def _connection_open(self, connection):
+        """Connection opened callback. Create a RabbitMQ channel.
+
+        :param connection: Connection that was just opened.
+        :type connection: :obj:`pika.connection.Connection`
+        """
+        _LOG.info("Connection opened")
+        _LOG.info("Creating a new channel.")
+        connection.channel(on_open_callback=self._channel_open)
+
+    def _connection_close(self, _, reason):
+        """Connection closed callback.
+
+        Close the connection if told to, otherwise attempt reconnect.
+
+        :param connection: Connection that was closed.
+        :type connection: :obj:`pika.connection.Connection`
+        :param reason: Reason the connection was closed.
+        :type reason: str
+        """
+        self._channel = None
+        if self._closing:
+            self._connection.ioloop.stop()
+        else:
+            _LOG.warning("Connection closed, reconnecting: %r", reason)
+            self.reconnect()
+
+    def _connection_open_error(self, _, error):
+        """Connection error when starting callback. Reconnect.
+
+        :param connection: Connection that was closed.
+        :type connection: :obj:`pika.Connection`
+        :param error: Reason the connection was closed.
+        :type error: str
+        """
+        _LOG.error("Failed to open connection: %r", error)
+        self.reconnect()
+
+    def _channel_open(self, channel):
+        """Channel opened callback. Set close callback and call setup channel.
+
+        :param channel: Channel that was just opened.
+        :type channel: :obj:`pika.channel.Channel`
+        """
+        _LOG.info("Channel opened")
+        self._channel = channel
+        self._channel.add_on_close_callback(self._channel_closed)
+        self._setup(channel)
+
+    def _channel_closed(self, channel, reason):
+        """Channel closed callback. Close connection."""
+        _LOG.warning("Channel %i was closed: %r", channel, reason)
+        self.close_connection()
+
+    def close_channel(self):
+        """Close the RabbitMQ channel."""
+        _LOG.info("Closing the channel.")
+        self._channel.close()
+
+    def run(self):
+        """Reset parameters, connect to RabbitMQ and start the ioloop.
+
+        This is a blocking method.
+        """
+        self._on_connect()
+        self.reset_parameters()
+        self.connect()
+        self._connection.ioloop.start()
+
+    def keep_alive(self):
+        """Reconnect forever if the should_reconnect flag is set.
+
+        This is a blocking method.
+        """
+        reconnect_delay = 0
+        while True:
+            try:
+                self.run()
+            except KeyboardInterrupt:
+                self.stop()
+                break
+            if self.should_reconnect:
+                self.stop()
+                if self._was_active:
+                    reconnect_delay = 0
+                else:
+                    reconnect_delay += 1
+                if reconnect_delay > 30:
+                    reconnect_delay = 30
+                _LOG.info("Reconnecting after %d seconds", reconnect_delay)
+                time.sleep(reconnect_delay)
+            else:
+                break
+        self.running = False
+
+    def is_alive(self):
+        """Check if connection is alive or not.
+
+        Note that this only checks the 'running' flag and does not
+        deep-dive into the RabbitMQ connection and channel.
+
+        :return: Whether this connection is running or not.
+        :rtype: bool
+        """
+        return self.running
+
+    def wait_start(self):
+        """Block until connection starts."""
+        while not self.is_alive():
+            time.sleep(0.1)
+
+    def wait_close(self):
+        """Block until publisher closes."""
+        while self.is_alive():
+            time.sleep(0.1)
+
+    def start(self, wait=True):
+        """Start the RabbitMQ connection in a thread.
+
+        :param wait: Block until the connection starts. Defaults to True.
+        :type wait: bool
+        """
+        self._on_start()
+        self.connection_thread = threading.Thread(target=self.keep_alive)
+        self.connection_thread.daemon = True
+        self.connection_thread.start()
+        if wait:
+            self.wait_start()
+
+    def stop(self):
+        """Stop the RabbitMQ connection."""
+        if not self._closing:
+            self._closing = True
+            _LOG.info("Stopping")
+            if self.active:
+                self._cancel()
+                try:
+                    # Start the ioloop again in order for the callbacks
+                    # to fire, so that we can do a clean shutdown.
+                    self._connection.ioloop.start()
+                except RuntimeError:
+                    pass
+            else:
+                self._connection.ioloop.stop()
+            _LOG.info("Stopped")
+
+    close = stop  # Backwards compatibility

eiffellib/publishers/eiffel_publisher.py

@@ -1,4 +1,4 @@
-# Copyright 2019 Axis Communications AB.
+# Copyright 2020 Axis Communications AB.
 #
 # For a full list of individual contributors, please see the commit history.
 #
@@ -27,9 +27,12 @@ def start(self):
         """Start the pika rabbitmq connection."""
         raise NotImplementedError
 
-    def send_event(self, event):
+    def send_event(self, event, block=True):
         """Validate and send an eiffel event to server.
 
+        :param block: Set to True in order to block until ready.
+                      Default: True
+        :type block: bool
         :param event: Event to send.
         :type event: :obj:`eiffellib.events.eiffel_base_event.EiffelBaseEvent`
         """