Many changes

- organizes some code
- Router sends acknowledgement after jobmanager INFORMs

18 files changed, 550 insertions, 146 deletions

diff --git a/docs/api.rst b/docs/api.rst
index b0ecf4e..b3cdc83 100644
--- a/docs/api.rst
+++ b/docs/api.rst
@@ -5,9 +5,10 @@ API Documentation
 .. toctree::
    :maxdepth: 2
 
+   exceptions
    jobmanager
+   poller
    receiver
    router
    sender
    utils
-   exceptions
diff --git a/docs/contributing.rst b/docs/contributing.rst
index c0a03fb..2489abb 100644
--- a/docs/contributing.rst
+++ b/docs/contributing.rst
@@ -2,7 +2,13 @@
 Contributing to EventMQ
 #######################
 
+A few tips when working on the code
+
+ * Use relative imports. If you use absolute imports then when you `import eventmq.exceptions` it's possible that you receive in return a different version of eventmq. exceptions installed somewhere else on the system.
+
 .. toctree::
    :maxdepth: 1
    
    protocol
+   Source Code <https://github.com/enderlabs/eventmq>
+   Issues <https://github.com/enderlabs/eventmq/issues>
diff --git a/docs/exceptions.rst b/docs/exceptions.rst
new file mode 100644
index 0000000..04f9c5a
--- /dev/null
+++ b/docs/exceptions.rst
@@ -0,0 +1,2 @@
+.. automodule:: eventmq.exceptions
+   :members:
diff --git a/docs/poller.rst b/docs/poller.rst
new file mode 100644
index 0000000..1aab472
--- /dev/null
+++ b/docs/poller.rst
@@ -0,0 +1,3 @@
+.. automodule:: eventmq.poller
+   :members:
+   :special-members:
diff --git a/docs/protocol.rst b/docs/protocol.rst
index d9bd9e1..c32501c 100644
--- a/docs/protocol.rst
+++ b/docs/protocol.rst
@@ -46,9 +46,22 @@ From the 0MQ manual[[2](http://api.zeromq.org/master:zmq-socket)]
 
 This extra frame is not shown in the specifications below.
 
+Global Frames
+-------------
+An **ACK** command consists of a 4-frame multipart message, formatted as follows.
+
+====== ============== ===========
+FRAME  Value          Description
+====== ============== ===========
+0      _EMPTY_        leave empty
+1      eMQP/1.0       Protocol version
+2      ACK            command
+3      _MSGID_        A unique id for the msg
+====== ============== ===========
+
 eMQP / Client
 -------------
-A **REQUEST** command consists of 7-frame multipart message, formatted as follows. 
+A **REQUEST** command consists of a 7-frame multipart message, formatted as follows.
 
 ====== ============== ===========
 FRAME  Value          Description
@@ -62,6 +75,20 @@ FRAME  Value          Description
 6      _MSG_          The message to send
 ====== ============== ===========
 
+A **PUBLISH** command consists of a 7-frame multipart messag, formatted as follows.
+
+====== ============== ===========
+FRAME  Value          Description
+====== ============== ===========
+0      _EMPTY_        leave empty
+1      eMQP/1.0       Protocol version
+2      PUBLISH        command
+3      _MSGID_        A unique id for the msg
+4      _TOPIC_NAME_   the name of the queue the worker belongs to
+5      _HEADERS_      dictionary of headers. can be an empty set
+6      _MSG_          The message to send
+====== ============== ===========
+
 eMQP / Worker
 -------------
 An **INFORM** command consists of a 5-frame multipart message, formatted as follows.
@@ -128,3 +155,17 @@ Heartbeating
  * Both worker 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.
+
+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.
+
+Below is a table which defines and describes the headers.
+
+=============== ======= ======= ======= ===========
+Header          REQUEST PUBLISH Default Description
+=============== ======= ======= ======= ===========
+reply-requested X               False     Once the job is finished, send a reply back with information from the job. If there is no information reply with a True value.
+retry-count:#   X               0       Retry a failed job this many times before accepting defeat.
+guarantee       X               False     Ensure the job completes by letting someone else worry about a success reply.
+=============== ======= ======= ======= ===========
diff --git a/docs/utils/classes.rst b/docs/utils/classes.rst
new file mode 100644
index 0000000..ab48e06
--- /dev/null
+++ b/docs/utils/classes.rst
@@ -0,0 +1,3 @@
+.. automodule:: eventmq.utils.classes
+   :members:
+   :special-members:
diff --git a/docs/utils/messages.rst b/docs/utils/messages.rst
new file mode 100644
index 0000000..e49897c
--- /dev/null
+++ b/docs/utils/messages.rst
@@ -0,0 +1,3 @@
+.. automodule:: eventmq.utils.messages
+   :members:
+   :special-members:
diff --git a/eventmq/constants.py b/eventmq/constants.py
index aff5a59..621a6a5 100644
--- a/eventmq/constants.py
+++ b/eventmq/constants.py
@@ -1,9 +1,12 @@
 class STATUS(object):
-    wtf = -1
-    ready = 100
-    starting = 101
-    listening = 201
-    connected = 202
+    wtf = -1          # Something went wrong
+    ready = 100       # Waiting to connect or listen
+    starting = 101    # Starting to bind
+    listening = 102   # bound
+    connecting = 200
+    connected = 201
     stopping = 300
+    stopped = 301
 
+# See doc/protocol.rst
 PROTOCOL_VERSION = 'eMQP/1.0'
diff --git a/eventmq/jobmanager.py b/eventmq/jobmanager.py
index 85a90aa..d00d3e6 100644
--- a/eventmq/jobmanager.py
+++ b/eventmq/jobmanager.py
@@ -17,14 +17,17 @@
 ================================
 Ensures things about jobs and spawns the actual tasks
 """
+from time import sleep
 import uuid
 
-from zmq.eventloop import ioloop
-
 from . import constants
+from . import exceptions
 from . import log
 from . import utils
+from .poller import Poller, POLLIN
 from .sender import Sender
+from .utils.messages import send_emqp_message as sendmsg
+import utils.messages
 
 logger = log.get_logger(__file__)
 
@@ -46,55 +49,80 @@ class JobManager(object):
             name (str): unique name of this instance. By default a uuid will be
                  generated.
         """
-        ioloop.install()
         self.name = kwargs.get('name', str(uuid.uuid4()))
         self.incoming = Sender()
+        self.poller = Poller()
+
+        # Alert us of both incoming and outgoing events
+        self.poller.register(self.incoming, POLLIN)
 
         self.status = constants.STATUS.ready
 
+        # Are we waiting for an acknowledgment for something?
+        self.awaiting_ack = False
+
     def start(self, addr='tcp://127.0.0.1:47291'):
         """
-        Begin listening for job requests
+        Connect to `addr` and begin listening for job requests
 
         Args:
             args (str): connection string to connect to
         """
+        self.status = constants.STATUS.connecting
         self.incoming.connect(addr)
-        self.status = constants.STATUS.listening
 
+        self.awaiting_ack = True
+
+        #while self.awaiting_ack:
         self.send_inform()
-        ioloop.IOLoop.instance().start()
+        #    sleep(5)
 
-    def process_job(self, msg):
-        pass
+        self.status = constants.STATUS.connected
 
-    def sync(self):
-        pass
+        while True:
+            events = self.poller.poll(1000)
 
-    def send_message(self, command, message):
-        """
-        send a message to `self.incoming`
-        Args:
-            message: a msg tuple to send
-        Raises:
+            if events.get(self.incoming) == POLLIN:
+                msg = self.incoming.recv_multipart()
+                self.process_message(msg)
 
-        Returns
+    def process_message(self, msg):
         """
-        msg = (str(command).upper(), utils.generate_msgid())
-        if isinstance(message, (tuple, list)):
-            msg += message
+        Processes a message
+        """
+        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()):
+            logger.debug('Calling on_%s' % command.lower())
+            func = getattr(self, "on_%s" % command.lower())
+            func(msgid, message)
         else:
-            msg += (message,)
+            logger.warning('No handler for %s found (tried: %s)' %
+                           (command, ('on_%s' % command.lower)))
 
-        logger.debug('Sending message: %s' % str(msg))
-        self.incoming.send_multipart(msg, constants.PROTOCOL_VERSION)
+    def process_job(self, msg):
+        pass
+
+    def sync(self):
+        pass
 
     def send_inform(self):
         """
         Send an INFORM frame
         """
-        self.send_message('INFORM', 'default_queuename')
-
+        sendmsg(self.incoming, 'INFORM', 'default_queuename')
 
-    def respond(self):
-        pass
+    def on_ack(self, msgid, message):
+        """
+        Sets :attr:`awaiting_ack` to False
+        """
+        logger.info('Recieved ACK')
+        self.awaiting_ack = False
diff --git a/eventmq/log.py b/eventmq/log.py
index 836126d..487042a 100644
--- a/eventmq/log.py
+++ b/eventmq/log.py
@@ -8,7 +8,9 @@ import zmq.log.handlers
 
 
 FORMAT_STANDARD = logging.Formatter(
-    '%(asctime)s - %(name)s  %(levelname)s - "%(message)s')
+    '%(asctime)s - %(name)s  %(levelname)s - %(message)s')
+FORMAT_NAMELESS = logging.Formatter(
+    '%(asctime)s - %(levelname)s - %(message)s')
 
 
 class PUBHandler(zmq.log.handlers.PUBHandler):
@@ -28,10 +30,12 @@ class handlers(object):
     STREAM_HANDLER = logging.StreamHandler
 
 
-def get_logger(name, formatter=FORMAT_STANDARD,
+def get_logger(name, formatter=FORMAT_NAMELESS,
                handler=handlers.STREAM_HANDLER):
     logger = logging.getLogger(name)
     logger.setLevel(logging.DEBUG)
+    for h in logger.handlers:
+        logger.removeHandler(h)
 
     if handler == handlers.PUBLISH_HANDLER:
         _handler_sock = zmq.Context.instance().socket(zmq.PUB)
diff --git a/eventmq/poller.py b/eventmq/poller.py
new file mode 100644
index 0000000..82b8c2d
--- /dev/null
+++ b/eventmq/poller.py
@@ -0,0 +1,105 @@
+# 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:`poller` -- Poller
+=======================
+Device for polling sockets
+"""
+import uuid
+
+import zmq
+from zmq import Poller as ZPoller
+
+from . import log
+
+logger = log.get_logger(__package__)
+
+POLLIN = zmq.POLLIN
+POLLOUT = zmq.POLLOUT
+POLLINOUT = zmq.POLLIN | zmq.POLLOUT
+
+
+class Poller(ZPoller):
+    """
+    :class:`zmq.Poller` based class.
+    """
+    def __init__(self, *args, **kwargs):
+        """
+        See :class:`zmq.Poller`
+        """
+        super(Poller, self).__init__(*args, **kwargs)
+        self.name = str(uuid.uuid4())
+
+        self._sockets = []
+
+    def register(self, socket, flag=0):
+        """
+        Register a socket to be polled by this poller
+
+        Args:
+            socket (socket): The socket object to register
+            flags (int): :attr:`POLLIN`, :attr:`POLLOUT`, or
+                :attr:`POLLIN`|:attr:`POLLOUT` for both. If undefined the
+                socket remains unregistered.
+        """
+        if not flag:
+            logger.warning("Leaving socket %s unregistered because no flag "
+                           "was defined" % socket.name)
+            return
+        logger.debug('Registering %s with poller. (flags:%s)' % (socket.name,
+                                                                 flag))
+        self._sockets.append(socket)
+
+        super(Poller, self).register(socket.zsocket, flag)
+
+    def unregister(self, socket):
+        """
+        Unregister a socket from being polled
+
+        Args:
+            socket (socket): The socket object to registering
+        """
+        logger.debug("Unregistering %s from poller" % socket.name)
+        if socket not in self._sockets:
+            logger.warning("Attempt to unregister unregistered socket from "
+                           "poller: socket: %s" % socket.name)
+
+        self._sockets.remove(socket.zsocket)
+        super(Poller, self).unregister(socket.zsocket)
+
+    def poll(self, timeout=1):
+        """
+        Calling :meth:`zmq.Poller.poll` directly returns a tuple set. This
+        method typecasts to a dictionary for convience using the socket object
+        as the key. If a socket doesn't appear in the returned dictionary then
+        no events happened.
+
+        Args:
+            timeout (int): How long should poller wait for before iterating
+                the next loop.
+
+        Returns (dict) Dictionary using the socket as the key and the event the
+            socket generated as the value
+        """
+        events = dict(super(Poller, self).poll(timeout))
+        return_events = {}
+
+        for socket in self._sockets:
+            if socket.zsocket not in events:
+                continue
+
+            return_events[socket] = events[socket.zsocket]
+
+        return return_events
diff --git a/eventmq/receiver.py b/eventmq/receiver.py
index dc1c9ef..5800653 100644
--- a/eventmq/receiver.py
+++ b/eventmq/receiver.py
@@ -22,13 +22,14 @@ import uuid
 import zmq
 from zmq.eventloop import zmqstream
 
-import eventmq
-import log
+from . import constants, log
+from .utils.classes import ZMQReceiveMixin, ZMQSendMixin
+
 
 logger = log.get_logger(__file__)
 
 
-class Receiver(object):
+class Receiver(ZMQReceiveMixin, ZMQSendMixin):
     """
     Receives messages and pass them to a on_recv.
 
@@ -72,6 +73,8 @@ class Receiver(object):
         self.zsocket.setsockopt(zmq.ROUTER_MANDATORY, 1)
 
         if not self.skip_zmqstream:
+            # To use the built in ioloop, we have to wrap the socket in this
+            # ZMQStream object
             if not callable(self.on_recv):
                 raise TypeError('Required argument "on_recv" is not actually '
                                 'callable')
@@ -80,7 +83,7 @@ class Receiver(object):
             self.zsocket = zmqstream.ZMQStream(self.zsocket)
             self.zsocket.on_recv(self.on_recv)
 
-        self.status = eventmq.STATUS.ready
+        self.status = constants.STATUS.ready
 
     def listen(self, addr=None):
         """
@@ -94,7 +97,7 @@ class Receiver(object):
         """
         if self.ready:
             self.zsocket.bind(addr)
-            self.status = eventmq.STATUS.listening
+            self.status = constants.STATUS.listening
             logger.info('Receiver %s: Listening on %s' % (self.name, addr))
         else:
             raise Exception('Receiver %s not ready. status=%s' %
@@ -112,8 +115,8 @@ class Receiver(object):
         """
         if self.ready:
             self.zsocket.connect(addr)
-            self.status = eventmq.STATUS.connected
-            logger.info('Receiver %s: Connected to %s' % (self.name, addr))
+            self.status = constants.STATUS.connected
+            logger.debug('Receiver %s: Connected to %s' % (self.name, addr))
         else:
             raise Exception('Receiver %s not ready. status=%s' %
                             (self.name, self.status))
@@ -127,4 +130,4 @@ class Receiver(object):
             bool: True if the receiver is ready to connect or listen, otherwise
                 False
         """
-        return self.status == eventmq.STATUS.ready
+        return self.status == constants.STATUS.ready
diff --git a/eventmq/router.py b/eventmq/router.py
index b71c6e8..77b752d 100644
--- a/eventmq/router.py
+++ b/eventmq/router.py
@@ -22,9 +22,11 @@ import uuid
 from zmq.eventloop import ioloop
 
 from .constants import STATUS
-from . import log
-from . import receiver
-from . import utils
+from . import exceptions, log, receiver, utils
+from .utils.messages import (
+    send_emqp_router_message as sendmsg,
+    parse_router_message
+)
 
 logger = log.get_logger(__file__)
 
@@ -48,6 +50,7 @@ class Router(object):
 
         self.status = STATUS.ready
         logger.info('Done initializing Router %s' % self.name)
+        self.queues = {}
 
     def start(self,
               frontend_addr='tcp://127.0.0.1:47290',
@@ -70,21 +73,67 @@ class Router(object):
 
         ioloop.IOLoop.instance().start()
 
+    def send_ack(self, socket, recipient):
+        """
+        Sends an ACK response
+        """
+        logger.info('Sending ACK to %s' % recipient)
+        sendmsg(socket, recipient, 'ACK')
+
+    def on_inform(self, sender, msgid, msg):
+        """
+        Handles an INFORM message. Usually when new worker coming online
+        """
+        logger.info('Received INFORM request from %s')
+        queue_name = msg[0]
+
+        if queue_name in self.queues:
+            self.queues[queue_name] += (sender,)
+        else:
+            self.queues[queue_name] = (sender,)
+
+        self.send_ack(self.outgoing, sender)
+
     def on_receive_request(self, msg):
         """
         This function is called when a message comes in from the client socket.
         It then calls `on_command`. If `on_command` isn't found, then a
         warning is created.
         """
-        logger.info(str(utils.parse_message(msg)))
+        try:
+            message = parse_router_message(msg)
+        except exceptions.InvalidMessageError:
+            logger.exception('Invalid message from clients: %s' % str(msg))
+
+        queue_name = message[3][0]
 
         # do some things and forward it to the workers
         self.outgoing.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")
+
     def on_receive_reply(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.
+
+        def on_inform(msg):
+            pass
         """
-        logger.info(str(utils.parse_message(msg)))
+        try:
+            message = parse_router_message(msg)
+        except exceptions.InvalidMessageError:
+            logger.exception('Invalid message from workers: %s' % str(msg))
+            return
+
+        sender = message[0]
+        command = message[1]
+        msgid = message[2]
+        message = message[3]
+
+        if hasattr(self, "on_%s" % command.lower()):
+            func = getattr(self, "on_%s" % command.lower())
+            func(sender, msgid, message)
diff --git a/eventmq/sender.py b/eventmq/sender.py
index 2c19899..ddada6a 100644
--- a/eventmq/sender.py
+++ b/eventmq/sender.py
@@ -25,11 +25,12 @@ from zmq.eventloop import zmqstream
 from . import constants
 from . import exceptions
 from . import log
+from .utils.classes import ZMQReceiveMixin, ZMQSendMixin
 
 logger = log.get_logger(__file__)
 
 
-class Sender(object):
+class Sender(ZMQSendMixin, ZMQReceiveMixin):
     """
     Sends messages to a particular socket
 
@@ -39,7 +40,7 @@ class Sender(object):
 
     Attributes:
         name (str): Name of this socket
-        zcontext (:class`zmq.Context`): socket context
+        zcontext (:class:`zmq.Context`): socket context
         zsocket (:class:`zmq.Socket`): socket wrapped up in a
             :class:`zmqstream.ZMQStream`
     """
@@ -67,7 +68,7 @@ class Sender(object):
         self.zsocket.setsockopt(zmq.IDENTITY, self.name)
 
         if not kwargs.get('skip_zmqstream', True):
-            logger.debug('Using ZMQStream')
+            logger.info('Using ZMQStream')
             self.zsocket = zmqstream.ZMQStream(self.zsocket)
             self.zsocket.on_recv(kwargs.get('on_recv'))
 
@@ -86,10 +87,10 @@ class Sender(object):
         if self.ready:
             self.zsocket.bind(addr)
             self.status = constants.STATUS.listening
-            logger.info('Receiver %s: Listening on %s' % (self.name, addr))
+            logger.info('Listening on %s' % (addr))
         else:
-            raise exceptions.EventMQError('Receiver %s not ready. status=%s' %
-                                          (self.name, self.status))
+            raise exceptions.EventMQError('Not ready. status=%s' %
+                                          (self.status))
 
     def connect(self, addr=None):
         """
@@ -104,10 +105,10 @@ class Sender(object):
         if self.ready:
             self.zsocket.connect(addr)
             self.status = constants.STATUS.connected
-            logger.info('Receiver %s: Connected to %s' % (self.name, addr))
+            logger.debug('Connecting to %s' % (addr))
         else:
-            raise exceptions.EventMQError('Receiver %s not ready. status=%s' %
-                                          (self.name, self.status))
+            raise exceptions.EventMQError('Not ready. status=%s' %
+                                          (self.status))
 
     @property
     def ready(self):
@@ -119,55 +120,3 @@ class Sender(object):
                 False
         """
         return self.status == constants.STATUS.ready
-
-    def send_multipart(self, message, protocol_version):
-        """
-        Send a message directly to the 0mq socket. Automatically inserts some
-        frames for your convience. The sent frame ends up looking something
-        like identity
-
-            (this, '', 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
-        """
-        supported_msg_types = (tuple, list)
-        if not isinstance(message, supported_msg_types):
-            raise exceptions.MessageError(
-                '%s message type not one of %s' %
-                (type(message), str(supported_msg_types)))
-
-        if isinstance(message, list):
-            message = tuple(message)
-
-        headers = ('', protocol_version, )
-
-        message = headers + message
-        print message
-        self.zsocket.send_multipart(message)
-
-    def send(self, message, protocol_version):
-        """
-        Sends a message
-
-        Args:
-            message: message to send to something
-            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)
-
-    def recv(self):
-        """
-        Receive a message
-        """
-        return self.zsocket.recv()
-
-    def recv_multipart(self):
-        """
-        Receive a multipart message
-        """
-        return self.zsocket.recv_multipart
diff --git a/eventmq/tests/test_utils.py b/eventmq/tests/test_utils.py
index f32c4af..db3307d 100644
--- a/eventmq/tests/test_utils.py
+++ b/eventmq/tests/test_utils.py
@@ -16,11 +16,12 @@ import unittest
 
 from .. import exceptions
 from .. import utils
+import utils.messages
 
 
 class TestCase(unittest.TestCase):
     def test_generate_msgid(self):
-        msgid = utils.generate_msgid()
+        msgid = utils.messages.generate_msgid()
 
         self.assertEqual(type(msgid), str)
 
@@ -30,9 +31,9 @@ class TestCase(unittest.TestCase):
         emq_frame_manymsg = emq_headers + ('many', 'parts')
         emq_frame_nomsg = emq_headers
 
-        singlemsg = utils.parse_message(emq_frame_singlemsg)
-        manymsg = utils.parse_message(emq_frame_manymsg)
-        nomsg = utils.parse_message(emq_frame_nomsg)
+        singlemsg = utils.messages.parse_router_message(emq_frame_singlemsg)
+        manymsg = utils.messages.parse_router_message(emq_frame_manymsg)
+        nomsg = utils.messages.parse_router_message(emq_frame_nomsg)
 
         self.assertEqual(singlemsg[0], emq_frame_singlemsg[0])
         self.assertEqual(singlemsg[1], emq_frame_singlemsg[3])
@@ -51,4 +52,4 @@ class TestCase(unittest.TestCase):
 
         broken_message = ('dlkajfs', 'lkasdjf')
         with self.assertRaises(exceptions.InvalidMessageError):
-            utils.parse_message(broken_message)
+            utils.messages.parse_router_message(broken_message)
diff --git a/eventmq/utils.py b/eventmq/utils/__init__.py
index 756331f..af8a4dc 100644
--- a/eventmq/utils.py
+++ b/eventmq/utils/__init__.py
@@ -17,40 +17,18 @@
 =========================
 This module contains a handful of utility classes to make dealing with things
 like creating message more simple.
-"""
-import uuid
 
-from . import exceptions
+.. toctree ::
+   :maxdepth: 2
 
+   utils/classes
+   utils/messages
+"""
+import uuid
 
-def generate_msgid():
+def random_characters():
     """
-    Returns a (universally) unique id to be used for messages
+    Returns some random characters of a specified length
     """
+    # TODO: Pull out the random_chars function from eb.io code
     return str(uuid.uuid4())
-
-
-def parse_message(message):
-    """
-    Parses the generic format of an eMQP/1.0 message and returns the
-    parts.
-
-    Args:
-        message: the message you wish to have parsed
-
-    Returns (tuple) (sender_id, command, message_id, (message_body, and_data))
-    """
-    try:
-        sender = message[0]
-        # noop = message[1]
-        # protocol_version = message[2]
-        command = message[3]
-        msgid = message[4]
-    except IndexError:
-        raise exceptions.InvalidMessageError('Invalid Message Encountered: %s'
-                                             % str(message))
-    if len(message) > 5:
-        msg = message[5:]
-    else:
-        msg = ()
-    return (sender, command, msgid, msg)
diff --git a/eventmq/utils/classes.py b/eventmq/utils/classes.py
new file mode 100644
index 0000000..84e88fe
--- /dev/null
+++ b/eventmq/utils/classes.py
@@ -0,0 +1,93 @@
+# 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:`classes` -- Utility Classes
+=================================
+Defines some classes to use when implementing ZMQ devices
+"""
+
+from .. import exceptions
+from .. import log
+
+logger = log.get_logger(__file__)
+
+
+class ZMQReceiveMixin(object):
+    """
+    Defines some methods for receiving messages. This class will not work if
+    used on it's own
+    """
+    def recv(self):
+        """
+        Receive a message
+        """
+        return self.zsocket.recv()
+
+    def recv_multipart(self):
+        """
+        Receive a multipart message
+        """
+        return self.zsocket.recv_multipart()
+
+
+class ZMQSendMixin(object):
+    """
+    Defines some methods for sending messages. This class will not work if used
+    on it's own
+    """
+    def send_multipart(self, message, protocol_version, _recipient_id=None):
+        """
+        Send a message directly to the 0mq socket. Automatically inserts some
+        frames for your convience. The sent frame ends up looking something
+        like identity
+
+            (this, '', 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
+        """
+        supported_msg_types = (tuple, list)
+        if not isinstance(message, supported_msg_types):
+            raise exceptions.MessageError(
+                '%s message type not one of %s' %
+                (type(message), str(supported_msg_types)))
+
+        if isinstance(message, list):
+            message = tuple(message)
+
+        if _recipient_id:
+            headers = (_recipient_id, '', protocol_version)
+        else:
+            headers = ('', protocol_version, )
+
+        msg = headers + message
+        logger.debug('Sending message: %s' % str(msg))
+        self.zsocket.send_multipart(msg)
+
+    def send(self, message, protocol_version):
+        """
+        Sends a message
+
+        Args:
+            message: message to send to something
+            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
new file mode 100644
index 0000000..8abd374
--- /dev/null
+++ b/eventmq/utils/messages.py
@@ -0,0 +1,132 @@
+# 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` -- Message Utilities
+==========================================
+"""
+from .. import constants, log, exceptions
+from . import random_characters
+
+
+def parse_router_message(message):
+    """
+    Parses the generic format of an eMQP/1.0 message and returns the
+    parts.
+
+    Args:
+        message: the message you wish to have parsed
+
+    Returns (tuple) (sender_id, command, message_id, (message_body, and_data))
+    """
+    try:
+        sender = message[0]
+        # noop = message[1]
+        # protocol_version = message[2]
+        command = message[3]
+        msgid = message[4]
+    except IndexError:
+        raise exceptions.InvalidMessageError('Invalid Message Encountered: %s'
+                                             % str(message))
+    if len(message) > 5:
+        msg = message[5:]
+    else:
+        msg = ()
+    return (sender, command, msgid, msg)
+
+
+def parse_message(message):
+    """
+    Parses the generic format of an eMQP/1.0 message and returns the
+    parts.
+
+    Args:
+        message: the message you wish to have parsed
+
+    Returns (tuple) (command, message_id, (message_body, and_data))
+    """
+    try:
+        # noop = message[0]
+        # protocol_version = message[1]
+        command = message[2]
+        msgid = message[3]
+    except IndexError:
+        raise exceptions.InvalidMessageError('Invalid Message Encountered: %s'
+                                             % str(message))
+
+    print len(message)
+    if len(message) > 4:
+        msg = message[4:]
+    else:
+        msg = ()
+    return (command, msgid, msg)
+
+
+def generate_msgid(prefix=None):
+    """
+    Returns a random string to be used for message ids. Optionally the ID can
+    be prefixed with `prefix`.
+
+    Args:
+        prefix (str): Value to prefix on to the random part of the id. Useful
+            for prefixing some meta data to use for things
+    """
+    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:
+
+    Raises:
+
+    Returns
+    """
+    msg = (str(command).upper(), generate_msgid())
+    if message and isinstance(message, (tuple, list)):
+        msg += message
+    elif message:
+        msg += (message,)
+
+    socket.send_multipart(msg, constants.PROTOCOL_VERSION)
+
+
+def send_emqp_router_message(socket, recipient_id, command, message=None):
+    """
+    Formats and sends an eMQP message taking into account the recipient frame
+    used by a :attr:`zmq.ROUTER` device.
+
+    Args:
+        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
+        message: a msg tuple to send
+
+    Raises:
+
+    Returns
+    """
+    msg = (str(command).upper(), generate_msgid())
+    if message and isinstance(message, (tuple, list)):
+        msg += message
+    elif message:
+        msg += (message,)
+
+    socket.send_multipart(msg, constants.PROTOCOL_VERSION,
+                          _recipient_id=recipient_id)