JobWorker gets clocks and heartbeats

Added a monotonic to JobWorker to schedule heartbeast

6 files changed, 98 insertions, 39 deletions

diff --git a/docs/protocol.rst b/docs/protocol.rst
index c32501c..72ab361 100644
--- a/docs/protocol.rst
+++ b/docs/protocol.rst
@@ -57,6 +57,7 @@ FRAME  Value          Description
 1      eMQP/1.0       Protocol version
 2      ACK            command
 3      _MSGID_        A unique id for the msg
+4      _MSGID_        The message id of the message this ACK is acknowledging
 ====== ============== ===========
 
 eMQP / Client
@@ -165,7 +166,7 @@ Below is a table which defines and describes the headers.
 =============== ======= ======= ======= ===========
 Header          REQUEST PUBLISH Default Description
 =============== ======= ======= ======= ===========
-reply-requested X               False     Once the job is finished, send a reply back with information from the job. If there is no information reply with a True value.
+reply-requested X               False   Once the job is finished, send a reply back with information from the job. If there is no information reply with a True value.
 retry-count:#   X               0       Retry a failed job this many times before accepting defeat.
-guarantee       X               False     Ensure the job completes by letting someone else worry about a success reply.
+guarantee       X               False   Ensure the job completes by letting someone else worry about a success reply.
 =============== ======= ======= ======= ===========
diff --git a/eventmq/jobmanager.py b/eventmq/jobmanager.py
index d00d3e6..9b57285 100644
--- a/eventmq/jobmanager.py
+++ b/eventmq/jobmanager.py
@@ -17,17 +17,14 @@
 ================================
 Ensures things about jobs and spawns the actual tasks
 """
-from time import sleep
 import uuid
 
-from . import constants
-from . import exceptions
-from . import log
-from . import utils
+from . import conf, constants, exceptions, log, utils
 from .poller import Poller, POLLIN
 from .sender import Sender
 from .utils.messages import send_emqp_message as sendmsg
 import utils.messages
+from .utils.timeutils import monotonic, timestamp
 
 logger = log.get_logger(__file__)
 
@@ -58,44 +55,101 @@ class JobManager(object):
 
         self.status = constants.STATUS.ready
 
-        # Are we waiting for an acknowledgment for something?
-        self.awaiting_ack = False
+        # Are we waiting for acknowledgement from someone that we've connected?
+        self.awaiting_startup_ack = False
+
+        # Meta data such as times, and counters are stored here
+        self._meta = {
+            'last_sent_heartbeat': 0,
+            'last_received_heartbeat': 0,
+            'heartbeat_miss_count': 0,
+        }
 
     def start(self, addr='tcp://127.0.0.1:47291'):
         """
         Connect to `addr` and begin listening for job requests
 
         Args:
-            args (str): connection string to connect to
+            addr (str): connection string to connect to
         """
         self.status = constants.STATUS.connecting
         self.incoming.connect(addr)
 
-        self.awaiting_ack = True
+        self.awaiting_startup_ack = True
+
+        while self.awaiting_startup_ack:
+            self.send_inform()
+            # Poller timeout is in ms so we multiply it to get seconds
+            events = self.poller.poll(conf.RECONNECT_TIMEOUT * 1000)
+            if self.incoming in events:
+                msg = self.incoming.recv_multipart()
+                # We don't want to accidentally start processing jobs before
+                # our conenction has been setup completely and acknowledged.
+                if msg[2] != "ACK":
+                    continue
+                self.process_message(msg)
 
-        #while self.awaiting_ack:
-        self.send_inform()
-        #    sleep(5)
+        if not self.awaiting_startup_ack:
+            logger.info('Starting to listen for jobs')
+            self._start_event_loop()
 
+    def _start_event_loop(self):
+        """
+        Starts the actual eventloop. Usually called by :meth:`JobManager.start`
+        """
         self.status = constants.STATUS.connected
 
         while True:
-            events = self.poller.poll(1000)
+            now = monotonic()
+            events = self.poller.poll()
 
             if events.get(self.incoming) == POLLIN:
                 msg = self.incoming.recv_multipart()
                 self.process_message(msg)
 
+            # Send a HEARTBEAT if necessary
+            if now - self._meta['last_sent_heartbeat'] >= \
+               conf.HEARTBEAT_INTERVAL:
+                if conf.SUPER_DEBUG:
+                    logger.debug(now - self._meta['last_sent_heartbeat'])
+                self.send_heartbeat()
+
+            # Do something about any missed HEARTBEAT
+            if now - self._meta['last_received_heartbeat'] >= \
+               conf.HEARTBEAT_TIMEOUT:
+                # Update as if we got the last heartbeat so we can check in
+                # interval again
+                self._meta['heartbeat_miss_count'] += 1
+                self._meta['last_received_heartbeat'] = monotonic()
+                if self._meta['heartbeat_miss_count'] >= \
+                   conf.HEARTBEAT_LIVENESS:
+                    logger.critical('The broker appears to have gone away. '
+                                    'Reconnecting...')
+                    break
+                print self._meta['heartbeat_miss_count']
+
     def process_message(self, msg):
         """
         Processes a message
+
+        Args:
+            msg: The message received from the socket to parse and process.
+                Processing takes form of calling an `on_COMMAND` method.
         """
+        # Any received message should count as a heartbeat
+        self._meta['last_received_heartbeat'] = monotonic()
+        if self._meta['heartbeat_miss_count']:
+            self._meta['heartbeat_miss_count'] = 0  # Reset the miss count too
+
         try:
             message = utils.messages.parse_message(msg)
         except exceptions.InvalidMessageError:
             logger.error('Invalid message: %s' % str(msg))
             return
 
+        if conf.SUPER_DEBUG:
+            logger.debug("Received Message: %s" % msg)
+
         command = message[0]
         msgid = message[1]
         message = message[2]
@@ -108,21 +162,24 @@ class JobManager(object):
             logger.warning('No handler for %s found (tried: %s)' %
                            (command, ('on_%s' % command.lower)))
 
-    def process_job(self, msg):
-        pass
-
-    def sync(self):
-        pass
-
     def send_inform(self):
         """
-        Send an INFORM frame
+        Send an INFORM command
         """
         sendmsg(self.incoming, 'INFORM', 'default_queuename')
 
-    def on_ack(self, msgid, message):
+    def send_heartbeat(self):
+        """
+        Send a HEARTBEAT command to the connected broker
+        """
+        sendmsg(self.incoming, 'HEARTBEAT', str(timestamp()))
+        self._meta['last_sent_heartbeat'] = monotonic()
+
+    def on_ack(self, msgid, ackd_msgid):
         """
         Sets :attr:`awaiting_ack` to False
         """
-        logger.info('Recieved ACK')
-        self.awaiting_ack = False
+        # The msgid is the only frame in the message
+        ackd_msgid = ackd_msgid[0]
+        logger.info('Received ACK for %s' % ackd_msgid)
+        self.awaiting_startup_ack = False
diff --git a/eventmq/router.py b/eventmq/router.py
index 77b752d..b6a4cdf 100644
--- a/eventmq/router.py
+++ b/eventmq/router.py
@@ -18,15 +18,16 @@
 Routes messages to workers (that are in named queues).
 """
 import uuid
-
 from zmq.eventloop import ioloop
 
 from .constants import STATUS
-from . import exceptions, log, receiver, utils
+from . import exceptions, log, receiver
 from .utils.messages import (
     send_emqp_router_message as sendmsg,
     parse_router_message
 )
+from .utils.time import monotonic
+
 
 logger = log.get_logger(__file__)
 
@@ -73,12 +74,12 @@ class Router(object):
 
         ioloop.IOLoop.instance().start()
 
-    def send_ack(self, socket, recipient):
+    def send_ack(self, socket, recipient, msgid):
         """
         Sends an ACK response
         """
         logger.info('Sending ACK to %s' % recipient)
-        sendmsg(socket, recipient, 'ACK')
+        sendmsg(socket, recipient, 'ACK', msgid)
 
     def on_inform(self, sender, msgid, msg):
         """
@@ -92,7 +93,7 @@ class Router(object):
         else:
             self.queues[queue_name] = (sender,)
 
-        self.send_ack(self.outgoing, sender)
+        self.send_ack(self.outgoing, sender, msgid)
 
     def on_receive_request(self, msg):
         """
diff --git a/eventmq/tests/test_utils.py b/eventmq/tests/test_utils.py
index db3307d..8ce7f6e 100644
--- a/eventmq/tests/test_utils.py
+++ b/eventmq/tests/test_utils.py
@@ -14,14 +14,13 @@
 # along with eventmq.  If not, see <http://www.gnu.org/licenses/>.
 import unittest
 
-from .. import exceptions
-from .. import utils
-import utils.messages
+from .. import exceptions, utils
+from ..utils import messages, classes
 
 
 class TestCase(unittest.TestCase):
     def test_generate_msgid(self):
-        msgid = utils.messages.generate_msgid()
+        msgid = messages.generate_msgid()
 
         self.assertEqual(type(msgid), str)
 
@@ -31,9 +30,9 @@ class TestCase(unittest.TestCase):
         emq_frame_manymsg = emq_headers + ('many', 'parts')
         emq_frame_nomsg = emq_headers
 
-        singlemsg = utils.messages.parse_router_message(emq_frame_singlemsg)
-        manymsg = utils.messages.parse_router_message(emq_frame_manymsg)
-        nomsg = utils.messages.parse_router_message(emq_frame_nomsg)
+        singlemsg = messages.parse_router_message(emq_frame_singlemsg)
+        manymsg = messages.parse_router_message(emq_frame_manymsg)
+        nomsg = messages.parse_router_message(emq_frame_nomsg)
 
         self.assertEqual(singlemsg[0], emq_frame_singlemsg[0])
         self.assertEqual(singlemsg[1], emq_frame_singlemsg[3])
@@ -52,4 +51,4 @@ class TestCase(unittest.TestCase):
 
         broken_message = ('dlkajfs', 'lkasdjf')
         with self.assertRaises(exceptions.InvalidMessageError):
-            utils.messages.parse_router_message(broken_message)
+            messages.parse_router_message(broken_message)
diff --git a/eventmq/utils/__init__.py b/eventmq/utils/__init__.py
index af8a4dc..4fd9a89 100644
--- a/eventmq/utils/__init__.py
+++ b/eventmq/utils/__init__.py
@@ -26,6 +26,7 @@ like creating message more simple.
 """
 import uuid
 
+
 def random_characters():
     """
     Returns some random characters of a specified length
diff --git a/requirements.txt b/requirements.txt
index c4848d0..9724011 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,7 +1,7 @@
 pyzmq
 six
 tornado
-
+monotonic==0.4    # A clock who's time is not changed. used for scheduling
 
 # Documentation
 sphinxcontrib-napoleon