13 files changed, 217 insertions, 44 deletions

diff --git a/bin/emq-cli b/bin/emq-cli
index 333f0f6..d781984 100755
--- a/bin/emq-cli
+++ b/bin/emq-cli
@@ -95,6 +95,10 @@ class Shell(cmd.Cmd):
         self.send_message('STATUS', ('',))
         pprint(self.recv_reply())
 
+    def do_show_scheduled_jobs(self, line):
+        self.send_message(('STATUS', (STATUS_COMMANDS.show_scheduled_jobs,)))
+        pprint(self.recv_reply())
+
     def do_show_managers(self, line):
         """
         Request the status of the connected job managers and queues
diff --git a/eventmq/client/jobs.py b/eventmq/client/jobs.py
index b61c65b..b06370f 100644
--- a/eventmq/client/jobs.py
+++ b/eventmq/client/jobs.py
@@ -86,8 +86,11 @@ class Job(object):
                 socket = Sender()
                 socket.connect(addr=self.broker_addr)
 
+                msgid = kwargs.get('msgid', None)
+
                 msgid = messages.defer_job(
-                    socket, f, args=args, kwargs=kwargs, queue=self.queue)
+                    socket, f, args=args, kwargs=kwargs, queue=self.queue,
+                    msgid=msgid)
 
                 return msgid
             else:
diff --git a/eventmq/client/messages.py b/eventmq/client/messages.py
index 45b323d..7188407 100644
--- a/eventmq/client/messages.py
+++ b/eventmq/client/messages.py
@@ -111,7 +111,8 @@ def schedule(socket, func, interval_secs=None, args=(), kwargs=None,
 def defer_job(
         socket, func, args=(), kwargs=None, class_args=(),
         class_kwargs=None, reply_requested=False, guarantee=False,
-        retry_count=0, timeout=0, queue=conf.DEFAULT_QUEUE_NAME):
+        retry_count=0, timeout=0, queue=conf.DEFAULT_QUEUE_NAME,
+        msgid=None):
     """
     Used to send a job to a worker to execute via `socket`.
 
@@ -141,6 +142,7 @@ def defer_job(
         queue (str): Name of queue to use when executing the job. If this value
             evaluates to False, the default is used. Default: is configured
             default queue name
+        msgid: (str): An optional msgid to use instead of a generated one
     Raises:
         TypeError: When one or more parameters are not JSON serializable.
     Returns:
@@ -191,13 +193,14 @@ def defer_job(
                          guarantee=guarantee,
                          retry_count=retry_count,
                          timeout=timeout,
-                         queue=queue)
+                         queue=queue,
+                         msgid=msgid)
 
     return msgid
 
 
 def send_request(socket, message, reply_requested=False, guarantee=False,
-                 retry_count=0, timeout=0, queue=None):
+                 retry_count=0, timeout=0, queue=None, msgid=None):
     """
     Send a REQUEST command.
 
@@ -238,6 +241,7 @@ def send_request(socket, message, reply_requested=False, guarantee=False,
             default: 0 which means infinite timeout
         queue (str): Name of queue to use when executing the job. Default: is
             configured default queue name
+        msgid: (str): An optional msgid to use instead of a generated one
 
     Returns:
         str: ID of the message
@@ -259,7 +263,8 @@ def send_request(socket, message, reply_requested=False, guarantee=False,
     msgid = send_emqp_message(socket, 'REQUEST',
                               (queue or conf.DEFAULT_QUEUE_NAME,
                                ",".join(headers),
-                               serialize(message)))
+                               serialize(message)),
+                              msgid=msgid)
 
     return msgid
 
diff --git a/eventmq/constants.py b/eventmq/constants.py
index b668f1e..cf2d5d0 100644
--- a/eventmq/constants.py
+++ b/eventmq/constants.py
@@ -27,11 +27,14 @@ class STATUS_COMMANDS(object):
     """
     Defines the STATUS sub commands
     """
-    #: Router subommand to show connected job managbers
+    #: Router subcommand to show connected job managbers
     show_managers = 'show_managers'
     #: Router subcommand to show connected schedulers
     show_schedulers = 'show_schedulers'
 
+    #: Scheduler subcommand to show scheduled jobs
+    show_scheduled_jobs = 'show_scheduled_jobs'
+
 
 # ENVIRONMENT VARIABLES
 ENV_BROKER_ADDR = 'EMQ_CONNECT_ADDR'
diff --git a/eventmq/router.py b/eventmq/router.py
index eceb080..89f92e8 100644
--- a/eventmq/router.py
+++ b/eventmq/router.py
@@ -24,7 +24,8 @@ import signal
 
 from . import constants, exceptions, poller, receiver
 from .constants import (
-    CLIENT_TYPE, DISCONNECT, KBYE, PROTOCOL_VERSION, STATUS, STATUS_COMMANDS
+    CLIENT_TYPE, DISCONNECT, KBYE, PROTOCOL_VERSION, STATUS, STATUS_CMD,
+    STATUS_COMMANDS
 )
 from .settings import conf, reload_settings
 from .utils import tuplify
@@ -280,30 +281,33 @@ class Router(HeartbeatMixin):
         queue_names = msg[0]
         client_type = msg[1]
 
-        if not queue_names:  # Ideally, this matches some workers
-            logger.error('Recieved INFORM message with no defined '
-                         'queues. Message was: {}'.format(msg))
-            return
-
-        try:
-            queues = list(map(tuplify, json.loads(queue_names)))
-        except ValueError:
-            # this was invalid json
-            logger.error(
-                'Received invalid queue names in INFORM. names:{} from:{} '
-                'type:{}'.format(
-                    queue_names, sender, client_type))
-            return
-
         logger.info('Received INFORM request from {} (type: {})'.format(
             sender, client_type))
 
         if client_type == CLIENT_TYPE.worker:
+            if not queue_names:  # Ideally, this matches some workers
+                logger.error('Recieved INFORM message with no defined '
+                             'queues. Message was: {}'.format(msg))
+                return
+
+            try:
+                queues = list(map(tuplify, json.loads(queue_names)))
+            except ValueError:
+                # this was invalid json
+                logger.error(
+                    'Received invalid queue names in INFORM. names:{} from:{} '
+                    'type:{}'.format(
+                        queue_names, sender, client_type))
+                return
+
             self.add_worker(sender, queues)
             self.send_ack(self.backend, sender, msgid)
         elif client_type == CLIENT_TYPE.scheduler:
             self.add_scheduler(sender)
             self.send_ack(self.frontend, sender, msgid)
+        else:
+            logger.error('Received invalid client type on INFORM ({}), '
+                         'ignoring'.format(client_type))
 
     def on_reply(self, sender, msgid, msg):
         """
@@ -819,7 +823,7 @@ class Router(HeartbeatMixin):
                 self.send_ack(
                     self.administrative_socket, msg[0], msg[4])
                 self.on_disconnect(msg[4], msg)
-            elif msg[3] == 'STATUS':
+            elif msg[3] == STATUS_CMD:
                 if msg[5] == STATUS_COMMANDS.show_managers:
                     sendmsg(self.administrative_socket, msg[0], 'REPLY',
                             (self.get_workers_status(),))
diff --git a/eventmq/scheduler.py b/eventmq/scheduler.py
index 799ae20..db72d82 100644
--- a/eventmq/scheduler.py
+++ b/eventmq/scheduler.py
@@ -30,13 +30,16 @@ from six import next
 
 from . import constants
 from .client.messages import send_request
-from .constants import KBYE
+from .constants import KBYE, STATUS_CMD, STATUS_COMMANDS
 from .poller import Poller, POLLIN
+from .receiver import Receiver
 from .sender import Sender
 from .settings import conf, reload_settings
 from .utils.classes import EMQPService, HeartbeatMixin
 from .utils.devices import generate_device_name
+from .utils.jsonencoders import EventMQEncoder
 from .utils.messages import send_emqp_message as sendmsg
+from .utils.messages import send_emqp_router_message as send_router_msg
 from .utils.timeutils import IntervalIter, monotonic, seconds_until, timestamp
 
 
@@ -80,6 +83,12 @@ class Scheduler(HeartbeatMixin, EMQPService):
         self.frontend = Sender(conf.NAME)
         self._redis_server = None
 
+        admin_addr = conf.SCHEDULER_ADMINISTRATIVE_LISTEN_ADDR
+
+        #: Port for administrative commands
+        self.administrative_socket = Receiver()
+        self.administrative_socket.listen(admin_addr)
+
         # contains dict of 4-item lists representing cron jobs key of this
         # dictionary is a hash of arguments, path, and callable from the
         # message of the SCHEDULE command received
@@ -105,6 +114,8 @@ class Scheduler(HeartbeatMixin, EMQPService):
         self.poller = Poller()
         self.load_jobs()
 
+        self.poller.register(self.administrative_socket, POLLIN)
+
         self._setup()
 
     def load_jobs(self):
@@ -143,6 +154,19 @@ class Scheduler(HeartbeatMixin, EMQPService):
             m_now = monotonic()
             events = self.poller.poll()
 
+            if events.get(self.administrative_socket) == POLLIN:
+                msg = self.administrative_socket.recv_multipart()
+                # ##############
+                # Admin Commands
+                # ##############
+                if len(msg) > 4:
+                    if msg[3] == STATUS_CMD:
+                        if msg[5] == STATUS_COMMANDS.show_scheduled_jobs:
+                            send_router_msg(self.administrative_socket,
+                                            msg[0],
+                                            'REPLY',
+                                            (self.get_scheduled_jobs(),))
+
             if events.get(self.frontend) == POLLIN:
                 msg = self.frontend.recv_multipart()
                 self.process_message(msg)
@@ -378,6 +402,9 @@ class Scheduler(HeartbeatMixin, EMQPService):
 
         schedule_hash = self.schedule_hash(message)
 
+        # If interval is negative, cron MUST be populated
+        interval_job = interval >= 0
+
         # Notify if this is updating existing, or new
         if (schedule_hash in self.cron_jobs or
                 schedule_hash in self.interval_jobs):
@@ -387,8 +414,7 @@ class Scheduler(HeartbeatMixin, EMQPService):
             logger.debug('Creating a new scheduled job with %s'
                          % schedule_hash)
 
-        # If interval is negative, cron MUST be populated
-        if interval >= 0:
+        if interval_job:
             inter_iter = IntervalIter(monotonic(), interval)
 
             self.interval_jobs[schedule_hash] = [
@@ -436,7 +462,8 @@ class Scheduler(HeartbeatMixin, EMQPService):
             if run_count > 0 or run_count == INFINITE_RUN_COUNT:
                 # Don't allow run_count to decrement below 0
                 if run_count > 0:
-                    self.interval_jobs[schedule_hash][4] -= 1
+                    if interval_job:
+                        self.interval_jobs[schedule_hash][4] -= 1
                 self.send_request(message[3], queue=queue)
 
     def get_run_count_from_headers(self, headers):
@@ -446,11 +473,25 @@ class Scheduler(HeartbeatMixin, EMQPService):
                 run_count = int(header.split(':')[1])
         return run_count
 
+    def on_status(self, msgid, message):
+
+        sendmsg(self.frontend, message[0], 'REPLY', (self.interval_jobs, ))
+
     def on_heartbeat(self, msgid, message):
         """
         Noop command. The logic for heartbeating is in the event loop.
         """
 
+    def get_scheduled_jobs(self):
+
+        return json.dumps(
+            {
+                'interval_jobs': self.interval_jobs,
+                'cron_jobs': self.cron_jobs,
+                'name': self.name,
+            },
+            cls=EventMQEncoder)
+
     @classmethod
     def schedule_hash(cls, message):
         """
diff --git a/eventmq/settings.py b/eventmq/settings.py
index a25e953..826b3a1 100644
--- a/eventmq/settings.py
+++ b/eventmq/settings.py
@@ -270,7 +270,15 @@ _CONFIG_DEFS = {
             'long-arg': '--redis-password',
             'type': str,
             'help': 'Password to redis'
-        }
+        },
+        'ADMINISTRATIVE_LISTEN_ADDR': {
+            'default': 'tcp://127.0.0.1:47294',
+            'long-arg': '--administrative-listen-addr',
+            'short-arg': '-S',
+            'type': str,
+            'help': 'Address to listen for administrative commands for '
+                    'schedulers',
+        },
     },
     'publisher': {
         'FRONTEND_LISTEN_ADDR': {
diff --git a/eventmq/tests/test_client_jobs.py b/eventmq/tests/test_client_jobs.py
index c658187..c66875a 100644
--- a/eventmq/tests/test_client_jobs.py
+++ b/eventmq/tests/test_client_jobs.py
@@ -34,13 +34,13 @@ class TestCase(unittest.TestCase):
         self.assertEqual(d(), 12)
         self.assertEqual(d.delay(), '43e14eaa-2034-4c84-8fe7-5577c70b6a7c')
         defer_job_mock.assert_called_with(Sender_mock(), test_func, args=(),
-                                          kwargs={}, queue=None)
+                                          msgid=None, kwargs={}, queue=None)
 
         d = jobs.job(test_func, broker_addr=self.BROKER_ADDR, queue='mojo')
         d.delay(1, 2, three=3)
         defer_job_mock.assert_called_with(Sender_mock(), test_func,
                                           args=(1, 2), kwargs={'three': 3},
-                                          queue='mojo')
+                                          msgid=None, queue='mojo')
 
     @mock.patch('eventmq.client.messages.defer_job')
     @mock.patch('eventmq.client.jobs.Sender')
@@ -85,7 +85,8 @@ class TestCase(unittest.TestCase):
         sched_mock.assert_called_with(
             Sender_mock(), test_func, args=(), class_args=(),
             class_kwargs=None, cron=None, headers=('guarantee',),
-            interval_secs=None, kwargs=None, queue='default', unschedule=True)
+            interval_secs=None, kwargs=None, queue='default',
+            unschedule=True)
 
 
 def test_func():
diff --git a/eventmq/tests/test_client_messages.py b/eventmq/tests/test_client_messages.py
index 68b847e..8939e9b 100644
--- a/eventmq/tests/test_client_messages.py
+++ b/eventmq/tests/test_client_messages.py
@@ -86,6 +86,7 @@ class TestCase(unittest.TestCase):
                                        guarantee=False,
                                        retry_count=3,
                                        timeout=0,
+                                       msgid=None,
                                        queue='test_queue')
 
         with LogCapture() as log_checker:
@@ -259,7 +260,8 @@ class TestCase(unittest.TestCase):
             socket, 'REQUEST',
             ('mozo',
              'reply-requested,retry-count:2',
-             messages.serialize(msg)))
+             messages.serialize(msg)),
+            msgid=None)
 
     @mock.patch('eventmq.client.messages.send_emqp_message')
     def test_send_request_all_headers(self, snd_empq_msg_mock):
@@ -282,7 +284,8 @@ class TestCase(unittest.TestCase):
             socket, 'REQUEST',
             ('default',
              'reply-requested,guarantee,retry-count:2,timeout:3',
-             messages.serialize(msg)))
+             messages.serialize(msg)),
+            msgid=None)
 
     @mock.patch('eventmq.client.messages.send_emqp_message')
     def test_send_schedule_request(self, snd_empq_msg_mock):
diff --git a/eventmq/tests/test_router.py b/eventmq/tests/test_router.py
index 0986f92..a737f89 100644
--- a/eventmq/tests/test_router.py
+++ b/eventmq/tests/test_router.py
@@ -75,9 +75,14 @@ class TestCase(unittest.TestCase):
 
             log_checker.check(
                 ('eventmq.router',
+                 'INFO',
+                 'Received INFORM request from {} (type: worker)'.format(
+                     sender_id)),
+
+                ('eventmq.router',
                  'ERROR',
-                 'Recieved INFORM message with no defined queues. Message '
-                 'was: {}'.format(msg))
+                 "Recieved INFORM message with no defined queues. Message "
+                 "was: {}".format(msg)),
             )
 
     def test_on_inform_invalid_queues(self):
@@ -93,9 +98,14 @@ class TestCase(unittest.TestCase):
 
             log_checker.check(
                 ('eventmq.router',
+                 'INFO',
+                 'Received INFORM request from {} (type: worker)'.format(
+                     sender_id)),
+
+                ('eventmq.router',
                  'ERROR',
                  'Received invalid queue names in INFORM. names:{} from:{} '
-                 'type:{}'.format(msg[0], sender_id, msg[1]))
+                 'type:{}'.format(msg[0], sender_id, msg[1])),
             )
 
     # @mock.patch('eventmq.router.Router.prioritize_queue_list')
diff --git a/eventmq/tests/test_scheduler.py b/eventmq/tests/test_scheduler.py
index dc96b09..82b459d 100644
--- a/eventmq/tests/test_scheduler.py
+++ b/eventmq/tests/test_scheduler.py
@@ -18,7 +18,7 @@ import unittest
 
 import mock
 
-from .. import constants, scheduler
+from .. import constants, scheduler, utils
 
 
 class TestCase(unittest.TestCase):
@@ -70,6 +70,81 @@ class TestCase(unittest.TestCase):
         h2 = scheduler.Scheduler.schedule_hash(msg2)
         self.assertEqual('4658982cab9d32bf1ef9113a9d8bdec01775e2bc', h2)
 
+    @mock.patch.object(utils.classes.ZMQSendMixin, 'send_multipart')
+    def test_on_schedule(self, send_mock):
+        override_settings = {}
+        sched = scheduler.Scheduler(override_settings=override_settings)
+
+        job_msg = json.dumps(['run', {
+            'path': 'test',
+            'args': [33, 'asdf'],
+            'kwargs': {'zeta': 'Z', 'alpha': 'α'},
+            'class_args': [0],
+            'class_kwargs': {
+                'donkey': True, 'apple': False},
+            'callable': 'do_the_thing'}])
+
+        msg = [
+            'default',
+            'run_count:3,guarantee',
+            '3',
+            job_msg,
+            None
+        ]
+
+        cron_msg = [
+            'default',
+            'run_count:3,guarantee',
+            '-1',
+            job_msg,
+            '* * * * *',
+        ]
+
+        sched.on_schedule('fake_msgid', msg)
+        self.assertEqual(1, len(sched.interval_jobs))
+        self.assertEqual(0, len(sched.cron_jobs))
+
+        self.assertEqual(1, len(json.loads(
+            sched.get_scheduled_jobs())['interval_jobs']))
+        self.assertEqual(0,
+                         len(json.loads(
+                             sched.get_scheduled_jobs())['cron_jobs']))
+
+        # Scheduling the same job as a cron should remove it from interval
+        sched.on_schedule('fake_msgid2', cron_msg)
+        self.assertEqual(0, len(sched.interval_jobs))
+        self.assertEqual(1, len(sched.cron_jobs))
+
+        self.assertEqual(0, len(json.loads(
+            sched.get_scheduled_jobs())['interval_jobs']))
+        self.assertEqual(1, len(json.loads(
+            sched.get_scheduled_jobs())['cron_jobs']))
+
+        # Change the job message and it should make new jobs
+        job_msg = json.dumps(['run', {
+            'path': 'test',
+            'args': [333, 'asdf'],
+            'kwargs': {'zeta': 'Z', 'alpha': 'α'},
+            'class_args': [0],
+            'class_kwargs': {
+                'donkey': True, 'apple': False},
+            'callable': 'do_the_thing'}])
+
+        msg[3] = job_msg
+
+        sched.on_schedule('fake_msgid3', msg)
+        self.assertEqual(1, len(sched.interval_jobs))
+        self.assertEqual(1, len(sched.cron_jobs))
+
+        self.assertEqual(1, len(json.loads(
+            sched.get_scheduled_jobs())['interval_jobs']))
+        self.assertEqual(1, len(json.loads(
+            sched.get_scheduled_jobs())['cron_jobs']))
+
+        # Make sure the scheduler obeyed 3 schedule commands with 'haste'
+        self.assertEqual(3, send_mock.call_count)
+
+
 # EMQP Tests
     def test_reset(self):
         sched = scheduler.Scheduler()
diff --git a/eventmq/utils/jsonencoders.py b/eventmq/utils/jsonencoders.py
new file mode 100644
index 0000000..a4d48a1
--- /dev/null
+++ b/eventmq/utils/jsonencoders.py
@@ -0,0 +1,9 @@
+import json
+
+from ..utils.timeutils import IntervalIter
+
+
+class EventMQEncoder(json.JSONEncoder):
+    def default(self, o):
+        if isinstance(o, IntervalIter):
+            return o.__dict__
diff --git a/eventmq/utils/messages.py b/eventmq/utils/messages.py
index 656d474..b664594 100644
--- a/eventmq/utils/messages.py
+++ b/eventmq/utils/messages.py
@@ -89,20 +89,23 @@ def generate_msgid(prefix=None):
     return id if not prefix else str(prefix) + id
 
 
-def send_emqp_message(socket, command, message=None):
+def send_emqp_message(socket, command, message=None, msgid=None):
     """
     Formats and sends an eMQP message
 
     Args:
-        socket
-        command
-        message
+        socket (zmq.socket) Socket to send on
+        command (str) Command to send
+        msgid: (str): An optional msgid to use instead of a generated one
+        message (tuple, str) Message to send
     Raises:
 
     Returns:
         str: Message id for this message
     """
-    msgid = generate_msgid()
+    if not msgid:
+        msgid = generate_msgid()
+
     msg = (str(command).upper(), msgid)
     if message and isinstance(message, (tuple, list)):
         msg += tuple(message)
@@ -114,7 +117,8 @@ def send_emqp_message(socket, command, message=None):
     return msgid
 
 
-def send_emqp_router_message(socket, recipient_id, command, message=None):
+def send_emqp_router_message(socket, recipient_id, command,
+                             message=None, msgid=None):
     """
     Formats and sends an eMQP message taking into account the recipient frame
     used by a :attr:`zmq.ROUTER` device.
@@ -123,12 +127,15 @@ def send_emqp_router_message(socket, recipient_id, command, message=None):
         socket: socket to send the message with
         recipient_id (str): the id of the connected device to reply to
         command (str): the eMQP command to send
+        msgid: (str): An optional msgid to use instead of a generated one
         message: a msg tuple to send
 
     Returns
         str: Message id for this message
     """
-    msgid = generate_msgid()
+    if not msgid:
+        msgid = generate_msgid()
+
     msg = (str(command).upper(), msgid)
     if message and isinstance(message, (tuple, list)):
         msg += message