diff --git a/docs/conf.py b/docs/conf.py index 743659ff..8ee7280d 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -25,7 +25,14 @@ sys.path.append(os.path.abspath('..')) # Add any Sphinx extension module names here, as strings. They can be extensions # coming with Sphinx (named 'sphinx.ext.*') or your custom ones. -extensions = ['sphinx.ext.autodoc', 'sphinx.ext.intersphinx', 'sphinx.ext.todo', 'sphinx.ext.coverage', 'sphinx.ext.mathjax', 'sphinx.ext.viewcode'] +extensions = [ + 'sphinx.ext.autodoc', + 'sphinx.ext.intersphinx', + 'sphinx.ext.todo', + 'sphinx.ext.coverage', + 'sphinx.ext.mathjax', + 'sphinx.ext.viewcode' +] # Add any paths that contain templates here, relative to this directory. templates_path = ['_templates'] diff --git a/zipline/component.py b/zipline/component.py index 782653f0..d5ddc37f 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -2,10 +2,12 @@ Commonly used messaging components. Contains the base class for all components. + """ import os import uuid +import time import socket import humanhash @@ -15,7 +17,7 @@ from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE class Component(object): """ Base class for components. Defines the the base messaging - interface between components. + interface for components. :param addresses: a dict of name_string -> zmq port address strings. Must have the following entries @@ -62,6 +64,10 @@ class Component(object): self.controller = None self.heartbeat_timeout = 2000 self.state_flag = COMPONENT_STATE.OK # OK | DONE | EXCEPTION + self._exception = None + + self.start_tic = None + self.stop_tic = None # Humanhashes make this way easier to debug because they # stick in your mind unlike a 32 byte string of random hex. @@ -82,37 +88,40 @@ class Component(object): # ------------ def open(self): - raise NotImplementedError - - def teardown_sockets(self): """ - Close all zmq sockets safely. - """ - #close all the sockets - for sock in self.sockets: - sock.close() - - def destroy(self): - """ - Clean shutdown. - - Tear down after normal operation. - """ - pass - - def kill(self): - """ - Unclean shutdown. - - Tear down ( fast ) as a mode of failure in the - simulation or on service halt. + Open the connections needed to start doing work. """ raise NotImplementedError + def ready(self): + """ + Return ``True`` if and only if the component has finished execution. + """ + return self.state_flag in [COMPONENT_STATE.DONE, \ + COMPONENT_STATE.EXCEPTION] + + def successful(self): + """ + Return ``True`` if and only if the component has finished execution + successfully, that is, without raising an error. + """ + return self.state_flag == COMPONENT_STATE.DONE and not \ + self.exception + + @property + def exception(self): + """ + Holds the exception that the component failed on, or + ``None`` if the component has not failed. + """ + return self._exception + def do_work(self): raise NotImplementedError def _run(self): + self.start_tick = time.clock() + self.done = False # TODO: use state flag self.sockets = [] @@ -132,9 +141,11 @@ class Component(object): self.setup_control() self.loop() - self.destroy() + self.shutdown() self.teardown_sockets() + self.end_tick = time.clock() + # shouldn't block if we've done our job correctly # self.context.term() @@ -159,6 +170,7 @@ class Component(object): self.signal_exception(exc) fail = exc finally: + # TODO: cleaner if self.context: self.context.destroy() if fail: @@ -186,12 +198,50 @@ class Component(object): self.receive_sync_ack() # blocking + def runtime(self): + if self.ready(): + return self.stop_tic - self.start_tic + + # ---------------------------- + # Cleanup & Modes of Failure + # ---------------------------- + + def teardown_sockets(self): + """ + Close all zmq sockets safely. This is universal, no matter + where this is running it will need the sockets closed. + """ + #close all the sockets + for sock in self.sockets: + sock.close() + + def shutdown(self): + """ + Clean shutdown. + + Tear down after normal operation. + """ + pass + + def kill(self): + """ + Unclean shutdown. + + Tear down ( fast ) as a mode of failure in the + simulation or on service halt. + + Context specific. + """ + raise NotImplementedError + # ---------------------- # Internal Maintenance # ---------------------- def signal_exception(self, exc=None): self.state_flag = COMPONENT_STATE.EXCEPTION + self._exception = exc + qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id)) def signal_done(self): @@ -311,7 +361,7 @@ class Component(object): def setup_sync(self): """ - Setup the sync socket and poller. + Setup the sync socket and poller. ( Connect ) """ qutil.LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id)) @@ -335,8 +385,33 @@ class Component(object): @property def get_id(self): + """ + The descriptive name of the component. + """ return 'UNKNOWN COMPONENT' + @property + def get_type(self): + """ + The data flow type of the component. + + - ``SOURCE`` + - ``CONDUIT`` + - ``SINK`` + + """ + raise NotImplementedError + + @property + def get_pure(self): + """ + Describes whehter this component purely functional, + i.e. for a given set of inputs is it guaranteed to + always give the same output . Components that are + side-effectful are, generally, not pure. + """ + return False + def debug(self): """ Debug information about the component. diff --git a/zipline/messaging.py b/zipline/messaging.py index 222a1011..d057a5f8 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -1,12 +1,14 @@ """ Commonly used messaging components. """ + import datetime import ujson as json import zipline.util as qutil from zipline.component import Component -from zipline.protocol import CONTROL_PROTOCOL +from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE, \ + COMPONENT_STATE class ComponentHost(Component): """ @@ -16,48 +18,67 @@ class ComponentHost(Component): def __init__(self, addresses, gevent_needed=False): Component.__init__(self) - self.addresses = addresses - self.gevent_needed = gevent_needed + self.addresses = addresses + self.gevent_needed = gevent_needed + self.running = False self.init() def init(self): + # Component Registry + # ---------------------- + self.components = {} + # ---------------------- + # workaround for defect in threaded use of strptime: # http://bugs.python.org/issue11108 qutil.parse_date("2012/02/13-10:04:28.114") - self.components = {} self.sync_register = {} self.timeout = datetime.timedelta(seconds=5) self.feed = ParallelBuffer() self.merge = MergedParallelBuffer() self.passthrough = PassthroughTransform() + self.controller = None #register the feed and the merge self.register_components([self.feed, self.merge, self.passthrough]) def register_controller(self, controller): + """ + Add the given components to the registry. Establish + communication with them. + """ + if self.controller != None: + raise Exception("There can be only one!") + self.controller = controller + # Propogate the controller to all the subcomponents for component in self.components.itervalues(): component.controller = controller def register_components(self, components): - for component in components: - component.gevent_needed = self.gevent_needed - component.addresses = self.addresses + """ + Add the given components to the registry. Establish + communication with them. + """ + assert isinstance(components, list) - if self.controller: - component.controller = self.controller + for component in components: + + component.gevent_needed = self.gevent_needed + component.addresses = self.addresses + component.controller = self.controller self.components[component.get_id] = component self.sync_register[component.get_id] = datetime.datetime.utcnow() - if(isinstance(component, DataSource)): + if isinstance(component, DataSource): self.feed.add_source(component.get_id) - if(isinstance(component, BaseTransform)): + if isinstance(component, BaseTransform): self.merge.add_source(component.get_id) def unregister_component(self, component_id): @@ -66,6 +87,7 @@ class ComponentHost(Component): def setup_sync(self): """ + Setup the sync socket and poller. ( Bind ) """ qutil.LOGGER.debug("Connecting sync server.") @@ -112,17 +134,17 @@ class ComponentHost(Component): if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN: msg = self.sync_socket.recv() - parts = msg.split(':') - sync_id, status = parts - # TODO: move into frame protocol - if len(parts) != 2: - qutil.LOGGER.info("got bad confirm: {msg}".format(msg=msg)) - continue + try: + parts = msg.split(':') + sync_id, status = parts + except ValueError as exc: + self.signal_exception(exc) if status == str(CONTROL_PROTOCOL.DONE): # TODO: other way around qutil.LOGGER.info("{id} is DONE".format(id=sync_id)) self.unregister_component(sync_id) + self.state_flag = COMPONENT_STATE.DONE else: self.sync_register[sync_id] = datetime.datetime.utcnow() @@ -143,6 +165,7 @@ class ComponentHost(Component): def teardown_component(self, component): raise NotImplementedError + class ParallelBuffer(Component): """ Connects to N PULL sockets, publishing all messages received to a PUB @@ -153,12 +176,16 @@ class ParallelBuffer(Component): def __init__(self): Component.__init__(self) + self.sent_count = 0 self.received_count = 0 self.draining = False - self.data_buffer = {} self.ds_finished_counter = 0 + # Depending on the size of this, might want to use a data + # structure with better asymptotics. + self.data_buffer = {} + def init(self): pass @@ -166,8 +193,13 @@ class ParallelBuffer(Component): def get_id(self): return "FEED" - def add_source(self, source_id): - self.data_buffer[source_id] = [] + @property + def get_type(self): + return COMPONENT_TYPE.CONDUIT + + # ------------- + # Core Methods + # ------------- def open(self): self.pull_socket = self.bind_data() @@ -191,9 +223,45 @@ class ParallelBuffer(Component): self.drain() self.signal_done() else: - event = json.loads(message) - self.append(event[u'id'], event) - self.send_next() + try: + event = json.loads(message) + + # JSON deserialization error + except ValueError as exc: + return self.signal_exception(exc) + + try: + self.append(event[u'id'], event) + self.send_next() + + # Invalid message + except KeyError as exc: + return self.signal_exception(exc) + + # ------------- + # Flow Control + # ------------- + + def drain(self): + """ + Send all messages in the buffer. + """ + self.draining = True + while self.pending_messages() > 0: + self.send_next() + + def send_next(self): + """ + Send the (chronologically) next message in the buffer. + """ + if(not(self.is_full() or self.draining)): + return + + event = self.next() + + if event != None: + self.feed_socket.send(json.dumps(event), self.zmq.NOBLOCK) + self.sent_count += 1 def append(self, source_id, value): """ @@ -242,25 +310,11 @@ class ParallelBuffer(Component): total += len(events) return total - def drain(self): + def add_source(self, source_id): """ - Send all messages in the buffer + Add a data source to the buffer. """ - self.draining = True - while(self.pending_messages() > 0): - self.send_next() - - def send_next(self): - """ - Send the (chronologically) next message in the buffer. - """ - if(not(self.is_full() or self.draining)): - return - - event = self.next() - if(event != None): - self.feed_socket.send(json.dumps(event), self.zmq.NOBLOCK) - self.sent_count += 1 + self.data_buffer[source_id] = [] def __len__(self): """ @@ -287,6 +341,10 @@ class MergedParallelBuffer(ParallelBuffer): def get_id(self): return "MERGE" + @property + def get_type(self): + return COMPONENT_TYPE.CONDUIT + def open(self): self.pull_socket = self.bind_merge() self.feed_socket = self.bind_result() @@ -309,13 +367,13 @@ class MergedParallelBuffer(ParallelBuffer): class BaseTransform(Component): """ - Top level execution entry point for the transform:: + Top level execution entry point for the transform - - connects to the feed socket to subscribe to events - - connets to the result socket (most oftened bound by a TransformsMerge) to PUSH transforms - - processes all messages received from feed, until DONE message received - - pushes all transforms - - sends DONE to result socket, closes all sockets and context + - connects to the feed socket to subscribe to events + - connets to the result socket (most oftened bound by a TransformsMerge) to PUSH transforms + - processes all messages received from feed, until DONE message received + - pushes all transforms + - sends DONE to result socket, closes all sockets and context Parent class for feed transforms. Subclass and override transform method to create a new derived value from the combined feed. @@ -323,8 +381,10 @@ class BaseTransform(Component): def __init__(self, name): Component.__init__(self) - self.state = {} - self.state['name'] = name + + self.state = { + 'name': name + } self.init() @@ -335,6 +395,10 @@ class BaseTransform(Component): def get_id(self): return self.state['name'] + @property + def get_type(self): + return COMPONENT_TYPE.CONDUIT + def open(self): """ Establishes zmq connections. @@ -347,9 +411,11 @@ class BaseTransform(Component): def do_work(self): """ Loops until feed's DONE message is received: - - receive an event from the data feed - - call transform (subclass' method) on event - - send the transformed event + + - receive an event from the data feed + - call transform (subclass' method) on event + - send the transformed event + """ socks = dict(self.poll.poll(self.heartbeat_timeout)) @@ -362,12 +428,28 @@ class BaseTransform(Component): self.signal_done() return - event = json.loads(message) - cur_state = self.transform(event) - cur_state['dt'] = event['dt'] - cur_state['id'] = self.state['name'] + try: + event = json.loads(message) + except ValueError as exc: + return self.signal_exception(exc) - self.result_socket.send(json.dumps(cur_state), self.zmq.NOBLOCK) + try: + cur_state = self.transform(event) + cur_state['dt'] = event['dt'] + cur_state['id'] = self.state['name'] + + # This is overloaded, so it can fail in all sorts of + # unknown ways. Its best to catch it in the + # Transformer itself. + except Exception as exc: + return self.signal_exception(exc) + + try: + json_frame = json.dumps(cur_state) + except ValueError as exc: + return self.signal_exception(exc) + + self.result_socket.send(json_frame, self.zmq.NOBLOCK) def transform(self, event): """ @@ -386,15 +468,30 @@ class BaseTransform(Component): class PassthroughTransform(BaseTransform): + """ + A bypass transform which is also an identity transform:: + + +-------+ + +---| f |---> + +-------+ + +------id-------> + + """ def __init__(self): BaseTransform.__init__(self, "PASSTHROUGH") + self.init() + def init(self): pass + @property + def get_type(self): + return COMPONENT_TYPE.CONDUIT + def transform(self, event): - return {'value':event} + return { 'value': event } class DataSource(Component): @@ -405,7 +502,8 @@ class DataSource(Component): """ def __init__(self, source_id): Component.__init__(self) - self.id = source_id + + self.id = source_id self.init() def init(self): @@ -415,19 +513,25 @@ class DataSource(Component): def get_id(self): return self.id + @property + def get_type(self): + return COMPONENT_TYPE.SOURCE + def open(self): - #create the data sink. Based on http://zguide.zeromq.org/py:tasksink2 self.data_socket = self.connect_data() def send(self, event): """ - event is expected to be a dict - sets id and type properties in the dict - sends to the data_socket. + Emit data. """ + assert isinstance(event, dict) + event['id'] = self.id event['type'] = self.get_type() - self.data_socket.send(json.dumps(event)) - def get_type(self): - raise NotImplementedError + try: + json_frame = json.dumps(event) + except ValueError as exc: + return self.signal_exception(exc) + + self.data_socket.send(json_frame) diff --git a/zipline/monitor.py b/zipline/monitor.py index a0cd0b08..2c869a03 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -1,4 +1,5 @@ import zmq +from zipline.protocol import CONTROL_PROTOCOL, CONTROL_FRAME class Controller(object): """ @@ -56,6 +57,7 @@ class Controller(object): self.push_socket = pull_socket # same port self.pub_socket = pub_socket self.sub_socket = pub_socket # same port + self.terminated = False if logging: self.logging = logging @@ -128,6 +130,8 @@ class Controller(object): self.failed += 1 continue + self.terminated = True + # ------------------- # Hooks for Endpoints # ------------------- @@ -161,20 +165,29 @@ class Controller(object): self.associated.append(s) return s + def shutdown(self, context=None): + self.polling = False + + if not context: + context = zmq.Context() + + s = self.message_sender(context) + s.send(CONTROL_FRAME( + 'controller', + CONTROL_PROTOCOL.SHUTDOWN, + )) + + #for asoc in self.associated: + #asoc.close() + def destroy(self): """ Manual cleanup. """ - self.polling = False - - for asoc in self.associated: - asoc.close() - - #if self._ctx: - #self._ctx.destroy() + self.shutdown() def __del__(self): - self.destroy() + self.shutdown() def qos(self): if not self.debug: diff --git a/zipline/protocol.py b/zipline/protocol.py index b13c5ae3..d6c41c6e 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -103,18 +103,18 @@ CONTROL_PROTOCOL = Enum( ) def CONTROL_FRAME(id, status): - assert isinstance(basestring, id) - assert isinstance(int, status) + assert isinstance(id, basestring,) + assert isinstance(status, int) return msgpack.dumps(tuple([id, status])) def CONTORL_UNFRAME(msg): - assert isinstance(basestring, msg) + assert isinstance(msg, basestring) try: id, status = msgpack.loads(msg) - assert isinstance(basestring, id) - assert isinstance(int, status) + assert isinstance(id, basestring) + assert isinstance(status, int) return id, status except TypeError: @@ -140,6 +140,12 @@ HEARTBEAT_PROTOCOL = namedict({ # Component State # ================== +COMPONENT_TYPE = Enum( + 'SOURCE' , # 0 + 'CONDUIT' , # 1 + 'SINK' , # 2 +) + COMPONENT_STATE = Enum( 'OK' , # 0 'DONE' , # 1 diff --git a/zipline/test/client.py b/zipline/test/client.py index ad4ee6c6..35db6943 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -2,7 +2,7 @@ import ujson as json import zipline.util as qutil import zipline.messaging as qmsg -from zipline.protocol import CONTROL_PROTOCOL +from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE class TestClient(qmsg.Component): @@ -15,13 +15,17 @@ class TestClient(qmsg.Component): self.init() def init(self): - self.received_count = 0 - self.prev_dt = None + self.received_count = 0 + self.prev_dt = None @property def get_id(self): return "TEST_CLIENT" + @property + def get_type(self): + return COMPONENT_TYPE.SINK + def open(self): self.data_feed = self.connect_result() @@ -31,25 +35,37 @@ class TestClient(qmsg.Component): if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN: msg = self.control_in.recv() - if self.data_feed in socks and socks[self.data_feed] == self.zmq.POLLIN: + if self.data_feed in socks and socks[self.data_feed] == self.zmq.POLLIN: msg = self.data_feed.recv() if msg == str(CONTROL_PROTOCOL.DONE): qutil.LOGGER.info("Client is DONE!") self.signal_done() - self.utest.assertEqual(self.expected_msg_count, self.received_count, - "The client should have received ({n}) the same number of messages as the feed sent ({m})." - .format(n=self.received_count, m=self.expected_msg_count)) + self.utest.assertEqual( + self.expected_msg_count, self.received_count, + "The client should have received ({n}) the same number of \ + messages as the feed sent ({m})." + .format(n=self.received_count, m=self.expected_msg_count)) return self.received_count += 1 - event = json.loads(msg) - if(self.prev_dt != None): + try: + event = json.loads(msg) + + # JSON deserialization error + except ValueError as exc: + return self.signal_exception(exc) + + if self.prev_dt != None: if not event['dt'] >= self.prev_dt: - raise Exception("Message out of order: {date} after {prev}".format(date=event['dt'], prev=prev_dt)) + raise Exception( + "Message out of order: {date} after {prev}".format( + date = event['dt'], prev = self.prev_dt + ) + ) + else: + self.prev_dt = event['dt'] - self.prev_dt = event['dt'] - - if(self.received_count % 100 == 0): + if self.received_count % 100 == 0: qutil.LOGGER.info("received {n} messages".format(n=self.received_count)) diff --git a/zipline/test/test_messaging.py b/zipline/test/test_messaging.py index 3ab7e69b..9aeb1653 100644 --- a/zipline/test/test_messaging.py +++ b/zipline/test/test_messaging.py @@ -4,12 +4,13 @@ Test suite for the messaging infrastructure of QSim. #don't worry about excessive public methods pylint: disable=R0904 from collections import defaultdict -import zipline.messaging as qmsg from zipline.transforms.technical import MovingAverage from zipline.sources import RandomEquityTrades from zipline.test.client import TestClient +from zipline.test.transform import DivideByZeroTransform + # Should not inherit form TestCase since test runners will pick # it up as a test. Its a Mixin of sorts at this point. @@ -111,15 +112,61 @@ class SimulatorTestCase(object): # Stop Running # ------------ - # TODO: less abrupt later, just shove a StopIteration - # down the pipe to make it stop spinning - sim.cuc._Thread__stop() + self.assertTrue(sim.ready()) + self.assertFalse(sim.exception) self.assertEqual(sim.feed.pending_messages(), 0, "The feed should be drained of all messages, found {n} remaining." .format(n=sim.feed.pending_messages()) ) + def test_simplefail(self): + + # Simple test just to make sure that the archiecture is + # responding. + + # Base Simuation + # -------------- + + # Allocate sockets for the simulator components + sockets = self.allocate_sockets(5) + + addresses = { + 'sync_address' : sockets[0], + 'data_address' : sockets[1], + 'feed_address' : sockets[2], + 'merge_address' : sockets[3], + 'result_address' : sockets[4] + } + + sim = self.get_simulator(addresses) + con = self.get_controller() + + # Simulation Components + # --------------------- + + ret1 = RandomEquityTrades(133, "ret1", 1) + ret2 = RandomEquityTrades(134, "ret2", 1) + fail_transform = DivideByZeroTransform("fail") + client = TestClient(self, expected_msg_count=ret1.count + ret2.count) + + sim.register_controller( con ) + sim.register_components([ret1, ret2, fail_transform, client]) + + # Simulation + # ---------- + sim.simulate() + + # Stop Running + # ------------ + + self.assertTrue(fail_transform.exception) + self.assertFalse(fail_transform.successful()) + + self.assertEqual(sim.feed.pending_messages(), 0, + "The feed should be drained of all messages, found {n} remaining." + .format(n=sim.feed.pending_messages()) + ) def test_sources_only(self): @@ -156,10 +203,8 @@ class SimulatorTestCase(object): # Stop Running # ------------ - - # TODO: less abrupt later, just shove a StopIteration - # down the pipe to make it stop spinning - sim.cuc._Thread__stop() + self.assertTrue(sim.ready()) + self.assertFalse(sim.exception) self.assertEqual(sim.feed.pending_messages(), 0, "The feed should be drained of all messages, found {n} remaining." @@ -203,30 +248,10 @@ class SimulatorTestCase(object): # Stop Running # ------------ - - # TODO: less abrupt later, just shove a StopIteration - # down the pipe to make it stop spinning - sim.cuc._Thread__stop() + self.assertTrue(sim.ready()) + self.assertFalse(sim.exception) self.assertEqual(sim.feed.pending_messages(), 0, "The feed should be drained of all messages, found {n} remaining." .format(n=sim.feed.pending_messages()) ) - - # TODO used? - def dtest_error_in_feed(self): - - ret1 = RandomEquityTrades(133, "ret1", 400) - ret2 = RandomEquityTrades(134, "ret2", 400) - sources = {"ret1":ret1, "ret2":ret2} - mavg1 = MovingAverage("mavg1", 30) - mavg2 = MovingAverage("mavg2", 60) - transforms = {"mavg1":mavg1, "mavg2":mavg2} - client = TestClient(self, expected_msg_count=0) - sim = self.get_simulator(sources, transforms, client) - - # TODO: way too long - sim.feed = DataFeedErr(sources.keys(), sim.data_address, sim.feed_address, sim.performance_address, qmsg.Sync(sim, "DataFeedErrorGenerator")) - sim.simulate() - - diff --git a/zipline/test/transform.py b/zipline/test/transform.py new file mode 100644 index 00000000..0f81a3f5 --- /dev/null +++ b/zipline/test/transform.py @@ -0,0 +1,22 @@ +from zipline.messaging import BaseTransform +from zipline.protocol import COMPONENT_TYPE + +class DivideByZeroTransform(BaseTransform): + """ + A transform that fails. + """ + + def __init__(self, name): + BaseTransform.__init__(self, "PASSTHROUGH") + self.state['name'] = name + self.init() + + def init(self): + pass + + @property + def get_type(self): + return COMPONENT_TYPE.CONDUIT + + def transform(self, event): + return { 'value': 0/0 } diff --git a/zipline/topology.py b/zipline/topology.py new file mode 100644 index 00000000..b7e842c9 --- /dev/null +++ b/zipline/topology.py @@ -0,0 +1,80 @@ +""" +Contains the various deployable topologies of ziplines. + +This is mostly hardcoded at the moment but as the topologies +becomes more sophisiticated this logic will be the primary +router of sockets. + +Ontology of Stream Processing +============================= + +Source +****** + +A producer of data. The data could be in a datastore, coming from a +socket, etc. To access this data, we pull from the source. Sources increase the +total amount of data flowing through the system. Sources are generally not +pure since they involve IO. + +Sink +**** + +A consumer of data. Basic examples would be a sum function (adding up a +stream of numbers fed in), a datastore sink, a socket etc. We push data +into a sink. When / If a sink completes processing, it may return some +value that exists outside of the system. + +Sinks decrease the total amount of information flowing through the system. + +Conduit +******* + +A transformer of data. We push data into a conduit. Similar to a sink, +but instead of returning a single value at the end, a conduit can +return multiple outputs every time it is pushed to. The returned values +remain in the system. + +Conduits may or may not be pure, it is usefull to distinguish between the +two since pure conduits have a variety of nice properties under composition + +""" + +from zipline.protocol import COMPONENT_TYPE + +class Topology(object): + pass + +class DiamondTopology(Topology): + """ + Exposes a feed, merge, and passthrough bypass:: + + +--------+ + +---------->| |---------------+ + | +--------+ | + | v + +---+----+ +---+----+ +--------+ +--------+ +---+----+ + | +-->| +----->| |---------->| |--->| | + +---+----+ +---+----+ +--------+ +--------+ +---+----+ + | ^ + | +--------+ | + +---------->| |---------------+ + | +--------+ | + | | + +------------passthru----------------+ + + """ + + flow = { + 'flow' : COMPONENT_TYPE.SOURCE , + 'serializers' : COMPONENT_TYPE.CONDUIT , + 'transforms' : COMPONENT_TYPE.CONDUIT , + 'merges' : COMPONENT_TYPE.CONDUIT , + 'clients' : COMPONENT_TYPE.SINK , + } + + def __init__(self): + self.sources = [] + self.serializers = [] + self.transforms = [] + self.merges = [] + self.clients = []