fixed exit logic to use destroy

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):
         """

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()
         

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):

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})."

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}