Merge pull request #15 from enderlabs/feature/scheduler0.1.2

Feature/scheduler

12 files changed, 990 insertions, 232 deletions

diff --git a/docs/protocol.rst b/docs/protocol.rst
index e468b06..f7089ad 100644
--- a/docs/protocol.rst
+++ b/docs/protocol.rst
@@ -86,10 +86,53 @@ FRAME  Value          Description
 2      PUBLISH        command
 3      _MSGID_        A unique id for the msg
 4      _TOPIC_NAME_   the name of the queue the worker belongs to
-5      _HEADERS_      dictionary of headers. can be an empty set
+5      _HEADERS_      csv list of headers
+6      _MSG_          The message to send
+====== ============== ===========
+
+A **SCHEDULE** command consists of a 7-frame multipart message, formatted as follows.
+
+====== ============== ===========
+FRAME   Value         Description
+====== ============== ===========
+0      _EMPTY_        leave empty
+1      eMQP/1.0       Protocol version
+2      SCHEDULE       command
+3      _MSGID_        A unique id for the msg
+4      _TOPIC_NAME_   name of queue that the job should run in
+5      _HEADERS_      csv list of headers for this message
+6      _MSG_          The message to send
+====== ============== ===========
+
+An **UNSCHEDULE** command consists of a 7-frame multipart message, formatted as follows.
+
+====== ============== ===========
+FRAME   Value         Description
+====== ============== ===========
+0      _EMPTY_        leave empty
+1      eMQP/1.0       Protocol version
+2      UNSCHEDULE     command
+3      _MSGID_        A unique id for the msg
+4      _TOPIC_NAME_   ignored for this command, broadcasted to all queues
+5      _HEADERS_      csv list of headers for this message
 6      _MSG_          The message to send
 ====== ============== ===========
 
+eMQP / Scheduler
+----------------
+An **INFORM** command consists of a 6-frame multipart message, formatted as follows.
+
+====== ============== ===========
+FRAME   Value         Description
+====== ============== ===========
+0      _EMPTY_        leave empty
+1      eMQP/1.0       Protocol version
+2      INFORM         command
+3      _MSGID_        A unique id for the msg
+4      _QUEUE_NAME_   csv seperated names of queue the worker belongs to
+5      scheduler      type of peer connecting
+====== ============== ===========
+
 eMQP / Worker
 -------------
 An **INFORM** command consists of a 5-frame multipart message, formatted as follows.
@@ -102,6 +145,7 @@ FRAME   Value         Description
 2      INFORM         command
 3      _MSGID_        A unique id for the msg
 4      _QUEUE_NAME_   csv seperated names of queue the worker belongs to
+5      worker         type of peer connecting
 ====== ============== ===========
 
 A **READY** frame consists of a 4-frame multipart message, formatted as follows.
@@ -154,11 +198,14 @@ Heartbeating
 ------------
  * HEARTBEAT commands are valid at any time after an INFORM command
  * Any command except DISCONNECT act as a heartbeat. Peers SHOULD NOT send HEARTBEAT commands while sending other commands.
- * Both worker and broker MUST send heartbeats at regular and agreed-upon intervals.
+ * Worker and broker MUST send heartbeats at regular and agreed-upon intervals.
+ * Scheduler and broker MUST send heartbeats at regular and agreed-upon intervals.
  * If the worker detects that the broker disconnected it SHOULD restart the conversation.
  * If the broker detects that a worker has disconnected it should stop sending it a message of any type.
+ * If the scheduler detects that the broker disconnects it SHOULD restart the conversation.
+ * If the broker detects that a scheduler has disconnected it should ??????????.
 
-Request Headers
+REQUEST Headers
 ---------------
 Headers MUST be 0 to many comma seperated values inserted into the header field. If there are no headers requried, send an empty string MUST be sent where headers are required.
 
diff --git a/eventmq/client/messages.py b/eventmq/client/messages.py
index 66af8a9..03732f9 100644
--- a/eventmq/client/messages.py
+++ b/eventmq/client/messages.py
@@ -26,6 +26,67 @@ from ..utils.messages import send_emqp_message
 logger = logging.getLogger(__name__)
 
 
+def schedule(socket, func, interval_secs, args=(), kwargs=None, class_args=(),
+             class_kwargs=None, headers=('guarantee',),
+             queue=conf.DEFAULT_QUEUE_NAME, unschedule=False):
+    """
+    Execute a task on a defined interval.
+
+    Args:
+        socket (socket): eventmq socket to use for sending the message
+        func (callable): the callable to be scheduled on a worker
+        minutes (int): minutes to wait in between executions
+        args (list): list of *args to pass to the callable
+        kwargs (dict): dict of **kwargs to pass to the callable
+        class_args (list): list of *args to pass to the class (if applicable)
+        class_kwargs (dict): dict of **kwargs to pass to the class (if
+            applicable)
+        headers (list): list of strings denoting enabled headers. Default:
+            guarantee is enabled to ensure the scheduler schedules the job.
+        queue (str): name of the queue to use when executing the job. The
+            default value is the default queue.
+    """
+    if not class_kwargs:
+        class_kwargs = {}
+    if not kwargs:
+        kwargs = {}
+
+    if callable(func):
+        path, callable_name = build_module_path(func)
+    else:
+        logger.error('Encountered non-callable func: {}'.format(func))
+        return False
+
+    if not callable_name:
+        logger.error('Encountered callable with no name in {}'.format(
+            func.__module__
+        ))
+        return False
+
+    if not path:
+        logger.error('Encountered callable with no __module__ path {}'.format(
+            func.__name__
+        ))
+        return False
+
+    # TODO: convert all the times to seconds for the clock
+
+    # TODO: send the schedule request
+
+    msg = ['run', {
+        'callable': callable_name,
+        'path': path,
+        'args': args,
+        'kwargs': kwargs,
+        'class_args': class_args,
+        'class_kwargs': class_kwargs,
+    }]
+
+    send_schedule_request(socket, interval_secs=interval_secs,
+                          message=msg, headers=headers, queue=queue,
+                          unschedule=unschedule)
+
+
 def defer_job(socket, func, args=(), kwargs=None, class_args=(),
               class_kwargs=None, reply_requested=False, guarantee=False,
               retry_count=0, queue=conf.DEFAULT_QUEUE_NAME):
@@ -94,7 +155,7 @@ def defer_job(socket, func, args=(), kwargs=None, class_args=(),
     send_request(socket, msg, reply_requested=reply_requested,
                  guarantee=guarantee, retry_count=retry_count, queue=queue)
 
-    return True
+    return True  # The message has successfully been queued for delivery
 
 
 def build_module_path(func):
@@ -196,6 +257,34 @@ def send_request(socket, message, reply_requested=False, guarantee=False,
                       )
 
 
+def send_schedule_request(socket, interval_secs, message, headers=(),
+                          queue=None, unschedule=False):
+    """
+    Send a SCHEDULE or UNSCHEDULE command.
+
+    Queues a message requesting that something happens on an
+    interval for the scheduler.
+
+    Args:
+        socket (socket):
+        interval_secs (int):
+        message: Message to send socket.
+        headers (list): List of headers for the message
+        queue (str): name of queue the job should be executed in
+    """
+
+    if unschedule:
+        command = 'UNSCHEDULE'
+    else:
+        command = 'SCHEDULE'
+
+    send_emqp_message(socket, command,
+                      (queue or conf.DEFAULT_QUEUE_NAME,
+                       ','.join(headers),
+                       str(interval_secs),
+                       serialize(message)))
+
+
 def job(block=False):  # Move to decorators.py
     """
     run the decorated function on a worker
diff --git a/eventmq/constants.py b/eventmq/constants.py
index 621a6a5..c379ef5 100644
--- a/eventmq/constants.py
+++ b/eventmq/constants.py
@@ -8,5 +8,11 @@ class STATUS(object):
     stopping = 300
     stopped = 301
 
+
+class CLIENT_TYPE(object):
+    worker = 'worker'
+    scheduler = 'scheduler'
+
+
 # See doc/protocol.rst
 PROTOCOL_VERSION = 'eMQP/1.0'
diff --git a/eventmq/jobmanager.py b/eventmq/jobmanager.py
index b52eebd..9499146 100644
--- a/eventmq/jobmanager.py
+++ b/eventmq/jobmanager.py
@@ -21,14 +21,13 @@ import json
 import logging
 import signal
 
-from . import conf, constants, exceptions, utils
+from . import conf
 from .poller import Poller, POLLIN
 from .sender import Sender
-from .utils.classes import HeartbeatMixin
+from .utils.classes import EMQPService, HeartbeatMixin
 from .utils.settings import import_settings
 from .utils.devices import generate_device_name
 from .utils.messages import send_emqp_message as sendmsg
-import utils.messages
 from .utils.timeutils import monotonic
 from .worker import MultiprocessWorker as Worker
 from eventmq.log import setup_logger
@@ -36,13 +35,14 @@ from eventmq.log import setup_logger
 logger = logging.getLogger(__name__)
 
 
-class JobManager(HeartbeatMixin):
+class JobManager(HeartbeatMixin, EMQPService):
     """
     The exposed portion of the worker. The job manager's main responsibility is
     to manage the resources on the server it's running.
 
     This job manager uses tornado's eventloop.
     """
+    SERVICE_TYPE = 'worker'
 
     def __init__(self, *args, **kwargs):
         """
@@ -58,7 +58,7 @@ class JobManager(HeartbeatMixin):
         #: Define the name of this JobManager instance. Useful to know when
         #: referring to the logs.
         self.name = kwargs.pop('name', generate_device_name())
-        logger.info('Initializing JobManager %s...'.format(self.name))
+        logger.info('Initializing JobManager {}...'.format(self.name))
 
         #: Number of workers that are available to have a job executed. This
         #: number changes as workers become busy with jobs
@@ -67,7 +67,8 @@ class JobManager(HeartbeatMixin):
         #: JobManager starts out by INFORMing the router of it's existance,
         #: then telling the router that it is READY. The reply will be the unit
         #: of work.
-        self.incoming = Sender()
+        # Despite the name, jobs are received on this socket
+        self.outgoing = Sender(name=self.name)
 
         #: Jobs that are running should be stored in `active_jobs`. There
         #: should always be at most `available_workers` count of active jobs.
@@ -78,109 +79,39 @@ class JobManager(HeartbeatMixin):
 
         self._setup()
 
-    def _setup(self):
+    def _start_event_loop(self):
         """
-        Prepares JobManager ready to connect to a broker. Actions that must
-        also run on a reset are here.
+        Starts the actual eventloop. Usually called by :meth:`start`
         """
-        # Look for incoming events
-        self.poller.register(self.incoming, POLLIN)
-        self.awaiting_startup_ack = False
-
-        self.status = constants.STATUS.ready
+        # Acknowledgment has come
+        # Send a READY for each available worker
+        for i in range(0, self.available_workers):
+            self.send_ready()
 
-    def start(self, addr='tcp://127.0.0.1:47291'):
-        """
-        Connect to `addr` and begin listening for job requests
-
-        Args:
-            addr (str): connection string to connect to
-        """
-        while True:
-            self.status = constants.STATUS.connecting
-            self.incoming.connect(addr)
-
-            self.awaiting_startup_ack = True
-            self.send_inform()
-
-            # We don't want to accidentally start processing jobs before our
-            # connection has been setup completely and acknowledged.
-            while self.awaiting_startup_ack:
-                # Poller timeout is in ms so the reconnect timeout is
-                # multiplied by 1000 to get seconds
-                events = self.poller.poll(conf.RECONNECT_TIMEOUT * 1000)
-
-                if self.incoming in events:  # A message from the Router!
-                    msg = self.incoming.recv_multipart()
-                    # TODO This will silently drop messages that aren't ACK
-                    if msg[2] == "ACK":
-                        # :meth:`on_ack` will set self.awaiting_startup_ack to
-                        # False
-                        self.process_message(msg)
-
-            # Acknowledgment has come
-            # Send a READY for each available worker
-            for i in range(0, self.available_workers):
-                self.send_ready()
-
-            self.status = constants.STATUS.connected
-            logger.info('Starting to listen for jobs')
 
             # handle any sighups by reloading config
             signal.signal(signal.SIGHUP, self.sighup_handler)
 
-            self._start_event_loop()
-            # When we return, soemthing has gone wrong and we should try to
-            # reconnect
-            self.reset()
-
-    def reset(self):
-        """
-        Resets the current connection by closing and reopening the socket
-        """
-        # Unregister the old socket from the poller
-        self.poller.unregister(self.incoming)
-
-        # Polish up a new socket to use
-        self.incoming.rebuild()
-
-        # Prepare the device to connect again
-        self._setup()
-
-    def _start_event_loop(self):
-        """
-        Starts the actual eventloop. Usually called by :meth:`JobManager.start`
-        """
         while True:
+            if self.received_disconnect:
+                # self.reset()
+                # Shut down if there are no active jobs waiting
+                if len(self.active_jobs) > 0:
+                    self.prune_active_jobs()
+                    continue
+                break
+
             now = monotonic()
             events = self.poller.poll()
 
-            if events.get(self.incoming) == POLLIN:
-                msg = self.incoming.recv_multipart()
+            if events.get(self.outgoing) == POLLIN:
+                msg = self.outgoing.recv_multipart()
                 self.process_message(msg)
 
-            # Maintain the list of active jobs
-            for job in self.active_jobs:
-                if not job.is_alive():
-                    self.active_jobs.remove(job)
-                    self.available_workers += 1
-                    self.send_ready()
+            self.prune_active_jobs()
 
-            # TODO: Optimization: Move the method calls into another thread so
-            # they don't block the event loop
-            if not conf.DISABLE_HEARTBEATS:
-                # Send a HEARTBEAT if necessary
-                if now - self._meta['last_sent_heartbeat'] >= \
-                   conf.HEARTBEAT_INTERVAL:
-                    self.send_heartbeat(self.incoming)
-
-                # Do something about any missed HEARTBEAT, if we have nothing
-                # waiting on the socket
-                if self.is_dead() and not events:
-                    logger.critical(
-                        'The broker appears to have gone away. '
-                        'Reconnecting...')
-                    break
+            if not self.maybe_send_heartbeat(events):
+                break
 
     def on_request(self, msgid, msg):
         """
@@ -222,58 +153,12 @@ class JobManager(HeartbeatMixin):
 
         self.available_workers -= 1
 
-    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
-
-        command = message[0]
-        msgid = message[1]
-        message = message[2]
-
-        if hasattr(self, "on_%s" % command.lower()):
-            func = getattr(self, "on_%s" % command.lower())
-            func(msgid, message)
-        else:
-            logger.warning('No handler for %s found (tried: %s)' %
-                           (command, ('on_%s' % command.lower())))
-
     def send_ready(self):
         """
         send the READY command upstream to indicate that JobManager is ready
         for another REQUEST message.
         """
-        sendmsg(self.incoming, 'READY')
-
-    def send_inform(self, queue=None):
-        """
-        Send an INFORM command
-        """
-        sendmsg(self.incoming, 'INFORM', queue or conf.DEFAULT_QUEUE_NAME)
-        self._meta['last_sent_heartbeat'] = monotonic()
-
-    def on_ack(self, msgid, ackd_msgid):
-        """
-        Sets :attr:`awaiting_ack` to False
-        """
-        # The msgid is the only frame in the message
-        ackd_msgid = ackd_msgid[0]
-        logger.info('Received ACK for router (or client) %s' % ackd_msgid)
-        self.awaiting_startup_ack = False
+        sendmsg(self.outgoing, 'READY')
 
     def on_heartbeat(self, msgid, message):
         """
@@ -282,6 +167,16 @@ class JobManager(HeartbeatMixin):
         HEARTBEAT
         """
 
+    def prune_active_jobs(self):
+        # Maintain the list of active jobs
+        for job in self.active_jobs:
+            if not job.is_alive():
+                self.active_jobs.remove(job)
+                self.available_workers += 1
+
+                if not self.received_disconnect:
+                    self.send_ready()
+
     def sighup_handler(self, signum, frame):
         logger.info('Caught signal %s' % signum)
         self.incoming.unbind(conf.FRONTEND_ADDR)
diff --git a/eventmq/router.py b/eventmq/router.py
index ac230ff..ecbe46c 100644
--- a/eventmq/router.py
+++ b/eventmq/router.py
@@ -19,12 +19,11 @@ Routes messages to workers (that are in named queues).
 """
 from copy import copy
 import logging
-import threading
 import warnings
 import signal
 
 from . import conf, exceptions, poller, receiver
-from .constants import STATUS
+from .constants import STATUS, CLIENT_TYPE
 from .utils.classes import HeartbeatMixin
 from .utils.messages import (
     send_emqp_router_message as sendmsg,
@@ -85,6 +84,25 @@ class Router(HeartbeatMixin):
         #: workers available to take the job
         self.waiting_messages = {}
 
+        #: Tracks the last time the scheduler queue was cleaned out of dead
+        #: schedulers
+        self._meta['last_scheduler_cleanup'] = 0
+
+        #: Queue for schedulers to use:
+        self.scheduler_queue = []
+
+        #: Scheduler clients. Clients are able to send SCHEDULE commands that
+        #: need to be routed to a scheduler, which will keep track of time and
+        #: run the job.
+        #: Contains dictionaries:
+        #:     self.schedulers[<scheduler_zmq_id>] = {
+        #:       'hb': <last_recv_heartbeat>,
+        #:     }
+        self.schedulers = {}
+
+        #: Set to True when the router should die.
+        self.received_disconnect = False
+
     def start(self,
               frontend_addr=conf.FRONTEND_ADDR,
               backend_addr=conf.BACKEND_ADDR):
@@ -113,6 +131,10 @@ class Router(HeartbeatMixin):
         Starts the actual eventloop. Usually called by :meth:`Router.start`
         """
         while True:
+
+            if self.received_disconnect:
+                break
+
             now = monotonic()
             events = self.poller.poll()
 
@@ -137,6 +159,22 @@ class Router(HeartbeatMixin):
                     # ones so the next one is alive
                     self.clean_up_dead_workers()
 
+                if now - self._meta['last_sent_scheduler_heartbeat'] >= \
+                   conf.HEARTBEAT_INTERVAL:
+                    self.send_schedulers_heartbeats()
+
+                if now - self._meta['last_scheduler_cleanup'] >= 10:
+                    self.clean_up_dead_schedulers()
+
+    def reset_heartbeat_counters(self):
+        """
+        Reset all the counters for heartbeats back to 0
+        """
+        super(Router, self).reset_heartbeat_counters()
+
+        # track the last time the router sent a heartbeat to the schedulers
+        self._meta['last_sent_scheduler_heartbeat'] = 0
+
     def send_ack(self, socket, recipient, msgid):
         """
         Sends an ACK response
@@ -162,13 +200,22 @@ class Router(HeartbeatMixin):
 
     def send_workers_heartbeats(self):
         """
-        Send heartbeats to all registered workers.
+        Send HEARTBEATs to all registered workers.
         """
         self._meta['last_sent_heartbeat'] = monotonic()
 
         for worker_id in self.workers:
             self.send_heartbeat(self.outgoing, worker_id)
 
+    def send_schedulers_heartbeats(self):
+        """
+        Send HEARTBEATs to all registered schedulers
+        """
+        self._meta['last_sent_scheduler_heartbeat'] = monotonic()
+
+        for scheduler_id in self.schedulers:
+            self.send_heartbeat(self.incoming, scheduler_id)
+
     def on_heartbeat(self, sender, msgid, msg):
         """
         a placeholder for a no-op command. The actual 'logic' for HEARTBEAT is
@@ -181,12 +228,22 @@ class Router(HeartbeatMixin):
         Handles an INFORM message. This happens when new worker coming online
         and announces itself.
         """
-        logger.info('Received INFORM request from %s' % sender)
         queue_name = msg[0]
+        client_type = msg[1]
+
+        logger.info('Received INFORM request from {} (type: {})'.format(
+            sender, client_type))
 
-        self.add_worker(sender, queue_name)
+        if client_type == CLIENT_TYPE.worker:
+            self.add_worker(sender, queue_name)
+            self.send_ack(self.outgoing, sender, msgid)
+        elif client_type == CLIENT_TYPE.scheduler:
+            self.add_scheduler(sender)
+            self.send_ack(self.incoming, sender, msgid)
 
-        self.send_ack(self.outgoing, sender, msgid)
+    def on_disconnect(self, msgid, msg):
+        # Loops event loops should check for this and break out
+        self.received_disconnect = True
 
     def on_ready(self, sender, msgid, msg):
         """
@@ -252,7 +309,7 @@ class Router(HeartbeatMixin):
         Adds a worker to worker queues
 
         Args:
-            worker_id: unique id of the worker to add
+            worker_id (str): unique id of the worker to add
             queues: queue or queues this worker should be a member of
         """
         # Add the worker to our worker dict
@@ -263,6 +320,36 @@ class Router(HeartbeatMixin):
         logger.debug('Adding {} to the self.workers for queues:{}'.format(
                      worker_id, str(queues)))
 
+    def clean_up_dead_schedulers(self):
+        """
+        Loops through the list of schedulers and remove any schedulers who
+        the router hasn't received a heartbeat in HEARTBEAT_TIMEOUT
+        """
+        now = monotonic()
+        self._meta['last_scheduler_cleanup'] = now
+        schedulers = copy(self.scheduler_queue)
+
+        for scheduler_id in schedulers:
+            last_hb_seconds = now - self.schedulers[scheduler_id]['hb']
+            if last_hb_seconds >= conf.HEARTBEAT_TIMEOUT:
+                logger.info("No HEARTBEAT from scheduler {} in {} Removing "
+                            "from the queue".format(scheduler_id,
+                                                    last_hb_seconds))
+                del self.schedulers[scheduler_id]
+                self.scheduler_queue.remove(scheduler_id)
+
+    def add_scheduler(self, scheduler_id):
+        """
+        Adds a scheduler to the queue to receive SCHEDULE commands
+
+        Args:
+            scheduler_id (str): unique id of the scheduler to add
+        """
+        self.scheduler_queue.append(scheduler_id)
+        self.schedulers[scheduler_id] = {}
+        self.schedulers[scheduler_id]['hb'] = monotonic()
+        logger.debug('Adding {} to self.schedulers'.format(scheduler_id))
+
     def requeue_worker(self, worker_id):
         """
         Add a worker back to the pools for which it is a member of.
@@ -295,10 +382,6 @@ class Router(HeartbeatMixin):
 
     def on_receive_request(self, msg):
         """
-        This function is called when a message comes in from the client socket.
-        It then calls `on_command`. If `on_command` isn't found, then a
-        warning is created.
-
         Args:
             msg: The untouched message from zmq
         """
@@ -307,43 +390,83 @@ class Router(HeartbeatMixin):
         except exceptions.InvalidMessageError:
             logger.exception('Invalid message from clients: %s' % str(msg))
 
-        queue_name = message[3][0]
-
-        # If we have no workers for the queue TODO something about it
-        if queue_name not in self.queues:
-            logger.warning("Received REQUEST with a queue I don't recognize: "
-                           "%s" % queue_name)
-            logger.critical("Discarding message")
-            # TODO: Don't discard the message
-            return
-
-        try:
-            worker_addr = self.queues[queue_name].pop()
-        except KeyError:
-            logger.critical("REQUEST for an unknown queue caught in exception")
-            logger.critical("Discarding message")
-            return
-        except IndexError:
-            logger.warning('No available workers for queue "%s". Buffering '
-                           'message to send later.' % queue_name)
-            if queue_name not in self.waiting_messages:
-                self.waiting_messages[queue_name] = []
-            self.waiting_messages[queue_name].append(msg)
-            logger.debug('%d waiting messages in queue "%s"' %
-                         (len(self.waiting_messages[queue_name]), queue_name))
-            return
+        sender = message[0]
+        command = message[1]
 
-        try:
-            # strip off the client id before forwarding because the worker
-            # isn't expecting it, and the zmq socket is going to put our
-            # id on it.
-            fwdmsg(self.outgoing, worker_addr, msg[1:])
-        except exceptions.PeerGoneAwayError:
-            logger.debug("Worker {} has unexpectedly gone away. Trying "
-                         "another worker".format(worker_addr))
-
-            # TODO: Rewrite this logic as a loop
-            self.on_receive_request(msg)
+        # Count this message as a heart beat if it came from a scheduler that
+        # the router is aware of.
+        if sender in self.schedulers and sender in self.scheduler_queue:
+            self.schedulers[sender]['hb'] = monotonic()
+
+            # If it is a heartbeat then there is nothing left to do
+            if command == "HEARTBEAT":
+                return
+
+        # REQUEST is the most common message so it goes at the top
+        if command == "REQUEST":
+            queue_name = message[3][0]
+            # If we have no workers for the queue TODO something about it
+            if queue_name not in self.queues:
+                logger.warning("Received %s with a queue I don't recognize: "
+                               "%s" % (msg[3], queue_name))
+                logger.critical("Discarding message")
+                # TODO: Don't discard the message
+                return
+
+            try:
+                worker_addr = self.queues[queue_name].pop()
+            except KeyError:
+                logger.critical("REQUEST for an unknown queue caught in "
+                                "exception")
+                logger.critical("Discarding message")
+                return
+            except IndexError:
+                logger.warning('No available workers for queue "%s". '
+                               'Buffering message to send later.' % queue_name)
+                if queue_name not in self.waiting_messages:
+                    self.waiting_messages[queue_name] = []
+                    self.waiting_messages[queue_name].append(msg)
+                    logger.debug('%d waiting messages in queue "%s"' %
+                                 (len(self.waiting_messages[queue_name]),
+                                  queue_name))
+                    return
+
+            try:
+                # strip off the client id before forwarding because the
+                # worker isn't expecting it, and the zmq socket is going
+                # to put this router's id on it.
+                fwdmsg(self.outgoing, worker_addr, msg[1:])
+            except exceptions.PeerGoneAwayError:
+                logger.debug("Worker {} has unexpectedly gone away. "
+                             "Trying another worker".format(worker_addr))
+
+                # TODO: Rewrite this logic as a loop, so it can't recurse
+                # into oblivion
+                self.on_receive_request(msg)
+        # elif command == "HEARTBEAT":
+        #     # The scheduler is heartbeating
+
+        elif command == "INFORM":
+            # This is a scheduler trying join
+            self.on_inform(message[0], message[2], message[3])
+
+        elif command == "SCHEDULE" or command == "UNSCHEDULE":
+            # Forward the schedule message to the schedulers
+            scheduler_addr = self.scheduler_queue.pop()
+            self.scheduler_queue.append(scheduler_addr)
+            self.schedulers[scheduler_addr] = {
+                'hb': monotonic(),
+            }
+
+            try:
+                # Strips off the client id before forwarding because the
+                # scheduler isn't expecting it.
+                fwdmsg(self.incoming, scheduler_addr, msg[1:])
+            except exceptions.PeerGoneAwayError:
+                logger.debug("Scheduler {} has unexpectedly gone away. Trying "
+                             "another scheduler.".format(scheduler_addr))
+                # TODO: rewrite this in a loop
+                self.on_receive_request(msg)
 
     def process_worker_message(self, msg):
         """
diff --git a/eventmq/scheduler.py b/eventmq/scheduler.py
index c5c5754..16f6095 100644
--- a/eventmq/scheduler.py
+++ b/eventmq/scheduler.py
@@ -17,17 +17,23 @@
 =============================
 Handles cron and other scheduled tasks
 """
+import json
 import logging
 import time
+import redis
 
 from croniter import croniter
 from six import next
 
 from . import conf
 from .sender import Sender
-from .utils.classes import HeartbeatMixin
+from .poller import Poller, POLLIN
+from .utils.classes import EMQPService, HeartbeatMixin
+from json import loads as deserialize
+from json import dumps as serialize
 from .utils.settings import import_settings
-from .utils.timeutils import seconds_until, timestamp
+from .utils.timeutils import IntervalIter
+from .utils.timeutils import seconds_until, timestamp, monotonic
 from .client.messages import send_request
 
 from eventmq.log import setup_logger
@@ -35,40 +41,65 @@ from eventmq.log import setup_logger
 logger = logging.getLogger(__name__)
 
 
-class Scheduler(HeartbeatMixin):
+class Scheduler(HeartbeatMixin, EMQPService):
     """
     Keeper of time, master of schedules
     """
+    SERVICE_TYPE = 'scheduler'
 
     def __init__(self, *args, **kwargs):
         logger.info('Initializing Scheduler...')
         super(Scheduler, self).__init__(*args, **kwargs)
         self.outgoing = Sender()
 
-        # 0 = the next ts this job should be executed
+        # Open connection to redis server for persistance
+        self.redis_server = redis.StrictRedis(host='localhost',
+                                              port=6379,
+                                              db=0)
+
+        # contains 4-item lists representing cron jobs
+        # IDX     Description
+        # 0 = the next ts this job should be executed in
         # 1 = the function to be executed
         # 2 = the croniter iterator for this job
-        self.jobs = []
+        # 3 = the queue to execute the job in
+        self.cron_jobs = []
+
+        # contains dict of 4-item lists representing jobs based on an interval
+        # key of this dictionary is a hash of company_id, path, and callable
+        # from the message of the SCHEDULE command received
+        # values of this list follow this format:
+        # IDX     Descriptions
+        # 0 = the next (monotonic) ts that this job should be executed in
+        # 1 = the function to be executed
+        # 2 = the interval iter for this job
+        # 3 = the queue to execute the job in
+        self.interval_jobs = {}
+
+        self.poller = Poller()
 
         self.load_jobs()
 
-    def connect(self, addr='tcp://127.0.0.1:47290'):
-        """
-        Connect the scheduler to worker/router at `addr`
-        """
-        self.outgoing.connect(addr)
+        self._setup()
 
     def load_jobs(self):
         """
         Loads the jobs that need to be scheduled
         """
         raw_jobs = (
-            ('* * * * *', 'eventmq.scheduler.test_job'),
+            # ('* * * * *', 'eventmq.scheduler.test_job'),
         )
         ts = int(timestamp())
         for job in raw_jobs:
             # Create the croniter iterator
             c = croniter(job[0])
+            path = '.'.join(job[1].split('.')[:-1])
+            callable_ = job.split('.')[-1]
+
+            msg = ['run', {
+                'path': path,
+                'callable': callable_
+            }]
 
             # Get the next time this job should be run
             c_next = next(c)
@@ -76,15 +107,24 @@ class Scheduler(HeartbeatMixin):
                 # If the next execution time has passed move the iterator to
                 # the following time
                 c_next = next(c)
-            self.jobs.append([c_next, job[1], c])
-
-    def start(self, addr='tcp://127.0.0.1:47290'):
-        """
-        Begin sending messages to execute scheduled jobs
-        """
-        self.connect(addr)
+            self.cron_jobs.append([c_next, msg, c, None])
 
-        self._start_event_loop()
+        # Restore persisted data if redis connection is alive and has jobs
+        if (self.redis_server):
+            interval_job_list = self.redis_server.lrange('interval_jobs',
+                                                         0,
+                                                         -1)
+            if interval_job_list is not None:
+                for i in interval_job_list:
+                    logger.debug('Restoring job with hash %s' % i)
+                    if (self.redis_server.get(i)):
+                        self.load_job_from_redis(
+                            message=deserialize(self.redis_server.get(i)))
+                    else:
+                        logger.warning('Expected scheduled job in redis,' +
+                                       'but none was found with hash %s' % i)
+            else:
+                logger.warning('Unabled to talk to redis server')
 
     def _start_event_loop(self):
         """
@@ -92,29 +132,146 @@ class Scheduler(HeartbeatMixin):
         """
         while True:
             ts_now = int(timestamp())
+            m_now = monotonic()
+            events = self.poller.poll()
+
+            if events.get(self.outgoing) == POLLIN:
+                msg = self.outgoing.recv_multipart()
+                self.process_message(msg)
 
-            for i in range(0, len(self.jobs)):
-                if self.jobs[i][0] <= ts_now:  # If the time is now, or passed
-                    job = self.jobs[i][1]
-                    path = '.'.join(job.split('.')[:-1])
-                    callable_ = job.split('.')[-1]
+            # TODO: distribute me!
+            for i in range(0, len(self.cron_jobs)):
+                # If the time is now, or passed
+                if self.cron_jobs[i][0] <= ts_now:
+                    msg = self.cron_jobs[i][1]
+                    queue = self.cron_jobs[i][3]
 
-                    # Run the job
+                    # Run the msg
                     logger.debug("Time is: %s; Schedule is: %s - Running %s"
-                                 % (ts_now, self.jobs[i][0], job))
+                                 % (ts_now, self.cron_jobs[i][0], msg))
 
-                    msg = ['run', {
-                        'path': path,
-                        'callable': callable_
-                    }]
-                    send_request(self.outgoing, msg)
+                    self.send_request(self.outgoing, msg, queue=queue)
 
                     # Update the next time to run
-                    self.jobs[i][0] = next(self.jobs[i][2])
+                    self.cron_jobs[i][0] = next(self.cron_jobs[i][2])
                     logger.debug("Next execution will be in %ss" %
-                                 seconds_until(self.jobs[i][0]))
+                                 seconds_until(self.cron_jobs[i][0]))
 
-            time.sleep(0.1)
+            for k, v in self.interval_jobs.iteritems():
+                if v[0] <= m_now:
+                    msg = v[1]
+                    queue = v[3]
+
+                    logger.debug("Time is: %s; Schedule is: %s - Running %s"
+                                 % (ts_now, v[0], msg))
+
+                    self.send_request(msg, queue=queue)
+                    v[0] = next(v[2])
+
+            if not self.maybe_send_heartbeat(events):
+                break
+
+    def send_request(self, jobmsg, queue=None):
+        jobmsg = json.loads(jobmsg)
+        send_request(self.outgoing, jobmsg, queue=queue)
+
+    def on_unschedule(self, msgid, message):
+        """
+           Unschedule an existing schedule job, if it exists
+        """
+        logger.info("Received new UNSCHEDULE request: {}".format(message))
+
+        schedule_hash = self.schedule_hash(message)
+
+        if schedule_hash in self.interval_jobs:
+            # Remove scheduled job
+            self.interval_jobs.pop(schedule_hash)
+        else:
+            logger.debug("Couldn't find matching schedule for unschedule " +
+                         "request")
+
+        # Double check the redis server even if we didn't find the hash
+        # in memory
+        if (self.redis_server):
+            if (self.redis_server.get(schedule_hash)):
+                self.redis_server.lrem('interval_jobs', 0, schedule_hash)
+                self.redis_server.save()
+
+    def load_job_from_redis(self, message):
+        """
+        """
+        from .utils.timeutils import IntervalIter
+
+        queue = message[0].encode('utf-8')
+        interval = int(message[2])
+        inter_iter = IntervalIter(monotonic(), interval)
+        schedule_hash = self.schedule_hash(message)
+
+        self.interval_jobs[schedule_hash] = [
+            next(inter_iter),
+            message[3],
+            inter_iter,
+            queue
+        ]
+
+    def on_schedule(self, msgid, message):
+        """
+        """
+        logger.info("Received new SCHEDULE request: {}".format(message))
+
+        queue = message[0]
+        interval = int(message[2])
+        inter_iter = IntervalIter(monotonic(), interval)
+        schedule_hash = self.schedule_hash(message)
+
+        # Notify if this is updating existing, or new
+        if (schedule_hash in self.interval_jobs):
+            logger.debug('Update existing scheduled job with %s'
+                         % schedule_hash)
+        else:
+            logger.debug('Creating a new scheduled job with %s'
+                         % schedule_hash)
+
+        self.interval_jobs[schedule_hash] = [
+            next(inter_iter),
+            message[3],
+            inter_iter,
+            queue
+        ]
+
+        # Persist the scheduled job
+        if (self.redis_server):
+            if schedule_hash not in self.redis_server.lrange('interval_jobs',
+                                                             0,
+                                                             -1):
+                self.redis_server.lpush('interval_jobs', schedule_hash)
+            self.redis_server.set(schedule_hash, serialize(message))
+            self.redis_server.save()
+
+        self.send_request(message[3], queue=queue)
+
+    def on_heartbeat(self, msgid, message):
+        """
+        Noop command. The logic for heartbeating is in the event loop.
+        """
+
+    def schedule_hash(self, message):
+        """
+        Create a unique identifier for this message for storing
+        and referencing later
+        """
+        # Items to use for uniquely identifying this scheduled job
+        # TODO: Pass company_id in a more rigid place
+        msg = deserialize(message[3])[1]
+        schedule_hash_items = {'company_id': msg['class_args'][0],
+                               'path': msg['path'],
+                               'callable': msg['callable']}
+
+        # Hash the sorted, immutable set of items in our identifying dict
+        schedule_hash = str(hash(tuple(frozenset(sorted(
+            schedule_hash_items.items())))))
+
+        return schedule_hash
 
     def scheduler_main(self):
         """
@@ -122,6 +279,7 @@ class Scheduler(HeartbeatMixin):
         """
         setup_logger("eventmq")
         import_settings()
+        self.__init__()
         self.start(addr=conf.SCHEDULER_ADDR)
 
 
@@ -134,6 +292,4 @@ def scheduler_main():
 def test_job():
     print "hello!"
     print "hello!"
-    print "hello!"
-    print "hello!"
     time.sleep(4)
diff --git a/eventmq/tests/test_jobmanager.py b/eventmq/tests/test_jobmanager.py
new file mode 100644
index 0000000..3c3941d
--- /dev/null
+++ b/eventmq/tests/test_jobmanager.py
@@ -0,0 +1,148 @@
+# This file is part of eventmq.
+#
+# eventmq is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# eventmq is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with eventmq.  If not, see <http://www.gnu.org/licenses/>.
+import json
+import threading
+import time
+import unittest
+
+import mock
+import zmq
+
+from .. import conf, constants, jobmanager
+from ..utils.classes import ZMQSendMixin, ZMQReceiveMixin
+from ..utils.messages import send_emqp_router_message
+
+ADDR = 'inproc://pour_the_rice_in_the_thing'
+
+
+class FakeDevice(ZMQReceiveMixin, ZMQSendMixin):
+    """
+    A fake router device so we can test with some of the nice utilities, but
+    still allowing manual control
+    """
+    def __init__(self, addr=ADDR):
+        super(FakeDevice, self).__init__()
+
+        self.zsocket = zmq.Context.instance().socket(zmq.ROUTER)
+
+
+class TestCase(unittest.TestCase):
+    def setUp(self):
+        self.jm = jobmanager.JobManager()
+
+        # Since JobManager runs as a process a thread is used to allow the loop
+        # to run
+        self.jm_thread = threading.Thread(target=start_jm,
+                                          args=(self.jm,))
+
+        self.addCleanup(self.cleanup)
+
+    def test__setup(self):
+        jm = jobmanager.JobManager(name='RuckusBringer')
+        self.assertEqual(jm.name, 'RuckusBringer')
+
+        self.assertFalse(jm.awaiting_startup_ack)
+        self.assertEqual(jm.status, constants.STATUS.ready)
+
+# EMQP Tests
+    def test_reset(self):
+        self.jm.reset()
+
+        self.assertFalse(self.jm.awaiting_startup_ack)
+        self.assertEqual(self.jm.status, constants.STATUS.ready)
+
+    @mock.patch('signal.signal')
+    def test_start(self, mock_signal_signal):
+        sock = FakeDevice()
+
+        self.jm_thread.start()
+        time.sleep(.1)  # wait for the manager to warm up
+
+        self.assertTrue(self.jm.awaiting_startup_ack)
+        self.assertEqual(self.jm.status, constants.STATUS.connecting)
+
+        # Give JM something to connect to.
+        sock.zsocket.bind(ADDR)
+
+        jm_addr, _, _, cmd, msgid, queues, type_ = sock.recv_multipart()
+        self.assertEqual(self.jm.name, jm_addr)
+        self.assertEqual(cmd, "INFORM")
+        self.assertEqual(type_, constants.CLIENT_TYPE.worker)
+
+        self.send_ack(sock, jm_addr, msgid)
+
+        time.sleep(.1)
+        self.assertEqual(self.jm.status, constants.STATUS.connected)
+
+    def send_ack(self, sock, jm_addr, msgid):
+        send_emqp_router_message(sock, jm_addr, "ACK", msgid)
+
+    @mock.patch('signal.signal')
+    def test__start_event_loop(self, mock_signal_signal):
+        # Tests the first part of the event loop
+        sock = FakeDevice()
+        sock.zsocket.bind(ADDR)
+
+        self.jm_thread.start()
+
+        # Consume the INFORM command
+        jm_addr, _, _, cmd, msgid, queues, type_ = sock.recv_multipart()
+        self.send_ack(sock, jm_addr, msgid)
+
+        # Test the correct number of READY messages is sent for the broker
+        # to know how many jobs the JM can handle
+        ready_msg_count = 0
+        for i in range(0, self.jm.available_workers):
+            msg = sock.recv_multipart()
+            if len(msg) > 4 and msg[3] == "READY":
+                ready_msg_count += 1
+        # If this fails, less READY messages were sent than were supposed
+        # to be sent.
+        self.assertEqual(ready_msg_count, self.jm.available_workers)
+
+    @mock.patch('signal.signal')
+    def test_on_request(self, mock_signal_signal):
+        from ..client.messages import build_module_path
+        sock = FakeDevice()
+        sock.zsocket.bind(ADDR)
+        self.jm_thread.start()
+
+        jm_addr, _, _, _, msgid, _, _ = sock.recv_multipart()
+        self.send_ack(sock, jm_addr, msgid)
+        time.sleep(.1)  # give time for the JM to process
+
+        path, callable_name = build_module_path(pretend_job)
+
+        run_msg = ['run', {
+            'path': path,
+            'callable': callable_name,
+        }]
+
+        msg = (conf.DEFAULT_QUEUE_NAME, '', json.dumps(run_msg))
+
+        send_emqp_router_message(sock, jm_addr, 'REQUEST', msg)
+        time.sleep(.1)  # give time for the job to start up.
+        self.assertEqual(len(self.jm.active_jobs), 1)
+
+    def cleanup(self):
+        self.jm.on_disconnect(None, None)
+
+
+def start_jm(jm):
+    jm.start(ADDR)
+
+
+def pretend_job():
+    time.sleep(1)
diff --git a/eventmq/tests/test_router.py b/eventmq/tests/test_router.py
new file mode 100644
index 0000000..11ca9ec
--- /dev/null
+++ b/eventmq/tests/test_router.py
@@ -0,0 +1,42 @@
+# This file is part of eventmq.
+#
+# eventmq is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# eventmq is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with eventmq.  If not, see <http://www.gnu.org/licenses/>.
+import threading
+import unittest
+
+import mock
+
+from .. import router
+
+ADDR = 'inproc://kodak_film_festival'
+
+
+class TestCase(unittest.TestCase):
+    def setUp(self):
+        self.router = router.Router()
+
+        self.thread = threading.Thread(target=start_router,
+                                       args=(self.router,))
+
+        self.addCleanup(self.cleanup)
+
+    @mock.patch('signal.signal')
+    def test_start(self, mock_signal_signal):
+        self.thread.start()
+
+    def cleanup(self):
+        self.router.on_disconnect(None, None)
+
+def start_router(router):
+    router.start(ADDR)
diff --git a/eventmq/tests/test_utils.py b/eventmq/tests/test_utils.py
index 0fc8042..7ed36d7 100644
--- a/eventmq/tests/test_utils.py
+++ b/eventmq/tests/test_utils.py
@@ -52,3 +52,6 @@ class TestCase(unittest.TestCase):
         broken_message = ('dlkajfs', 'lkasdjf')
         with self.assertRaises(exceptions.InvalidMessageError):
             messages.parse_router_message(broken_message)
+
+    def test_parse_router_message(self):
+        ['aef451a0-5cef-4f03-818a-221061c8ab68', '', 'eMQP/1.0', 'INFORM', '5caeb5fd-15d4-4b08-89e8-4e536672eef3', 'default', 'worker']
diff --git a/eventmq/utils/classes.py b/eventmq/utils/classes.py
index 3459c65..cfd06d5 100644
--- a/eventmq/utils/classes.py
+++ b/eventmq/utils/classes.py
@@ -21,13 +21,198 @@ import logging
 
 import zmq.error
 
-from .. import conf, exceptions
+from .. import conf, constants, exceptions, poller, utils
 from ..utils.messages import send_emqp_message as sendmsg
 from ..utils.timeutils import monotonic, timestamp
 
 logger = logging.getLogger(__name__)
 
 
+class EMQPService(object):
+    """
+    Helper for devices that connect to brokers.
+
+    Implements utility methods for sending EMQP messages for the following
+    EMQP commands.
+     - INFORM
+
+    Also implements utlitiy methods for managing long-running processes.
+
+    To use you must define:
+     - `self.outgoing` - socket where messages can be sent to the Router
+     - `self.SERVICE_TYPE` - defines the service type for INFORM. See
+       :meth:`send_inform` for more information.
+     - `self.poller` - the poller that `self.outgoing` will be using.
+       Usually: `self.poller = eventmq.poller.Poller()`
+
+    When messages are received from the router, they are processed in
+    :meth:`process_message` which then calls `on_COMMAND`. This should be used
+    in the event loop so if you want to respond to the SCHEDULE command, you
+    would define the method `on_schedule` in your service class.
+
+    See the code for :class:`Scheduler` and :class:`JobManager` for examples.
+    """
+    def send_inform(self, queue=None):
+        """
+        Queues an INFORM command to `self.outgoing`.
+
+        Args:
+            type_ (str): Either 'worker' or 'scheduler'
+            queue (list):
+                - For 'worker' type, the queues the worker is listening on
+                - Ignored for 'scheduler' type
+
+        Raises:
+            ValueError: When `type_` does not match a specified type
+        """
+        valid_types = (constants.CLIENT_TYPE.worker,
+                       constants.CLIENT_TYPE.scheduler)
+
+        if self.SERVICE_TYPE not in valid_types:
+            raise ValueError('{} not one of {}'.format(self.SERVICE_TYPE,
+                                                       valid_types))
+
+        sendmsg(self.outgoing, 'INFORM', [
+            queue or conf.DEFAULT_QUEUE_NAME,
+            self.SERVICE_TYPE
+        ])
+
+        # If heartbeating is active, update the last heartbeat time
+        if hasattr(self, '_meta') and 'last_sent_heartbeat' in self._meta:
+            self._meta['last_sent_heartbeat'] = monotonic()
+
+    def _setup(self):
+        """
+        Prepares the service to connect to a broker. Actions that must
+        also run on a reset are here.
+        """
+        # Look for incoming events
+        self.poller.register(self.outgoing, poller.POLLIN)
+        self.awaiting_startup_ack = False
+        self.received_disconnect = False
+
+        self.status = constants.STATUS.ready
+
+    def start(self, addr, queues=conf.DEFAULT_QUEUE_NAME):
+        """
+        Connect to `addr` and begin listening for job requests
+
+        Args:
+            addr (str): connection string to connect to
+        """
+        while not self.received_disconnect:
+            self.status = constants.STATUS.connecting
+            self.outgoing.connect(addr)
+
+            # Setting this to false is how the loop is broken and the
+            # _event_loop is started.
+            self.awaiting_startup_ack = True
+
+            # If this is inside the loop, then many inform messages will stack
+            # up on the buffer until something is actually connected to.
+            self.send_inform(queues)
+
+            # We don't want to accidentally start processing jobs before our
+            # connection has been setup completely and acknowledged.
+            while self.awaiting_startup_ack:
+                # Poller timeout is in ms so the reconnect timeout is
+                # multiplied by 1000 to get seconds
+                events = self.poller.poll(conf.RECONNECT_TIMEOUT * 1000)
+
+                if self.outgoing in events:  # A message from the Router!
+                    msg = self.outgoing.recv_multipart()
+                    # TODO This will silently drop messages that aren't
+                    # ACK/DISCONNECT
+                    if msg[2] == "ACK" or msg[2] == "DISCONNECT":
+                        # :meth:`on_ack` will set self.awaiting_startup_ack to
+                        # False
+                        self.process_message(msg)
+
+            self.status = constants.STATUS.connected
+            logger.info('Starting event loop...')
+            self._start_event_loop()
+            # When we return, soemthing has gone wrong and try to reconnect
+            # unless self.received_disconnect is True
+            if not self.received_disconnect:
+                self.reset()
+
+        logger.info('Death.')
+
+    def reset(self):
+        """
+        Resets the current connection by closing and reopening the socket
+        """
+        # Unregister the old socket from the poller
+        self.poller.unregister(self.outgoing)
+
+        # Polish up a new socket to use
+        self.outgoing.rebuild()
+
+        # Prepare the device to connect again
+        self._setup()
+
+    def process_message(self, msg):
+        """
+        Processes a message. Processing takes form of calling an
+        `on_EMQP_COMMAND` method. The method must accept `msgid` and `message`
+        as the first arguments.
+
+        Args:
+            msg: The message received from the socket to parse and process.
+        """
+        if self.is_heartbeat_enabled:
+            # Any received message should count as a heartbeat
+            self._meta['last_received_heartbeat'] = monotonic()
+            if self._meta['heartbeat_miss_count']:
+                # Reset the miss count too
+                self._meta['heartbeat_miss_count'] = 0
+
+        try:
+            message = utils.messages.parse_message(msg)
+        except exceptions.InvalidMessageError:
+            logger.exception('Invalid message: %s' % str(msg))
+            return
+
+        command = message[0].lower()
+        msgid = message[1]
+        message = message[2]
+
+        if hasattr(self, "on_%s" % command):
+            func = getattr(self, "on_%s" % command)
+            func(msgid, message)
+        else:
+            logger.warning('No handler for %s found (tried: %s)' %
+                           (command.upper(), ('on_%s' % command)))
+
+    def on_ack(self, msgid, ackd_msgid):
+        """
+        Sets :attr:`awaiting_ack` to False
+        """
+        # The msgid is the only frame in the message
+        ackd_msgid = ackd_msgid[0]
+        logger.info('Received ACK for router (or client) %s' % ackd_msgid)
+        self.awaiting_startup_ack = False
+
+    def on_disconnect(self, msgid, msg):
+        # To break out of the connecting loop if necessary
+        self.awaiting_startup_ack = False
+
+        # Loops event loops should check for this and break out
+        self.received_disconnect = True
+
+    @property
+    def is_heartbeat_enabled(self):
+        """
+        Property to check if heartbeating is enabled. Useful when certain
+        properties must be updated for heartbeating
+        Returns:
+            bool - True if heartbeating is enabled, False if it isn't
+        """
+        if hasattr(self, '_meta') and 'last_sent_heartbeat' in self._meta:
+            return True
+        return False
+
+
 class HeartbeatMixin(object):
     """
     Provides methods for implementing heartbeats
@@ -36,6 +221,7 @@ class HeartbeatMixin(object):
         """
         Sets up some variables to track the state of heartbeaty things
         """
+        super(HeartbeatMixin, self).__init__()
         if not hasattr(self, '_meta'):
             self._meta = {}
 
@@ -91,6 +277,25 @@ class HeartbeatMixin(object):
 
         return False
 
+    def maybe_send_heartbeat(self, events):
+        # TODO: Optimization: Move the method calls into another thread so
+        # they don't block the event loop
+        if not conf.DISABLE_HEARTBEATS:
+            now = monotonic()
+            # Send a HEARTBEAT if necessary
+            if now - self._meta['last_sent_heartbeat'] >= \
+               conf.HEARTBEAT_INTERVAL:
+                self.send_heartbeat(self.outgoing)
+
+            # Do something about any missed HEARTBEAT, if we have nothing
+            # waiting on the socket
+            if self.is_dead() and not events:
+                logger.critical(
+                    'The broker appears to have gone away. '
+                    'Reconnecting...')
+                return False
+            return True
+
 
 class ZMQReceiveMixin(object):
     """
@@ -162,7 +367,7 @@ class ZMQSendMixin(object):
         if conf.SUPER_DEBUG:
             # If it's not at least 4 frames long then most likely it isn't an
             # eventmq message
-            if len(msg) == 4 and \
+            if len(msg) > 4 and \
                not ("HEARTBEAT" == msg[2] or "HEARTBEAT" == msg[3]) or \
                not conf.HIDE_HEARTBEAT_LOGS:
                 logger.debug('Sending message: %s' % str(msg))
diff --git a/eventmq/utils/timeutils.py b/eventmq/utils/timeutils.py
index 9d494eb..70fd6ce 100644
--- a/eventmq/utils/timeutils.py
+++ b/eventmq/utils/timeutils.py
@@ -44,3 +44,45 @@ def seconds_until(ts):
     time.time()
     """
     return ts - timestamp()
+
+
+class IntervalIter(object):
+    """
+    represents an interval (in seconds) and it's `next()` execution time
+
+    Usage:
+        # interval of 5min using monotonic clock (assume it starts at 0 for the
+        # sake of the example)
+        interval = IntervalIter(monotonic, 300)
+        # Py2
+
+        interval.next()  # 300
+        interval.next()  # 600
+
+        # Py3
+        next(interval)  # 300
+        next(interval)  # 600
+    """
+    def __init__(self, start_value, interval_secs):
+        """
+        Args:
+            start_value (numeric) - the timestamp to begin with. usually gotten
+                via :func:`monotonic` or :func:`timestamp`
+            interval_secs (int) - the number of seconds between intervals
+        """
+        self.current = start_value
+        self.interval_secs = interval_secs
+
+        # iterate the first time so the first call to .next() is interval_secs
+        # + start_value
+        self.__next__()
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):  # Py3
+        self.current += self.interval_secs
+        return self.current - self.interval_secs
+
+    def next(self):
+        return self.__next__()
diff --git a/requirements.txt b/requirements.txt
index ada0351..8872619 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,7 +1,8 @@
-pyzmq==14.6.0
+pyzmq==15.1.0
 six==1.10.0
 monotonic==0.4    # A clock who's time is not changed. used for scheduling
 croniter==0.3.10
+redis==2.10.3
 
 # Documentation
 sphinxcontrib-napoleon==0.4.3
@@ -11,3 +12,4 @@ Sphinx==1.3.1              # must come after napoleon to get the latest version
 nose==1.3.6
 coverage==4.0.3
 testfixtures==4.7.0        # To test that logging exists
+mock==1.3.0