fix to logging, send multiple messages to workers

- JobManager now replies with READY when it's 'finished' with the job to
  requeue the worker in the broker.
- some improvements to logging that prevents duplicate messages

15 files changed, 220 insertions, 72 deletions

diff --git a/bin/router b/bin/router
index 356ac4b..0081e97 100755
--- a/bin/router
+++ b/bin/router
@@ -1,8 +1,9 @@
 #!/usr/bin/env python
 # -*- mode: python -*-
-
+from eventmq.log import setup_logger
 from eventmq.router import Router
 
 if __name__ == "__main__":
+    setup_logger('eventmq')
     r = Router()
     r.start()
diff --git a/bin/scheduler b/bin/scheduler
index b591ce9..33f164e 100755
--- a/bin/scheduler
+++ b/bin/scheduler
@@ -1,8 +1,9 @@
 #!/usr/bin/env python
 # -*- mode: python -*-
-
+from eventmq.log import setup_logger
 from eventmq.scheduler import Scheduler
 
 if __name__ == "__main__":
+    setup_logger("eventmq")
     s = Scheduler()
     s.start()
diff --git a/docs/protocol.rst b/docs/protocol.rst
index 86491dc..e468b06 100644
--- a/docs/protocol.rst
+++ b/docs/protocol.rst
@@ -5,7 +5,7 @@ EventMQ Protocol Specification
 
 Goals
 =====
-The EventMQ Protocol (eMQP) defines a reliable service-oriented request-reply and pub-sub dialog between a set of clients, a broker, and a set of workers. This goal is to 
+The EventMQ Protocol (eMQP) defines a reliable service-oriented request-reply and pub-sub dialog between a set of clients, a broker, and a set of workers. This goal is to
 
 The goals are to:
 
@@ -13,7 +13,7 @@ The goals are to:
  * Allow requests to be routed to workers by an abstracted service name.
  * Detect disconnected peers through heartbeating.
  * Allow for message tracing and debugging.
- 
+
 
 License
 =======
@@ -69,7 +69,7 @@ FRAME  Value          Description
 ====== ============== ===========
 0      _EMPTY_        leave empty
 1      eMQP/1.0       Protocol version
-2      READY          command
+2      REQUEST        command
 3      _MSGID_        A unique id for the msg
 4      _QUEUE_NAME_   the name of the queue the worker belongs to
 5      _HEADERS_      dictionary of headers. can be an empty set
diff --git a/eventmq/client/__init__.py b/eventmq/client/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/eventmq/client/__init__.py
diff --git a/eventmq/client/messages.py b/eventmq/client/messages.py
new file mode 100644
index 0000000..091b36f
--- /dev/null
+++ b/eventmq/client/messages.py
@@ -0,0 +1,61 @@
+# 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/>.
+"""
+:mod:`messages` -- Client Messaging
+===================================
+"""
+from json import dumps as serialize
+
+from .. import conf
+from ..utils.messages import send_emqp_message
+
+
+def send_request(socket, message, reply_requested=False, guarantee=False,
+                 retry_count=0, queue=None):
+    """
+    Send a REQUEST command.
+
+    Default headers are always all disabled by default. If they are included in
+    the headers then they have been enabled.
+    """
+    headers = []
+
+    if reply_requested:
+        headers.append('reply-requested')
+
+    if guarantee:
+        headers.append('guarantee')
+
+    if retry_count > 0:
+        headers.append('retry-count:%d' % retry_count)
+
+    send_emqp_message(socket, 'REQUEST',
+                      (queue or conf.DEFAULT_QUEUE_NAME,
+                       ",".join(headers),
+                       serialize(message))
+                      )
+
+
+def job(block=False):  # Move to decorators.py
+    """
+    run the decorated function on a worker
+
+    Args:
+        block (bool): Set to True if you wish to block and wait for the
+            response. This may be useful for running quick but cpu intesive
+            that would otherwise overwhelm a box that has to do it all alone.
+            (decryption?)
+    """
+    pass
diff --git a/eventmq/conf.py b/eventmq/conf.py
index ff4fe0f..6e48080 100644
--- a/eventmq/conf.py
+++ b/eventmq/conf.py
@@ -2,6 +2,11 @@
 # at different levels in the application
 SUPER_DEBUG = True
 
+# When a queue name isn't specified use this queue name for the default. It
+# would be a good idea to have a handful of workers listening on this queue
+# unless you're positive that everything specifies a queue with workers.
+DEFAULT_QUEUE_NAME = 'default'
+
 # {{{Job Manager
 # How long should we wait before retrying to connect to a broker?
 RECONNECT_TIMEOUT = 5  # in seconds
diff --git a/eventmq/jobmanager.py b/eventmq/jobmanager.py
index 17b0bbd..10a4a34 100644
--- a/eventmq/jobmanager.py
+++ b/eventmq/jobmanager.py
@@ -17,18 +17,19 @@
 ================================
 Ensures things about jobs and spawns the actual tasks
 """
+import logging
 import uuid
 
-from . import conf, constants, exceptions, log, utils
+from . import conf, constants, exceptions, utils
 from .poller import Poller, POLLIN
 from .sender import Sender
 from .utils.classes import HeartbeatMixin
 from .utils.messages import send_emqp_message as sendmsg
 import utils.messages
-from .utils.timeutils import monotonic, timestamp
+from .utils.timeutils import monotonic
 
 
-logger = log.get_logger(__file__)
+logger = logging.getLogger(__name__)
 
 
 class JobManager(HeartbeatMixin):
@@ -49,7 +50,6 @@ class JobManager(HeartbeatMixin):
                  generated.
         """
         super(JobManager, self).__init__(*args, **kwargs)
-        print self._meta
         self.name = kwargs.get('name', str(uuid.uuid4()))
         logger.info('Initializing JobManager %s...' % self.name)
         self.incoming = Sender()
@@ -141,6 +141,13 @@ class JobManager(HeartbeatMixin):
                         'Reconnecting...')
                     break
 
+    def on_request(self, msgid, msg):
+        """
+        Handles a REQUEST command
+        """
+        logger.debug("WHAT")
+        self.send_ready()
+
     def process_message(self, msg):
         """
         Processes a message
@@ -160,26 +167,31 @@ class JobManager(HeartbeatMixin):
             logger.error('Invalid message: %s' % str(msg))
             return
 
-        if conf.SUPER_DEBUG:
-            logger.debug("Received Message: %s" % msg)
-
         command = message[0]
         msgid = message[1]
         message = message[2]
 
         if hasattr(self, "on_%s" % command.lower()):
-            logger.debug('Calling on_%s' % command.lower())
+            if conf.SUPER_DEBUG:
+                logger.debug('Calling 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_inform(self):
+    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', 'default_queuename')
+        sendmsg(self.incoming, 'INFORM', queue or conf.DEFAULT_QUEUE_NAME)
         self._meta['last_sent_heartbeat'] = monotonic()
 
     def on_ack(self, msgid, ackd_msgid):
@@ -188,7 +200,7 @@ class JobManager(HeartbeatMixin):
         """
         # The msgid is the only frame in the message
         ackd_msgid = ackd_msgid[0]
-        logger.info('Received ACK for %s' % ackd_msgid)
+        logger.info('Received ACK for router (or client) %s' % ackd_msgid)
         self.awaiting_startup_ack = False
 
     def on_heartbeat(self, msgid, message):
diff --git a/eventmq/log.py b/eventmq/log.py
index 487042a..072e1f3 100644
--- a/eventmq/log.py
+++ b/eventmq/log.py
@@ -1,11 +1,15 @@
 """
 log module for eventmq
+
+this needs so much work.
 """
 import logging
 
 import zmq
 import zmq.log.handlers
 
+import watchtower
+
 
 FORMAT_STANDARD = logging.Formatter(
     '%(asctime)s - %(name)s  %(levelname)s - %(message)s')
@@ -28,12 +32,16 @@ class handlers(object):
     """
     PUBLISH_HANDLER = PUBHandler
     STREAM_HANDLER = logging.StreamHandler
+    CLOUDWATCH_HANDLER = watchtower.CloudWatchLogHandler
 
 
-def get_logger(name, formatter=FORMAT_NAMELESS,
-               handler=handlers.STREAM_HANDLER):
-    logger = logging.getLogger(name)
+def setup_logger(base_name, formatter=FORMAT_STANDARD,
+                 handler=handlers.STREAM_HANDLER):
+
+    logger = logging.getLogger(base_name)
     logger.setLevel(logging.DEBUG)
+
+    # remove handlers we don't want
     for h in logger.handlers:
         logger.removeHandler(h)
 
@@ -45,7 +53,9 @@ def get_logger(name, formatter=FORMAT_NAMELESS,
         time.sleep(1)
 
         handler = handler(_handler_sock)
-        handler.root_topic = name
+        handler.root_topic = base_name
+    elif handler == handlers.CLOUDWATCH_HANDLER:
+        handler = handler(log_group='eventmq-dev')
     else:
         handler = handler()
 
diff --git a/eventmq/poller.py b/eventmq/poller.py
index 9c05b2d..73a00a4 100644
--- a/eventmq/poller.py
+++ b/eventmq/poller.py
@@ -17,14 +17,14 @@
 =======================
 Device for polling sockets
 """
+import logging
 import uuid
 
 import zmq
 from zmq import Poller as ZPoller
 
-from . import log
 
-logger = log.get_logger(__package__)
+logger = logging.getLogger(__name__)
 
 POLLIN = zmq.POLLIN
 POLLOUT = zmq.POLLOUT
diff --git a/eventmq/receiver.py b/eventmq/receiver.py
index 70b1279..e1ec062 100644
--- a/eventmq/receiver.py
+++ b/eventmq/receiver.py
@@ -17,15 +17,16 @@
 ===========================
 The receiver is responsible for receiveing messages
 """
+import logging
 import uuid
 
 import zmq
 
-from . import constants, log
+from . import constants
 from .utils.classes import ZMQReceiveMixin, ZMQSendMixin
 
 
-logger = log.get_logger(__file__)
+logger = logging.getLogger(__name__)
 
 
 class Receiver(ZMQReceiveMixin, ZMQSendMixin):
diff --git a/eventmq/router.py b/eventmq/router.py
index 821f422..145918b 100644
--- a/eventmq/router.py
+++ b/eventmq/router.py
@@ -18,19 +18,21 @@
 Routes messages to workers (that are in named queues).
 """
 from copy import copy
-import uuid
+import logging
 
-from . import conf, exceptions, log, poller, receiver
+from . import conf, exceptions, poller, receiver
 from .constants import STATUS
 from .utils.classes import HeartbeatMixin
 from .utils.messages import (
     send_emqp_router_message as sendmsg,
+    fwd_emqp_router_message as fwdmsg,
     parse_router_message
 )
+from .utils.devices import generate_device_name
 from .utils.timeutils import monotonic, timestamp
 
 
-logger = log.get_logger(__file__)
+logger = logging.getLogger(__name__)
 
 
 class Router(HeartbeatMixin):
@@ -40,7 +42,7 @@ class Router(HeartbeatMixin):
     def __init__(self, *args, **kwargs):
         super(Router, self).__init__(*args, **kwargs)  # Creates _meta
 
-        self.name = str(uuid.uuid4())
+        self.name = generate_device_name()
         logger.info('Initializing Router %s...' % self.name)
 
         self.poller = poller.Poller()
@@ -141,14 +143,14 @@ class Router(HeartbeatMixin):
         """
         self._meta['last_sent_heartbeat'] = monotonic()
 
-        for k in self.workers:
-            self.send_heartbeat(self.outgoing, k)
+        for worker_id in self.workers:
+            self.send_heartbeat(self.outgoing, worker_id)
 
     def on_heartbeat(self, sender, msgid, msg):
         """
         a placeholder for a noop command. The actual 'logic' for HEARTBEAT is
-        in :meth:`self.process_worker_message` because any message from a worker
-        counts as a HEARTBEAT
+        in :meth:`self.process_worker_message` because any message from a
+        worker counts as a HEARTBEAT
         """
 
     def on_inform(self, sender, msgid, msg):
@@ -158,20 +160,16 @@ class Router(HeartbeatMixin):
         logger.info('Received INFORM request from %s' % sender)
         queue_name = msg[0]
 
-        # Add the worker to our worker dict
-        self.workers[sender] = {}
-        self.workers[sender]['queues'] = queue_name
-
-        # Add the worker to the queues it supports
-        if queue_name in self.queues:
-            self.queues[queue_name] += (sender,)
-        else:
-            self.queues[queue_name] = (sender,)
-        logger.debug('Adding %s to the worker pool for %s' %
-                     (sender, queue_name))
+        self.add_worker(sender, queue_name)
 
         self.send_ack(self.outgoing, sender, msgid)
 
+    def on_ready(self, sender, msgid, msg):
+        """
+        A worker that we should already know about is ready for another job
+        """
+        self.requeue_worker(sender)
+
     def clean_up_dead_workers(self):
         """
         Loops through the worker queues and removes any workers who haven't
@@ -189,6 +187,7 @@ class Router(HeartbeatMixin):
 
             # If a worker started, then immediatly died then no hb dictionary
             # was created so we should just remove that worker.
+            # hb stands for heartbeat
             if 'hb' not in self.workers[worker_id]:
                 logger.info('Removing worker %s from the queue due to no '
                             'heartbeat' % (worker_id))
@@ -202,7 +201,7 @@ class Router(HeartbeatMixin):
                 # Remove the worker from the actual worker queues
                 del self.workers[worker_id]
 
-    def add_worker(self, id, queues=None):
+    def add_worker(self, worker_id, queues=None):
         """
         Adds a worker to worker queues
 
@@ -210,6 +209,32 @@ class Router(HeartbeatMixin):
             worker_id: 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
+        self.workers[worker_id] = {}
+        self.workers[worker_id]['queues'] = queues
+
+        # Add the worker to the queues it supports
+        if queues in self.queues:
+            self.queues[queues] += [worker_id, ]
+        else:
+            self.queues[queues] = [worker_id, ]
+        logger.debug('Adding %s to the worker pool for %s' %
+                     (worker_id, str(queues)))
+
+    def requeue_worker(self, worker_id):
+        """
+        Add a worker back to the queue pool
+        """
+        if worker_id in self.workers:
+            queues = self.workers[worker_id].get('queues', None)
+        else:
+            queues = None
+
+        if queues:
+            logger.debug('Readding worker {} to queues {}'.
+                         format(worker_id, queues))
+
+            self.queues[queues].append(worker_id)
 
     def on_receive_request(self, msg):
         """
@@ -224,18 +249,24 @@ class Router(HeartbeatMixin):
 
         queue_name = message[3][0]
 
-        # cheat here and forward the message to the workers
-        self.outgoing.zsocket.send_multipart(msg)
-
         # 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")
 
+        try:
+            worker_addr = self.queues[queue_name].pop()
+        except IndexError:
+            # There were no workers in the queue
+            raise NotImplementedError("TODO: buffer when there are no workers "
+                                      "waiting in the queue")
+
+        fwdmsg(self.outgoing, worker_addr, msg[1:])
+
     def process_worker_message(self, msg):
         """
         This method is called when a message comes in from the worker socket.
-        It then calls `on_command`. If `on_command` isn't found, then a warning
-        is created.
+        It then calls `on_COMMAND.lower()`. If `on_command` isn't found, then
+        a warning is created.
 
         def on_inform(msg):
             pass
diff --git a/eventmq/scheduler.py b/eventmq/scheduler.py
index a9a823b..a234c34 100644
--- a/eventmq/scheduler.py
+++ b/eventmq/scheduler.py
@@ -17,18 +17,17 @@
 =============================
 Handles cron and other scheduled tasks
 """
+import logging
 import time
 
 from croniter import croniter
 from six import next
 
-from . import log
 from .sender import Sender
 from .utils.classes import HeartbeatMixin
-from .utils.devices import generate_device_name
 from .utils.timeutils import monotonic, seconds_until, timestamp
 
-logger = log.get_logger(__file__)
+logger = logging.getLogger(__name__)
 
 
 class Scheduler(HeartbeatMixin):
@@ -48,11 +47,12 @@ class Scheduler(HeartbeatMixin):
 
         self.load_jobs()
 
-    def connect(self, addr=''):
+    def connect(self, addr='tcp://127.0.0.1:47290'):
         """
         Connect the scheduler to worker/router at `addr`
         """
 
+
     def load_jobs(self):
         """
         Loads the jobs that need to be scheduled
@@ -117,4 +117,4 @@ class Scheduler(HeartbeatMixin):
                     logger.debug("Next execution will be in %ss" %
                                  seconds_until(self.jobs[i][0]))
 
-            time.sleep(1)
+            time.sleep(0.1)
diff --git a/eventmq/sender.py b/eventmq/sender.py
index e9eb655..7a780c4 100644
--- a/eventmq/sender.py
+++ b/eventmq/sender.py
@@ -17,14 +17,15 @@
 =======================
 The sender is responsible for sending messages
 """
+import logging
 import uuid
 
 import zmq
 
-from . import constants, exceptions, log
+from . import constants, exceptions
 from .utils.classes import ZMQReceiveMixin, ZMQSendMixin
 
-logger = log.get_logger(__file__)
+logger = logging.getLogger(__name__)
 
 
 class Sender(ZMQSendMixin, ZMQReceiveMixin):
diff --git a/eventmq/utils/classes.py b/eventmq/utils/classes.py
index 5a82b97..8b34431 100644
--- a/eventmq/utils/classes.py
+++ b/eventmq/utils/classes.py
@@ -17,13 +17,15 @@
 =================================
 Defines some classes to use when implementing ZMQ devices
 """
+import logging
+
 import zmq.error
 
-from .. import conf, exceptions, log
+from .. import conf, exceptions
 from ..utils.messages import send_emqp_message as sendmsg
 from ..utils.timeutils import monotonic, timestamp
 
-logger = log.get_logger(__file__)
+logger = logging.getLogger(__name__)
 
 
 class HeartbeatMixin(object):
@@ -99,13 +101,19 @@ class ZMQReceiveMixin(object):
         """
         Receive a message
         """
-        return self.zsocket.recv()
+        msg = self.zsocket.recv()
+        if conf.SUPER_DEBUG:
+            logger.debug('Received message: {}'.format(msg))
+        return msg
 
     def recv_multipart(self):
         """
         Receive a multipart message
         """
-        return self.zsocket.recv_multipart()
+        msg = self.zsocket.recv_multipart()
+        if conf.SUPER_DEBUG:
+            logger.debug('Received message: {}'.format(msg))
+        return msg
 
 
 class ZMQSendMixin(object):
@@ -117,16 +125,16 @@ class ZMQSendMixin(object):
         """
         Send a message directly to the 0mq socket. Automatically inserts some
         frames for your convience. The sent frame ends up looking something
-        like identity
+        like this
 
-            (this, '', protocol_version) + (your, tuple)
+            (_recipient_id, '', protocol_version) + (your, tuple)
 
         Args:
             message (tuple): Raw message to send.
             protocol_version (str): protocol version. it's good practice but
                 you may explicitly specify None to skip adding the version
             _recipient_id (object): When using a :attr:`zmq.ROUTER` you must
-                specify the the recipient id of the
+                specify the the recipient id of the remote socket
         """
         supported_msg_types = (tuple, list)
         if not isinstance(message, supported_msg_types):
@@ -143,7 +151,10 @@ class ZMQSendMixin(object):
             headers = ('', protocol_version, )
 
         msg = headers + message
-        logger.debug('Sending message: %s' % str(msg))
+
+        if conf.SUPER_DEBUG:
+            logger.debug('Sending message: %s' % str(msg))
+
         try:
             self.zsocket.send_multipart(msg)
         except zmq.error.ZMQError as e:
@@ -159,5 +170,4 @@ class ZMQSendMixin(object):
             protocol_version (str): protocol version. it's good practice, but
                 you may explicitly specify None to skip adding the version
         """
-        logger.debug('Sending message: %s' % str(message))
         self.send_multipart((message, ), protocol_version)
diff --git a/eventmq/utils/messages.py b/eventmq/utils/messages.py
index 643712c..3b26a1f 100644
--- a/eventmq/utils/messages.py
+++ b/eventmq/utils/messages.py
@@ -16,9 +16,13 @@
 :mod:`messages` -- Message Utilities
 ==========================================
 """
-from .. import constants, log, exceptions
+import logging
+
+from .. import constants, exceptions
 from . import random_characters
 
+logger = logging.getLogger(__name__)
+
 
 def parse_router_message(message):
     """
@@ -84,22 +88,20 @@ def generate_msgid(prefix=None):
     id = random_characters()
     return id if not prefix else str(prefix) + id
 
-logger = log.get_logger(__file__)
-
 
 def send_emqp_message(socket, command, message=None):
     """
     Formats and sends an eMQP message
 
     Args:
-
+        socket
+        command
+        message
     Raises:
-
-    Returns
     """
     msg = (str(command).upper(), generate_msgid())
     if message and isinstance(message, (tuple, list)):
-        msg += message
+        msg += tuple(message)
     elif message:
         msg += (message,)
 
@@ -129,3 +131,16 @@ def send_emqp_router_message(socket, recipient_id, command, message=None):
 
     socket.send_multipart(msg, constants.PROTOCOL_VERSION,
                           _recipient_id=recipient_id)
+
+
+def fwd_emqp_router_message(socket, recipient_id, payload):
+    """
+    Forwards `payload` to socket untouched.
+
+    .. note:
+       Because it's untouched, and because this function targets
+       :prop:`zmq.ROUTER`, it may be a good idea to first strip off the
+       leading sender id before forwarding it. If you dont you will need to
+       account for that on the recipient side.
+    """
+    socket.zsocket.send_multipart([recipient_id, ] + payload)