Finalizing Heartbeat bug patch

Author: eandersson

.travis.yml

@@ -13,8 +13,7 @@ install:
   - if [[ $TRAVIS_PYTHON_VERSION == '2.6' ]]; then pip install unittest2; fi
   - pip install -r requirements.txt
   - pip install -r test-requirements.txt
-  - if [[ $TRAVIS_PYTHON_VERSION == '3.2' ]]; then pip install coverage==3.7.1; fi
-script: nosetests -v -l DEBUG --logging-level=DEBUG --with-coverage --cover-package=amqpstorm
+script: nosetests -v -l DEBUG --logging-level=DEBUG --with-coverage --cover-package=amqpstorm --with-timer --timer-top-n 25
 after_success:
   - bash <(curl -s https://codecov.io/bash)
 services:

CHANGELOG

@@ -1,5 +1,10 @@
 # Changelog
 
+### Version 1.4.0
+- All classes are now slotted.
+- New improved Heartbeat Monitor.
+    - If no data has been sent within the Heartbeat interval, the client will now send a Heartbeat to the server. - Thanks David Schneider.
+
 ### Version 1.3.4
 - Dropped Python 3.2 Support.
 - Fixed incorrect SSL warning when adding heartbeat or timeout to uri string (#18) - Thanks Adam Mills.

amqpstorm/__init__.py

@@ -1,5 +1,5 @@
 """AMQP-Storm."""
-__version__ = '1.3.4'  # noqa
+__version__ = '1.4.0'  # noqa
 __author__ = 'eandersson'  # noqa
 
 import logging

amqpstorm/base.py

@@ -1,13 +1,20 @@
 """AMQP-Storm Base."""
 
+import locale
 import threading
 
+AUTH_MECHANISM = 'PLAIN'
 IDLE_WAIT = 0.01
 FRAME_MAX = 131072
+MAX_CHANNELS = 65535
+LOCALE = locale.getdefaultlocale()[0] or 'en_US'
 
 
 class Stateful(object):
     """Stateful"""
+    __slots__ = [
+        '_exceptions', '_lock', '_state'
+    ]
     CLOSED = 0
     CLOSING = 1
     OPENING = 2
@@ -81,6 +88,9 @@ def check_for_errors(self):
 
 class BaseChannel(Stateful):
     """AMQP BaseChannel"""
+    __slots__ = [
+        '_channel_id', '_consumer_tags'
+    ]
 
     def __init__(self, channel_id):
         super(BaseChannel, self).__init__()
@@ -129,7 +139,9 @@ def remove_consumer_tag(self, tag=None):
 
 class BaseMessage(object):
     """AMQP BaseMessage"""
-    __slots__ = ['_body', '_channel', '_method', '_properties']
+    __slots__ = [
+        '_body', '_channel', '_method', '_properties'
+    ]
 
     def __init__(self, channel, **message):
         """
@@ -169,7 +181,9 @@ def to_tuple(self):
 
 class Handler(object):
     """Operations Handler (e.g. Queue, Exchange)"""
-    __slots__ = ['_channel']
+    __slots__ = [
+        '_channel'
+    ]
 
     def __init__(self, channel):
         self._channel = channel

amqpstorm/basic.py

@@ -19,6 +19,7 @@
 
 class Basic(Handler):
     """AMQP Channel.basic"""
+    __slots__ = []
 
     def qos(self, prefetch_count=0, prefetch_size=0, global_=False):
         """Specify quality of service.

amqpstorm/channel.py

@@ -26,6 +26,10 @@
 
 class Channel(BaseChannel):
     """Connection.channel"""
+    __slots__ = [
+        'confirming_deliveries', 'consumer_callback', 'rpc', '_basic',
+        '_connection', '_exchange', '_inbound', '_queue'
+    ]
 
     def __init__(self, channel_id, connection, rpc_timeout):
         super(Channel, self).__init__(channel_id)
@@ -34,9 +38,9 @@ def __init__(self, channel_id, connection, rpc_timeout):
         self.consumer_callback = None
         self._inbound = []
         self._connection = connection
-        self._basic = None
-        self._exchange = None
-        self._queue = None
+        self._basic = Basic(self)
+        self._exchange = Exchange(self)
+        self._queue = Queue(self)
 
     def __enter__(self):
         return self
@@ -58,23 +62,23 @@ def basic(self):
 
         :rtype: Basic
         """
-        return self._lazy_load_handler('_basic', Basic)
+        return self._basic
 
     @property
     def exchange(self):
         """RabbitMQ Exchange Operations.
 
         :rtype: Exchange
         """
-        return self._lazy_load_handler('_exchange', Exchange)
+        return self._exchange
 
     @property
     def queue(self):
         """RabbitMQ Queue Operations.
 
         :rtype: Queue
         """
-        return self._lazy_load_handler('_queue', Queue)
+        return self._queue
 
     def open(self):
         """Open Channel.
@@ -340,10 +344,10 @@ def _build_message(self):
         with self.lock:
             if len(self._inbound) < 2:
                 return None
-            result = self._build_message_headers()
-            if not result:
+            headers = self._build_message_headers()
+            if not headers:
                 return None
-            basic_deliver, content_header = result
+            basic_deliver, content_header = headers
             body = self._build_message_body(content_header.body_size)
 
         message = Message(channel=self,
@@ -387,16 +391,3 @@ def _build_message_body(self, body_size):
                 break
             body += body_piece.value
         return body
-
-    def _lazy_load_handler(self, name, handler_class):
-        """Lazy load operations (e.g. Queue, Exchange)
-
-        :param name:
-        :param Handler handler_class: Handler (e.g. Queue)
-        :return:
-        """
-        handler = getattr(self, name)
-        if not handler:
-            handler = handler_class(self)
-            setattr(self, name, handler)
-        return handler

amqpstorm/channel0.py

@@ -1,6 +1,5 @@
 """AMQP-Storm Connection.Channel0."""
 
-import locale
 import logging
 import platform
 
@@ -9,15 +8,16 @@
 from pamqp.specification import Connection as pamqp_connection
 
 from amqpstorm import __version__
+from amqpstorm.base import AUTH_MECHANISM
 from amqpstorm.base import FRAME_MAX
+from amqpstorm.base import LOCALE
+from amqpstorm.base import MAX_CHANNELS
 from amqpstorm.base import Stateful
 from amqpstorm.exception import AMQPConnectionError
 from amqpstorm.compatibility import try_utf8_decode
 
-AUTH_MECHANISM = 'PLAIN'
-LOCALE = locale.getdefaultlocale()[0] or 'en_US'
+
 LOGGER = logging.getLogger(__name__)
-MAX_CHANNELS = 65535
 
 
 class Channel0(object):
@@ -118,7 +118,7 @@ def _write_frame(self, frame_out):
     def _send_start_ok_frame(self, frame_in):
         """Send Start OK frame.
 
-        :param pamqp_spec.Frame frame_out: Amqp frame.
+        :param pamqp_spec.Connection.StartOk frame_in: Amqp frame.
         :return:
         """
         if 'PLAIN' not in try_utf8_decode(frame_in.mechanisms):

amqpstorm/connection.py

@@ -25,6 +25,9 @@
 
 class Connection(Stateful):
     """AMQP Connection"""
+    __slots__ = [
+        'heartbeat', 'parameters', '_channel0', '_channels', '_io'
+    ]
 
     def __init__(self, hostname, username, password, port=5672, **kwargs):
         """
@@ -56,13 +59,12 @@ def __init__(self, hostname, username, password, port=5672, **kwargs):
             'ssl_options': kwargs.get('ssl_options', {})
         }
         self._validate_parameters()
+        self._io = IO(self.parameters, exceptions=self._exceptions,
+                      on_read=self._read_buffer)
         self._channel0 = Channel0(self)
+        self._channels = {}
         self.heartbeat = Heartbeat(self.parameters['heartbeat'],
                                    self._channel0.send_heartbeat)
-        self._io = IO(self.parameters, on_read=self._read_buffer,
-                      on_write=self.heartbeat.register_write)
-
-        self._channels = {}
         if not kwargs.get('lazy', False):
             self.open()
 
@@ -119,7 +121,7 @@ def open(self):
         LOGGER.debug('Connection Opening')
         self.set_state(self.OPENING)
         self._exceptions = []
-        self._io.open(self._exceptions)
+        self._io.open()
         self._send_handshake()
         self._wait_for_connection_to_open()
         self.heartbeat.start(self._exceptions)
@@ -183,6 +185,7 @@ def write_frame(self, channel_id, frame_out):
         :return:
         """
         frame_data = pamqp_frame.marshal(frame_out, channel_id)
+        self.heartbeat.register_write()
         self._io.write_to_socket(frame_data)
 
     def write_frames(self, channel_id, multiple_frames):
@@ -195,6 +198,7 @@ def write_frames(self, channel_id, multiple_frames):
         frame_data = EMPTY_BUFFER
         for single_frame in multiple_frames:
             frame_data += pamqp_frame.marshal(single_frame, channel_id)
+        self.heartbeat.register_write()
         self._io.write_to_socket(frame_data)
 
     def _validate_parameters(self):

amqpstorm/exchange.py

@@ -13,6 +13,7 @@
 
 class Exchange(Handler):
     """AMQP Channel.exchange"""
+    __slots__ = []
 
     def declare(self, exchange='', exchange_type='direct', passive=False,
                 durable=False, auto_delete=False, arguments=None):

amqpstorm/heartbeat.py

@@ -12,6 +12,7 @@ class Heartbeat(object):
     """AMQP Internal Heartbeat Checker"""
 
     def __init__(self, interval, send_heartbeat=None):
+        self.send_heartbeat = send_heartbeat
         self._lock = threading.Lock()
         self._running = threading.Event()
         self._timer = None
@@ -20,7 +21,6 @@ def __init__(self, interval, send_heartbeat=None):
         self._writes_since_check = 0
         self._interval = interval
         self._threshold = 0
-        self.send_heartbeat = send_heartbeat
 
     def register_read(self):
         """Register that a frame has been received.
@@ -67,12 +67,13 @@ def _check_for_life_signs(self):
             First check if any data has been sent, if not send a heartbeat.
 
             If we have not received a heartbeat, or any data what so ever
-            we should raise an exception so that we can close the connection.
+            within two intervals, we need to raise an exception so
+            that we can close the connection.
 
             RabbitMQ may not necessarily send heartbeats if the connection
             is busy, so we only raise if no frame has been received.
 
-        :return:
+        :rtype: bool
         """
         if not self._running.is_set():
             return False
@@ -90,7 +91,7 @@ def _check_for_life_signs(self):
                     if self._exceptions is None:
                         raise why
                     self._exceptions.append(why)
-                    return
+                    return False
             else:
                 self._threshold = 0
         finally: