8 files changed, 310 insertions, 141 deletions

diff --git a/bin/scheduler b/bin/scheduler
index 33f164e..6be810c 100755
--- a/bin/scheduler
+++ b/bin/scheduler
@@ -6,4 +6,4 @@ from eventmq.scheduler import Scheduler
 if __name__ == "__main__":
     setup_logger("eventmq")
     s = Scheduler()
-    s.start()
+    s.start(addr='tcp://127.0.0.1:47290')
diff --git a/bin/worker b/bin/worker
index 2dbf41b..a44f143 100755
--- a/bin/worker
+++ b/bin/worker
@@ -6,4 +6,4 @@ from eventmq.log import setup_logger
 if __name__ == "__main__":
     setup_logger('')
     j = JobManager()
-    j.start()
+    j.start(addr='tcp://127.0.0.1:47291')
diff --git a/docs/protocol.rst b/docs/protocol.rst
index e468b06..a0ab6be 100644
--- a/docs/protocol.rst
+++ b/docs/protocol.rst
@@ -90,6 +90,34 @@ FRAME  Value          Description
 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      _QUEUE_NAME_   csv seperated names of queue the worker belongs to
+5      _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 +130,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 +183,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..801e861 100644
--- a/eventmq/client/messages.py
+++ b/eventmq/client/messages.py
@@ -26,6 +26,62 @@ from ..utils.messages import send_emqp_message
 logger = logging.getLogger(__name__)
 
 
+def schedule(socket, func, interval_mins, args=(), kwargs=None, class_args=(),
+             class_kwargs=None, queue=conf.DEFAULT_QUEUE_NAME):
+    """
+    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)
+        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, 300, msg, queue)
+
+
 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 +150,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 +252,24 @@ def send_request(socket, message, reply_requested=False, guarantee=False,
                       )
 
 
+def send_schedule_request(socket, interval_secs, message, queue=None):
+    """
+    Send a SCHEDULE 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.
+    """
+    send_emqp_message(socket, 'SCHEDULE',
+                      (queue or conf.DEFAULT_QUEUE_NAME,
+                       str(interval_secs),
+                       serialize(message)))
+
+
 def job(block=False):  # Move to decorators.py
     """
     run the decorated function on a worker
diff --git a/eventmq/jobmanager.py b/eventmq/jobmanager.py
index 9ea6800..6244c02 100644
--- a/eventmq/jobmanager.py
+++ b/eventmq/jobmanager.py
@@ -20,26 +20,26 @@ Ensures things about jobs and spawns the actual tasks
 import json
 import logging
 
-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.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
 
 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):
         """
@@ -64,7 +64,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()
 
         #: Jobs that are running should be stored in `active_jobs`. There
         #: should always be at most `available_workers` count of active jobs.
@@ -75,81 +76,21 @@ class JobManager(HeartbeatMixin):
 
         self._setup()
 
-    def _setup(self):
-        """
-        Prepares JobManager ready to connect to a broker. Actions that must
-        also run on a reset are here.
-        """
-        # Look for incoming events
-        self.poller.register(self.incoming, POLLIN)
-        self.awaiting_startup_ack = False
-
-        self.status = constants.STATUS.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')
-            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`
+        Starts the actual eventloop. Usually called by :meth:`start`
         """
+        # Acknowledgment has come
+        # Send a READY for each available worker
+        for i in range(0, self.available_workers):
+            self.send_ready()
+
         while True:
             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
@@ -165,7 +106,7 @@ class JobManager(HeartbeatMixin):
                 # Send a HEARTBEAT if necessary
                 if now - self._meta['last_sent_heartbeat'] >= \
                    conf.HEARTBEAT_INTERVAL:
-                    self.send_heartbeat(self.incoming)
+                    self.send_heartbeat(self.outgoing)
 
                 # Do something about any missed HEARTBEAT, if we have nothing
                 # waiting on the socket
@@ -215,49 +156,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()
+        sendmsg(self.outgoing, 'READY')
 
     def on_ack(self, msgid, ackd_msgid):
         """
diff --git a/eventmq/router.py b/eventmq/router.py
index 23700bf..d896853 100644
--- a/eventmq/router.py
+++ b/eventmq/router.py
@@ -82,6 +82,11 @@ class Router(HeartbeatMixin):
         #: workers available to take the job
         self.waiting_messages = {}
 
+        #: 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.
+        self.schedulers = []
+
     def start(self,
               frontend_addr='tcp://127.0.0.1:47290',
               backend_addr='tcp://127.0.0.1:47291'):
@@ -176,9 +181,9 @@ 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)
 
         self.send_ack(self.outgoing, sender, msgid)
@@ -305,8 +310,8 @@ class Router(HeartbeatMixin):
 
         # 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.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
diff --git a/eventmq/scheduler.py b/eventmq/scheduler.py
index b1b3fb6..24e95f5 100644
--- a/eventmq/scheduler.py
+++ b/eventmq/scheduler.py
@@ -24,7 +24,8 @@ from croniter import croniter
 from six import next
 
 from .sender import Sender
-from .utils.classes import HeartbeatMixin
+from .poller import Poller
+from .utils.classes import EMQPService, HeartbeatMixin
 from .utils.timeutils import seconds_until, timestamp
 from .client.messages import send_request
 
@@ -32,28 +33,27 @@ from .client.messages import send_request
 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()
 
+        # IDX     Description
         # 0 = the next ts this job should be executed
         # 1 = the function to be executed
         # 2 = the croniter iterator for this job
         self.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):
         """
@@ -75,14 +75,6 @@ class Scheduler(HeartbeatMixin):
                 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._start_event_loop()
-
     def _start_event_loop(self):
         """
         Starts the actual event loop. Usually called by :meth:`Scheduler.start`
@@ -117,6 +109,4 @@ class Scheduler(HeartbeatMixin):
 def test_job():
     print "hello!"
     print "hello!"
-    print "hello!"
-    print "hello!"
     time.sleep(4)
diff --git a/eventmq/utils/classes.py b/eventmq/utils/classes.py
index e98b9b7..b6ae03c 100644
--- a/eventmq/utils/classes.py
+++ b/eventmq/utils/classes.py
@@ -21,13 +21,176 @@ 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`, 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 = ('worker', '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.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 True:
+            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
+                    if msg[2] == "ACK":
+                        # :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 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.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)))
+
+    @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 +199,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 = {}
 
@@ -162,7 +326,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))