Merge pull request #39 from com4/feature/status_subcommands

Feature/status subcommands

5 files changed, 64 insertions, 37 deletions

diff --git a/bin/emq-cli b/bin/emq-cli
index 52cf180..333f0f6 100755
--- a/bin/emq-cli
+++ b/bin/emq-cli
@@ -24,14 +24,14 @@ from pprint import pprint
 import sys
 
 from eventmq import exceptions
-from eventmq.client.messages import send_emqp_message
 from eventmq.constants import (
-        PROTOCOL_VERSION, ROUTER_SHOW_WORKERS, ROUTER_SHOW_SCHEDULERS
+        PROTOCOL_VERSION, STATUS_COMMANDS
 )
 from eventmq.poller import Poller, POLLIN
 from eventmq.sender import Sender
 from eventmq.utils.messages import generate_msgid
 
+
 class Shell(cmd.Cmd):
     """
     Interactive EventMQ Shell.
@@ -92,21 +92,21 @@ class Shell(cmd.Cmd):
         """
         Request the status of the connected component
         """
-        self.send_message('STATUS')
+        self.send_message('STATUS', ('',))
         pprint(self.recv_reply())
 
-    def do_show_workers(self, line):
+    def do_show_managers(self, line):
         """
-        Request the status of the connected workers and queues
+        Request the status of the connected job managers and queues
         """
-        self.send_message(ROUTER_SHOW_WORKERS)
+        self.send_message(('STATUS', (STATUS_COMMANDS.show_managers,)))
         pprint(self.recv_reply())
 
     def do_show_schedulers(self, line):
         """
         Request the status of the connected schedulers
         """
-        self.send_message(ROUTER_SHOW_SCHEDULERS)
+        self.send_message('STATUS', (STATUS_COMMANDS.show_schedulers,))
         pprint(self.recv_reply())
 
     def do_shutdown(self, line):
@@ -128,7 +128,7 @@ class Shell(cmd.Cmd):
             return
 
         message = (command, generate_msgid('admin:')) + message
-
+        print('sending: {}'.format(message))
         self.socket.send_multipart(message, PROTOCOL_VERSION)
 
     def recv_reply(self, timeout_secs=1000):
diff --git a/docs/protocol.rst b/docs/protocol.rst
index 66b6829..d63d4b1 100644
--- a/docs/protocol.rst
+++ b/docs/protocol.rst
@@ -72,7 +72,7 @@ FRAME  Value          Description
 4      _MSG_          The reply to respond with
 ====== ============== ===========
 
-A **HEARTBEAT** frame consists of a
+A **HEARTBEAT** frame consists of a 5-frame multipart message formatted as follows.
 
 ====== ============== ===========
 FRAME  Value          Description
@@ -84,7 +84,7 @@ FRAME  Value          Description
 4      _UNIX_TS_      A unix timestamp
 ====== ============== ===========
 
-A **DISCONNECT** frame consists of
+A **DISCONNECT** frame consists of a 4-frame multipart message formatted as follows.
 
 ====== ============== ===========
 FRAME  Value          Description
@@ -95,7 +95,7 @@ FRAME  Value          Description
 3      _MSGID_        A unique id for the msg
 ====== ============== ===========
 
-A **KBAI** frame consists of
+A **KBYE** frame consists of a 4-frame multipart message formatted as follows.
 
 ====== ============== ===========
 FRAME  Value          Description
@@ -106,6 +106,19 @@ FRAME  Value          Description
 3      _MSGID_        A unique id for the msg
 ====== ============== ===========
 
+A **STATUS** frame consists of a 5-frame multipart message formatted as follows.
+
+====== ============== ===========
+FRAME  Value          Description
+====== ============== ===========
+0      _EMPTY_        leave empty
+1      eMQP/1.0       Protocol version
+2      STATUS           command
+3      _MSGID_        A unique id for the msg
+4      _SUB_COMMAND_  A sub command (or empty string). Defined on a per device basis.
+====== ============== ===========
+
+
 eMQP / Client
 -------------
 A **REQUEST** command consists of a 7-frame multipart message, formatted as follows.
diff --git a/eventmq/constants.py b/eventmq/constants.py
index 8cc5d7a..b668f1e 100644
--- a/eventmq/constants.py
+++ b/eventmq/constants.py
@@ -18,13 +18,20 @@ class CLIENT_TYPE(object):
 # See doc/protocol.rst
 PROTOCOL_VERSION = 'eMQP/1.0'
 
-# PROTOCOL COMMANDS
 DISCONNECT = "DISCONNECT"
 KBYE = "KBYE"
+STATUS_CMD = "STATUS"
+
+
+class STATUS_COMMANDS(object):
+    """
+    Defines the STATUS sub commands
+    """
+    #: Router subommand to show connected job managbers
+    show_managers = 'show_managers'
+    #: Router subcommand to show connected schedulers
+    show_schedulers = 'show_schedulers'
 
-# ADMINISTRATIVE COMMANDS
-ROUTER_SHOW_WORKERS = 'ROUTER_SHOW_WORKERS'
-ROUTER_SHOW_SCHEDULERS = 'ROUTER_SHOW_SCHEDULERS'
 
 # ENVIRONMENT VARIABLES
 ENV_BROKER_ADDR = 'EMQ_CONNECT_ADDR'
diff --git a/eventmq/router.py b/eventmq/router.py
index 9971c4c..2b27026 100644
--- a/eventmq/router.py
+++ b/eventmq/router.py
@@ -24,8 +24,7 @@ import signal
 
 from . import constants, exceptions, poller, receiver
 from .constants import (
-    CLIENT_TYPE, DISCONNECT, KBYE, PROTOCOL_VERSION, ROUTER_SHOW_SCHEDULERS,
-    ROUTER_SHOW_WORKERS, STATUS
+    CLIENT_TYPE, DISCONNECT, KBYE, PROTOCOL_VERSION, STATUS, STATUS_COMMANDS
 )
 from .settings import conf, reload_settings
 from .utils import tuplify
@@ -176,25 +175,7 @@ class Router(HeartbeatMixin):
 
             if events.get(self.administrative_socket) == poller.POLLIN:
                 msg = self.administrative_socket.recv_multipart()
-                logger.debug('ADMIN: {}'.format(msg))
-                # ##############
-                # Admin Commands
-                # ##############
-                if len(msg) > 4:
-                    if msg[3] == DISCONNECT:
-                        logger.info('Received DISCONNECT from administrator')
-                        self.send_ack(
-                            self.administrative_socket, msg[0], msg[4])
-                        self.on_disconnect(msg[4], msg)
-                    elif msg[3] == 'STATUS':
-                        sendmsg(self.administrative_socket, msg[0], 'REPLY',
-                                (self.get_status(),))
-                    elif msg[3] == ROUTER_SHOW_WORKERS:
-                        sendmsg(self.administrative_socket, msg[0], 'REPLY',
-                                (self.get_workers_status(),))
-                    elif msg[3] == ROUTER_SHOW_SCHEDULERS:
-                        sendmsg(self.administrative_socket, msg[0], 'REPLY',
-                                (self.get_schedulers_status(),))
+                self.process_admin_message(msg)
 
             # TODO: Optimization: the calls to functions could be done in
             #     another thread so they don't block the loop. synchronize
@@ -822,6 +803,32 @@ class Router(HeartbeatMixin):
             func = getattr(self, "on_%s" % command.lower())
             func(sender, msgid, message)
 
+    def process_admin_message(self, msg):
+        """
+        This method is called when a message comes in from the administrative
+        socket.
+
+        Args:
+            msg: The untouched admin message from zmq
+        """
+        logger.debug('ADMIN MSG: {}'.format(msg))
+        if len(msg) > 4:
+            if msg[3] == DISCONNECT:
+                logger.info('Received DISCONNECT from administrator')
+                self.send_ack(
+                    self.administrative_socket, msg[0], msg[4])
+                self.on_disconnect(msg[4], msg)
+            elif msg[3] == 'STATUS':
+                if msg[5] == STATUS_COMMANDS.show_managers:
+                    sendmsg(self.administrative_socket, msg[0], 'REPLY',
+                            (self.get_workers_status(),))
+                elif msg[5] == STATUS_COMMANDS.show_schedulers:
+                    sendmsg(self.administrative_socket, msg[0], 'REPLY',
+                            (self.get_schedulers_status(),))
+                else:
+                    sendmsg(self.administrative_socket, msg[0], 'REPLY',
+                            (self.get_status(),))
+
     def _remove_worker(self, worker_id):
         """
         Remove worker with given id from any queues it belongs to.
diff --git a/setup.py b/setup.py
index 72e8e35..3f4eb70 100644
--- a/setup.py
+++ b/setup.py
@@ -19,7 +19,7 @@ setup(
                       'psutil==5.0.0',
                       'python-dateutil>=2.1,<3.0.0'],
     extras_require={
-          'docs': ['Sphinx==1.5.2', ],
+          'docs': ['Sphinx==1.5.3', ],
           'testing': [
               'flake8==3.2.1',
               'flake8-import-order==0.11',