diff --git a/zipline/messaging.py b/zipline/messaging.py index 8c5076d1..05c77c70 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -10,7 +10,7 @@ import zipline.util as qutil class Component(object): - def __init__(self, addresses): + def __init__(self): """ :addresses: a dict of name_string -> zmq port address strings. Must have the following entries:: @@ -31,11 +31,12 @@ class Component(object): will also return a Poller. """ - self.zmq = None - self.context = None - self.addresses = addresses - self.out_socket = None - + self.zmq = None + self.context = None + self.addresses = None + self.out_socket = None + self.gevent_needed = False + def get_id(self): NotImplemented @@ -51,14 +52,15 @@ class Component(object): self.done = False #TODO: figure out why sockets can't be set in the __init__ self.sockets = [] - #TODO: figure out why addresses can't be set in the __init__ - self.addresses = {'sync_address' : "tcp://127.0.0.1:{port}".format(port=10100), - 'data_address' : "tcp://127.0.0.1:{port}".format(port=10101), - 'feed_address' : "tcp://127.0.0.1:{port}".format(port=10102), - 'merge_address' : "tcp://127.0.0.1:{port}".format(port=10103), - 'result_address' : "tcp://127.0.0.1:{port}".format(port=10104) - } - + + if self.gevent_needed: + qutil.LOGGER.info("Loading gevent specific zmq for {id}".format(id=self.get_id())) + module = __import__('gevent_zmq', 'zmq') + else: + qutil.LOGGER.debug("NOT Loading gevent specific zmq for {id}".format(id=self.get_id())) + module = __import__('zmq') + self.zmq = module + self.context = self.zmq.Context() self.open() self.setup_sync() @@ -66,6 +68,8 @@ class Component(object): #close all the sockets for sock in self.sockets: sock.close() + except: + qutil.LOGGER.exception("Problem running component {id}.".format(id=self.get_id())) finally: self.context.destroy() @@ -168,12 +172,8 @@ class ComponentHost(Component): """Component that can launch multiple sub-components, synchronize their start, and then wait for all components to be finished.""" def __init__(self, addresses, gevent_needed=False): - Component.__init__(self, addresses) - if gevent_needed: - module = __import__('gevent_zmq', 'zmq') - else: - module = __import__('zmq') - self.zmq = module + Component.__init__(self) + self.addresses = addresses #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 = {} @@ -188,7 +188,8 @@ class ComponentHost(Component): def register_components(self, component_list): for component in component_list: - component.zmq = self.zmq + component.gevent_needed = self.gevent_needed + component.addresses = self.addresses self.components[component.get_id()] = component self.sync_register[component.get_id()] = datetime.datetime.utcnow() if(isinstance(component, DataSource)): @@ -446,8 +447,8 @@ class DataSource(Component): means looping through all records in a store, converting to a dict, and calling send(map). """ - def __init__(self, source_id, addresses): - Component.__init__(self, addresses) + def __init__(self, source_id): + Component.__init__(self) self.id = source_id self.cur_event = None diff --git a/zipline/sources.py b/zipline/sources.py index 164576cb..385ffe0d 100644 --- a/zipline/sources.py +++ b/zipline/sources.py @@ -11,8 +11,8 @@ import zipline.messaging as qmsg class RandomEquityTrades(qmsg.DataSource): """Generates a random stream of trades for testing.""" - def __init__(self, sid, source_id, count, addresses): - qmsg.DataSource.__init__(self, source_id, addresses) + def __init__(self, sid, source_id, count): + qmsg.DataSource.__init__(self, source_id) self.count = count self.incr = 0 self.sid = sid diff --git a/zipline/test/client.py b/zipline/test/client.py index f13dfc4c..0fad7187 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -4,8 +4,8 @@ import zipline.messaging as qmsg class TestClient(qmsg.Component): - def __init__(self, addresses, utest, expected_msg_count=0): - qmsg.Component.__init__(self, addresses) + def __init__(self, utest, expected_msg_count=0): + qmsg.Component.__init__(self) self.received_count = 0 self.expected_msg_count = expected_msg_count self.utest = utest diff --git a/zipline/test/test_messaging.py b/zipline/test/test_messaging.py index 9e1d5747..71e05bc1 100644 --- a/zipline/test/test_messaging.py +++ b/zipline/test/test_messaging.py @@ -22,13 +22,14 @@ class ThreadSimulatorTestCase(unittest.TestCase): def setUp(self): """generate some config objects for the datafeed, sources, and transforms.""" - self.addresses = {'sync_address' : "tcp://127.0.0.1:{port}".format(port=10100), - 'data_address' : "tcp://127.0.0.1:{port}".format(port=10101), - 'feed_address' : "tcp://127.0.0.1:{port}".format(port=10102), - 'merge_address' : "tcp://127.0.0.1:{port}".format(port=10103), - 'result_address' : "tcp://127.0.0.1:{port}".format(port=10104) + self.addresses = {'sync_address' : "tcp://127.0.0.1:10100", + 'data_address' : "tcp://127.0.0.1:10101", + 'feed_address' : "tcp://127.0.0.1:10102", + 'merge_address' : "tcp://127.0.0.1:10103", + 'result_address' : "tcp://127.0.0.1:10104" } - + + self.addressesblarg = "test" def get_simulator(self): return ThreadSimulator(self.addresses) @@ -37,9 +38,9 @@ class ThreadSimulatorTestCase(unittest.TestCase): def test_sources_only(self): """streams events from two data sources, no transforms.""" sim = self.get_simulator() - ret1 = RandomEquityTrades(133, "ret1", 400, self.addresses) - ret2 = RandomEquityTrades(134, "ret2", 400, self.addresses) - client = TestClient(self.addresses, self, expected_msg_count=800) + ret1 = RandomEquityTrades(133, "ret1", 400) + ret2 = RandomEquityTrades(134, "ret2", 400) + client = TestClient(self, expected_msg_count=800) sim.register_components([ret1, ret2, client]) sim.simulate() @@ -54,11 +55,11 @@ class ThreadSimulatorTestCase(unittest.TestCase): verify message count at client. """ sim = self.get_simulator() - ret1 = RandomEquityTrades(133, "ret1", 5000, self.addresses) - ret2 = RandomEquityTrades(134, "ret2", 5000, self.addresses) + ret1 = RandomEquityTrades(133, "ret1", 5000) + ret2 = RandomEquityTrades(134, "ret2", 5000) mavg1 = MovingAverage("mavg1", 30) mavg2 = MovingAverage("mavg2", 60) - client = TestClient(self.addresses, self, expected_msg_count=10000) + client = TestClient(self, expected_msg_count=10000) sim.register_components([ret1, ret2, mavg1, mavg2, client]) sim.simulate() @@ -71,7 +72,7 @@ class ThreadSimulatorTestCase(unittest.TestCase): mavg1 = MovingAverage("mavg1", 30) mavg2 = MovingAverage("mavg2", 60) transforms = {"mavg1":mavg1, "mavg2":mavg2} - client = TestClient(self.addresses, self, expected_msg_count=0) + client = TestClient(self, expected_msg_count=0) sim = self.get_simulator(sources, transforms, client) sim.feed = DataFeedErr(sources.keys(), sim.data_address, sim.feed_address, sim.performance_address, qmsg.Sync(sim, "DataFeedErrorGenerator")) sim.simulate()