From 40f6c17be781fe0f4b0648be07692ffd42774f45 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Mon, 27 Feb 2012 09:56:43 -0500 Subject: [PATCH] Added serializers. --- docs/zipline.rst | 22 +++++------ docs/zipline.test.rst | 16 ++++++++ zipline/monitor.py | 47 ++++++++++++++++++++-- zipline/protocol.py | 2 +- zipline/serial.py | 91 +++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 163 insertions(+), 15 deletions(-) create mode 100644 zipline/serial.py diff --git a/docs/zipline.rst b/docs/zipline.rst index 6b639385..386b03de 100644 --- a/docs/zipline.rst +++ b/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 ----------- diff --git a/docs/zipline.test.rst b/docs/zipline.test.rst index ca56b7d6..443c1e28 100644 --- a/docs/zipline.test.rst +++ b/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 ------------------------- diff --git a/zipline/monitor.py b/zipline/monitor.py index d5fea061..4d208c0d 100644 --- a/zipline/monitor.py +++ b/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() diff --git a/zipline/protocol.py b/zipline/protocol.py index 7bde3fdf..d1d21728 100644 --- a/zipline/protocol.py +++ b/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 -- diff --git a/zipline/serial.py b/zipline/serial.py new file mode 100644 index 00000000..518dfe47 --- /dev/null +++ b/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')