Draft heartbeat support

dashi/__init__.py

@@ -6,7 +6,8 @@
 import sys
 import logging
 
-from kombu.connection import BrokerConnection
+from datetime import datetime, timedelta
+from kombu.connection import Connection
 from kombu.messaging import Consumer
 from kombu.pools import connections, producers
 from kombu.entity import Queue, Exchange
@@ -16,6 +17,7 @@
 
 log = logging.getLogger(__name__)
 
+DEFAULT_HEARTBEAT = 30
 
 class DashiConnection(object):
 
@@ -24,7 +26,8 @@ class DashiConnection(object):
     #TODO support connection info instead of uri
 
     def __init__(self, name, uri, exchange, durable=False, auto_delete=True,
-                 serializer=None, transport_options=None, ssl=False):
+                 serializer=None, transport_options=None, ssl=False, 
+                 heartbeat=DEFAULT_HEARTBEAT):
         """Set up a Dashi connection
 
         @param name: name of destination service queue used by consumers
@@ -38,7 +41,9 @@ def __init__(self, name, uri, exchange, durable=False, auto_delete=True,
         @param transport_options: custom parameter dict for the transport backend
         """
 
-        self._conn = BrokerConnection(uri, transport_options=transport_options, ssl=ssl)
+        self._heartbeat_interval = heartbeat
+        self._conn = Connection(uri, transport_options=transport_options,
+                ssl=ssl, heartbeat=self._heartbeat_interval)
         self._name = name
         self._exchange_name = exchange
         self._exchange = Exchange(name=exchange, type='direct',
@@ -216,6 +221,7 @@ def __init__(self, dashi, connection, name, exchange):
         self._ops = {}
         self._cancelled = False
         self._consumer_lock = threading.Lock()
+        self._last_heartbeat_check = datetime.min
 
         self.connect()
 
@@ -272,6 +278,9 @@ def _consume_one(self, timeout=None):
 
         # keep trying until a single event is drained or timeout hit
         while not self._cancelled:
+
+            self.heartbeat()
+
             try:
                 self._conn.drain_events(timeout=inner_timeout)
                 break
@@ -285,6 +294,20 @@ def _consume_one(self, timeout=None):
                     if elapsed + inner_timeout > timeout:
                         inner_timeout = timeout - elapsed
 
+    def heartbeat(self):
+        time_between_tics = timedelta(seconds=self._dashi._heartbeat_interval / 2)
+
+        if self._dashi.consumer_timeout > time_between_tics.seconds:
+            msg = "dashi consumer timeout (%s) must be half or smaller than the heartbeat interval %s" % (
+                    self._dashi.consumer_timeout, self._dashi._heartbeat_interval)
+
+            raise DashiError(msg)
+
+        if datetime.now() - self._last_heartbeat_check > time_between_tics:
+            self._last_heartbeat_check = datetime.now()
+            self._conn.heartbeat_check()
+
+
     def cancel(self, block=True):
         self._cancelled = True
         if block:

dashi/tests/test_dashi.py

@@ -6,12 +6,14 @@
 import logging
 import time
 
+from mock import Mock
 from nose.plugins.skip import SkipTest
 from kombu.pools import connections
 import kombu.pools
 
 import dashi
 import dashi.util
+from dashi.exceptions import DashiError
 from dashi.tests.util import who_is_calling
 
 log = logging.getLogger(__name__)
@@ -366,6 +368,32 @@ def test_handle_sender_kwarg(self):
         receiver.disconnect()
         assert_kombu_pools_empty()
 
+    def test_heartbeats(self):
+
+        receiver = TestReceiver(uri=self.uri, exchange="x1",
+            transport_options=self.transport_options)
+        receiver.conn.consumer_timeout = 100
+
+        receiver.handle("test1", "hello", sender_kwarg="sender")
+
+        caught_exp = None
+        try:
+            receiver.consume(1)
+        except DashiError, e:
+            caught_exp = e
+        assert caught_exp
+
+        receiver.conn.consumer_timeout = 1
+        receiver.consume(1)
+
+        receiver.clear()
+
+        receiver.cancel()
+
+        receiver.disconnect()
+        assert_kombu_pools_empty()
+
+
     def test_exceptions(self):
         class CustomNotFoundError(Exception):
             pass

setup.py

@@ -33,8 +33,8 @@
     "download_url" : "http://www.nimbusproject.org/downloads/dashi-%s.tar.gz" % VERSION,
 }
 
-install_requires = ['kombu>=2.1.2,<2.5.0', 'pyyaml']
-tests_require = ["nose"]
+install_requires = ['kombu>=2.5', 'pyyaml']
+tests_require = ["nose", "mock"]
 
 from setuptools import setup, find_packages
 setupdict['packages'] = find_packages()