Implement Scheduler status command

4 files changed, 65 insertions, 17 deletions

diff --git a/eventmq/constants.py b/eventmq/constants.py
index 8cc5d7a..cd6fe8a 100644
--- a/eventmq/constants.py
+++ b/eventmq/constants.py
@@ -22,9 +22,12 @@ PROTOCOL_VERSION = 'eMQP/1.0'
 DISCONNECT = "DISCONNECT"
 KBYE = "KBYE"
 
-# ADMINISTRATIVE COMMANDS
+# Router ADMINISTRATIVE COMMANDS
 ROUTER_SHOW_WORKERS = 'ROUTER_SHOW_WORKERS'
 ROUTER_SHOW_SCHEDULERS = 'ROUTER_SHOW_SCHEDULERS'
 
+# Scheduler ADMINISTRATIVE COMMANDS
+SCHEDULER_SHOW_SCHEDULED_JOBS = 'SCHEDULER_SHOW_SCHEDULED_JOBS'
+
 # ENVIRONMENT VARIABLES
 ENV_BROKER_ADDR = 'EMQ_CONNECT_ADDR'
diff --git a/eventmq/router.py b/eventmq/router.py
index 9971c4c..d2b442a 100644
--- a/eventmq/router.py
+++ b/eventmq/router.py
@@ -298,25 +298,25 @@ 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:
diff --git a/eventmq/scheduler.py b/eventmq/scheduler.py
index 3abe58a..09f8db2 100644
--- a/eventmq/scheduler.py
+++ b/eventmq/scheduler.py
@@ -31,11 +31,14 @@ from . import constants
 from .client.messages import send_request
 from .constants import KBYE
 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
 
 
@@ -75,6 +78,12 @@ class Scheduler(HeartbeatMixin, EMQPService):
         self.frontend = Sender()
         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
@@ -100,6 +109,8 @@ class Scheduler(HeartbeatMixin, EMQPService):
         self.poller = Poller()
         self.load_jobs()
 
+        self.poller.register(self.administrative_socket, POLLIN)
+
         self._setup()
 
     def load_jobs(self):
@@ -135,6 +146,18 @@ 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] == constants.SCHEDULER_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)
@@ -417,11 +440,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 18ec306..c07f275 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'
-        }
+        },
+        'SCHEDULER_ADMINISTRATIVE_LISTEN_ADDR': {
+            'default': 'tcp://127.0.0.1:47294',
+            'long-arg': '--scheduler-administrative-listen-addr',
+            'short-arg': '-S',
+            'type': str,
+            'help': 'Address to listen for administrative commands for '
+                    'schedulers',
+        },
     },
     'publisher': {
         'FRONTEND_LISTEN_ADDR': {