Added serializers.

docs/zipline.rst

@@ -49,6 +49,14 @@ zipline Package
     :undoc-members:
     :show-inheritance:
 
+:mod:`serial` Module
+--------------------
+
+.. automodule:: zipline.serial
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
 :mod:`sources` Module
 ---------------------
 
@@ -57,10 +65,10 @@ zipline Package
     :undoc-members:
     :show-inheritance:
 
-:mod:`topology` Module
-----------------------
+:mod:`topos` Module
+-------------------
 
-.. automodule:: zipline.topology
+.. automodule:: zipline.topos
     :members:
     :undoc-members:
     :show-inheritance:
@@ -73,14 +81,6 @@ zipline Package
     :undoc-members:
     :show-inheritance:
 
-:mod:`webui` Module
--------------------
-
-.. automodule:: zipline.webui
-    :members:
-    :undoc-members:
-    :show-inheritance:
-
 Subpackages
 -----------
 

docs/zipline.test.rst

@@ -9,6 +9,14 @@ test Package
     :undoc-members:
     :show-inheritance:
 
+:mod:`test_devsimulator` Module
+-------------------------------
+
+.. automodule:: zipline.test.test_devsimulator
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
 :mod:`test_messaging` Module
 ----------------------------
 
@@ -17,6 +25,14 @@ test Package
     :undoc-members:
     :show-inheritance:
 
+:mod:`test_monitor` Module
+--------------------------
+
+.. automodule:: zipline.test.test_monitor
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
 :mod:`test_sanity` Module
 -------------------------
 

zipline/monitor.py

@@ -2,7 +2,45 @@ import zmq
 
 class Controller(object):
     """
-    A broker of sorts.
+    A N to N messaging system for inter component communication.
+    Ostensibly a broker of sorts. Putting messages to the broker
+    is durable, if the broker goes down messages will queue up
+    until the HWM and then go out when the new broker comes up.
+
+    The other end is not durable, it is simply PUB/SUB which has
+    the benefit of of allowing more fluid time evolution of the
+    whole system since the messaging passing topology will not
+    alter itself as a result of more nodes listening.
+
+    The actual brokerin' is either a Python loop ( slow ) or a
+    zmq.FORWARDER device ( fast ).
+
+    :param pull_socket: Socket to subscribe to for republication of
+                        published messages. The endpoint for
+                        :func message_sender:.
+    :param pub_socket: Socket to publish messages, the starting
+                       point of :func message_listener:.
+    :param logging: Logging interface for tracking broker state
+        Defaults to None
+
+    Usage::
+
+        controller = Controller(
+            'tcp://127.0.0.1:5000',
+            'tcp://127.0.0.1:5001',
+        )
+
+        # typically you'd want to run this async to your main
+        # program since it blocks indefinetely.
+        controller.run()
+
+
+        sub = self.controller.message_listener()
+        push = self.controller.message_sender()
+
+        push.send('DIE')
+        sub.recv()
+
     """
 
     polling = False
@@ -30,6 +68,9 @@ class Controller(object):
         self.failed = 0
 
     def run(self, debug=False, context=None):
+        """
+        Run's the loop for the broker.
+        """
         self.polling = True
 
         if not context:
@@ -60,11 +101,11 @@ class Controller(object):
         self.pull = self._ctx.socket(zmq.PULL)
         self.pub = self._ctx.socket(zmq.PUB)
 
-        self.associated.extend([self.pull, self.pub])
-
         self.pull.bind(self.pull_socket)
         self.pub.bind(self.pub_socket)
 
+        self.associated.extend([self.pull, self.pub])
+
         while self.polling:
             try:
                 msg = self.pull.recv()

zipline/protocol.py

@@ -72,7 +72,7 @@ def FrameExceptionFactory(name):
 
 class namedict(object):
     """
-    So that you can use:
+    So that you can use::
 
         foo.BAR
         -- or --

zipline/serial.py

@@ -0,0 +1,91 @@
+"""
+Format serializer for Zipline.
+
+Because I'm opinionated about how you should send things over
+ZeroMQ. :)
+"""
+
+import zlib
+#import blosc
+import hmac
+import base64
+import numpy
+import pandas
+
+import cPickle as pickle
+
+# Pickle does the equivelant of builtin ``eval``. Be afraid, be
+# very afraid.
+
+def send_zipped_pickle(socket, obj, flags=0, protocol=-1):
+    """
+    Pickle an object, and zip the pickle before sending it.
+    """
+    p = pickle.dumps(obj, protocol)
+    z = zlib.compress(p)
+    return socket.send(z, flags=flags)
+
+def recv_zipped_pickle(socket, flags=0, protocol=-1):
+    """
+    Unpickle and uncompress a received object.
+    """
+    z = socket.recv(flags)
+    p = zlib.uncompress(z)
+    return pickle.loads(p)
+
+# HDF5, Numpy Byte Strings, Pandas arrays should use
+# blosc and reconstruct the Python container from the byte string
+# on the other side.
+
+def send_numpy(socket, obj, flags=0):
+    packed = blosc.pack_array(obj)
+    return socket.send(packed, flags=flags)
+
+def recv_numpy(socket, flags=0):
+    packed = blosc.unpack_array(socket.recv(flags))
+    return socket.send(packed, flags=flags)
+
+def send_pandas(socket, obj, flags=0):
+    ndarray = obj._data.blocks[0].values
+    socket.send_multipart(ndarray, flags=flags)
+    spec = (
+        obj._data.index,
+        obj._data.columns,
+        obj._data.blocks[0].dtype
+    )
+    return socket.send_multipart(spec, flags)
+
+def recv_pandas(socket, flags=0):
+    ndarray = socket.recv_multipart(flags)
+    spec = socket.recv_multipart(flags)
+    return pandas.DataFrame._init_ndarray(ndarray, *spec)
+
+def send_hdf5(self):
+    pass
+
+def recv_hdf5(self):
+    pass
+
+# Cryptographically secure wire protocol for ZeroMQ Using HMAC.
+
+# Compare byte strings, backported from Python 3.
+def byte_eq(a, b):
+    return not sum(0 if x==y else 1 for x, y in zip(a, b)) and len(a) == len(b)
+
+def send_secure(socket, data, key, flags=0):
+    msg = base64.b64encode(data)
+    sig = base64.b64encode(hmac.new(key, msg).digest())
+    return socket.send(bytes('!') + sig + bytes('?') + msg, flags=flags)
+
+def recv_secure(socket, data, key, flags):
+    data = socket.recv(flags=flags)
+
+    try:
+        sig, msg = data.split(bytes('?'), 1)
+    except ValueError:
+        raise Exception('Invalid signature/message pair.')
+
+    if byte_eq(sig[1:], base64.b64encode(hmac.new(key, msg).digest())):
+        return base64.b64decode(msg)
+    else:
+        raise Exception('Cryptographically invalid message received')