diff --git a/qsim/core.py b/qsim/core.py index 1c5415a7..452b27ed 100644 --- a/qsim/core.py +++ b/qsim/core.py @@ -5,6 +5,7 @@ import multiprocessing import zmq import json import copy +import threading import qsim.util as qutil import qsim.messaging as qmsg @@ -32,7 +33,7 @@ class Simulator(object): self.merge_address = "tcp://127.0.0.1:{port}".format(port=10103) self.result_address = "tcp://127.0.0.1:{port}".format(port=10104) - def launch(self): + def simulate(self): self.feed = DataFeed(self.sources.keys(), self.data_address, self.feed_address, qmsg.Sync(self,"DataFeed")) self.launch_component("DataFeed", self.feed) for name, data_source in self.sources.iteritems(): @@ -66,8 +67,14 @@ class Simulator(object): self.sync_components() client_proc.join() #wait for client to complete processing - + def launch_component(self, name, component): + qutil.LOGGER.info("starting {name}".format(name=name)) + thread = threading.Thread(target=component.run) + thread.start() + return thread + + def launch_component_proc(self, name, component): qutil.LOGGER.info("starting {name}".format(name=name)) proc = multiprocessing.Process(target=component.run) proc.start() diff --git a/qsim/test/test_messaging.py b/qsim/test/test_messaging.py index 20b2f062..22ab6a6f 100644 --- a/qsim/test/test_messaging.py +++ b/qsim/test/test_messaging.py @@ -30,7 +30,7 @@ class MessagingTestCase(unittest.TestCase): sources = {"ret1":ret1, "ret2":ret2} client = TestClient(self, expected_msg_count=800) sim = Simulator(sources, {}, client) - sim.launch() + sim.simulate() self.assertEqual(sim.feed.data_buffer.pending_messages(), 0, "The feed should be drained of all messages, found {n} remaining." @@ -51,7 +51,7 @@ class MessagingTestCase(unittest.TestCase): transforms = {"mavg1":mavg1, "mavg2":mavg2} client = TestClient(self, expected_msg_count=800) sim = Simulator(sources, transforms, client) - sim.launch() + sim.simulate() self.assertEqual(sim.feed.data_buffer.pending_messages(), 0, "The feed should be drained of all messages.")