diff --git a/qsim/core.py b/qsim/core.py index dbf97bb6..a485aa33 100644 --- a/qsim/core.py +++ b/qsim/core.py @@ -7,6 +7,7 @@ import json import copy import threading import datetime +import atexit import qsim.util as qutil import qsim.messaging as qmsg @@ -175,6 +176,7 @@ class DataFeed(object): self.data_socket = None self.context = None self.poller = None + def open(self): # Prepare our context and sockets @@ -196,10 +198,14 @@ class DataFeed(object): self.sync.open() def close(self): - self.data_socket.close() - self.feed_socket.close() - self.sync.close() - self.context.term() + try: + self.data_socket.close() + self.feed_socket.close() + self.sync.close() + except: + qutil.LOGGER.exception("Error closing DataFeed") + finally: + self.context.destroy() def handle_all(self): qutil.LOGGER.info("entering feed loop on {addr}".format(addr=self.data_address)) @@ -332,10 +338,14 @@ class BaseTransform(object): r=self.received_count, s=self.sent_count)) - self.feed_socket.close() - self.result_socket.close() - self.sync.close() - self.context.term() + try: + self.feed_socket.close() + self.result_socket.close() + self.sync.close() + except: + qutil.LOGGER.exception("Error closing Transforms") + finally: + self.context.destroy() def transform(self, event): """ Must return the transformed value as a map with {name:"name of new transform", value: "value of new field"} @@ -413,10 +423,14 @@ class TransformsMerge(object): """ Close all zmq sockets and context. """ - self.transform_socket.close() - self.feed_socket.close() - self.result_socket.close() - self.context.term() + try: + self.transform_socket.close() + self.feed_socket.close() + self.result_socket.close() + except: + qutil.LOGGER.exception("Error closing merge") + finally: + self.context.destroy() def process_all(self): """ diff --git a/qsim/messaging.py b/qsim/messaging.py index f6051ece..0f8946c0 100644 --- a/qsim/messaging.py +++ b/qsim/messaging.py @@ -145,7 +145,8 @@ class Sync(object): try: self.sync_socket.send(self.sync_id + ":DONE", zmq.NOBLOCK) self.sync_socket.close() - self.context.term() except: - pass #just don't want to error out on closing + qutil.LOGGER.exception("Error closing Sync object") + finally: + self.context.destroy() \ No newline at end of file diff --git a/qsim/sources.py b/qsim/sources.py index d50414b7..d5b387c5 100644 --- a/qsim/sources.py +++ b/qsim/sources.py @@ -72,20 +72,16 @@ class DataSource(object): done_msg['type'] = 'DONE' done_msg['s'] = self.source_id self.data_socket.send(json.dumps(done_msg), zmq.NOBLOCK) + + qutil.LOGGER.info("closing data socket") + self.data_socket.close() + qutil.LOGGER.info("closing sync") + self.sync.close() + qutil.LOGGER.info("closing context") except: qutil.LOGGER.exception("failed to send DONE message") - pass #continue with the closing. - - qutil.LOGGER.info("closing data socket") - self.data_socket.close() - qutil.LOGGER.info("closing sync") - self.sync.close() - qutil.LOGGER.info("closing context") - try: - self.context.term() - qutil.LOGGER.info("done") - except: - qutil.LOGGER.exception("error closing context") + finally: + self.context.destroy() qutil.LOGGER.info("finished processing data source") class RandomEquityTrades(DataSource): diff --git a/qsim/test/client.py b/qsim/test/client.py index 58d0cf7c..9705dbc8 100644 --- a/qsim/test/client.py +++ b/qsim/test/client.py @@ -51,13 +51,13 @@ class TestClient(object): qutil.LOGGER.info("received {n} messages".format(n=self.received_count)) qutil.LOGGER.info("received {n} messages".format(n=self.received_count)) + self.data_feed.close() + self.sync.close() except: self.error = True qutil.LOGGER.exception("**********************Error in test client.") finally: - self.data_feed.close() - self.sync.close() - self.context.term() + self.context.destroy() 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})." diff --git a/qsim/test/test_messaging.py b/qsim/test/test_messaging.py index 7c933449..ce8c66e8 100644 --- a/qsim/test/test_messaging.py +++ b/qsim/test/test_messaging.py @@ -24,9 +24,9 @@ class MessagingTestCase(unittest.TestCase): qutil.configure_logging() def get_simulator(self, sources, transforms, client, feed=None, merge=None): - return ThreadSimulator(sources, transforms, client, feed=feed, merge=merge) + return ProcessSimulator(sources, transforms, client, feed=feed, merge=merge) - def dtest_sources_only(self): + def test_sources_only(self): """streams events from two data sources, no transforms.""" ret1 = RandomEquityTrades(133, "ret1", 400) @@ -41,7 +41,7 @@ class MessagingTestCase(unittest.TestCase): .format(n=sim.feed.data_buffer.pending_messages())) - def dtest_merged_to_client(self): + def test_merged_to_client(self): """ 2 datasources -> feed -> 2 moving average transforms -> transform merge -> testclient verify message count at client. @@ -60,7 +60,7 @@ class MessagingTestCase(unittest.TestCase): self.assertEqual(sim.feed.data_buffer.pending_messages(), 0, "The feed should be drained of all messages.") - def test_zerror_in_feed(self): + def dtest_error_in_feed(self): ret1 = RandomEquityTrades(133, "ret1", 400) ret2 = RandomEquityTrades(134, "ret2", 400) sources = {"ret1":ret1, "ret2":ret2}