diff --git a/data/feed.py b/data/feed.py index 7e0865dc..6c9a63e1 100644 --- a/data/feed.py +++ b/data/feed.py @@ -38,9 +38,10 @@ class DataFeed(object): self.syncservice = self.context.socket(zmq.REP) self.syncservice.bind(self.sync_address) - subscribers = 0 - while subscribers < (self.subscriber_count + len(self.data_workers)): - self.logger.info("sync'ing {count}".format(count=subscribers)) + subscribers = 1 + total = self.subscriber_count + len(self.data_workers) + while subscribers <= total: + self.logger.info("sync'ing {count} of {total}".format(count=subscribers, total=total)) # wait for synchronization request msg = self.syncservice.recv() # send synchronization reply diff --git a/data/transforms.py b/data/transforms.py index 37d828ba..9300af69 100644 --- a/data/transforms.py +++ b/data/transforms.py @@ -14,9 +14,11 @@ class Transform(object): feed_address - zmq socket address, Transform will CONNECT a PULL socket and receive messages until "DONE" is received. result_address - zmq socket address, Transform will CONNECT a PUSH socket and send messaes until feed_socket receives "DONE" sync_address - zmq socket address, Transform will CONNECT a REQ socket and send/receive one message before entering feed loop - config - must be a config.Config object with at least an entry for 'name':string value + config - must be a dict that can be wrapped in a config.Config object with at least an entry for 'name':string value + server - if True, transform will bind to the result address (and act as a server), if False it will connect. The + the last transform in a series should be server=True so that clients can connect. """ - self.logger = logging.getLogger() + self.logger = logging.getLogger() self.feed_address = feed_address self.result_address = result_address self.sync_address = sync_address @@ -55,9 +57,11 @@ class Transform(object): message = self.feed_socket.recv() self.received_count += 1 if(message == "DONE"): + self.result_socket.send("DONE") break; event = json.loads(message) cur_state = self.update(event) + cur_state['dt'] = event['dt'] self.result_socket.send(json.dumps(cur_state)) self.sent_count += 1 @@ -106,7 +110,7 @@ class Merge(Transform): if(self.transform_socket == None): #create the feed PULL. self.transform_socket = self.context.socket(zmq.PULL) - self.transform_socket.connect(self.transform_address) + self.transform_socket.bind(self.transform_address) def update(self, event): @@ -125,7 +129,7 @@ class Merge(Transform): class MovingAverage(Transform): - def __init__(self, feed_address, result_address, sync_address, props): + def __init__(self, feed_address, result_address, sync_address, props, server=False): Transform.__init__(self, feed_address, result_address, sync_address, props) self.events = [] @@ -155,4 +159,6 @@ class MovingAverage(Transform): self.average = total/len(self.events) - self.state['avg'] = self.average \ No newline at end of file + self.state['avg'] = self.average + + return self.state \ No newline at end of file diff --git a/qbt_client.py b/qbt_client.py index ad6e440a..1af88ecc 100644 --- a/qbt_client.py +++ b/qbt_client.py @@ -8,10 +8,11 @@ from backtest.util import * class TestClient(object): - def __init__(self,feed_address, sync_address): + def __init__(self,feed_address, sync_address, bind=False): self.logger = logging.getLogger() self.feed_address = feed_address self.sync_address = sync_address + self.bind = bind def run(self): @@ -19,7 +20,14 @@ class TestClient(object): self.context = zmq.Context() self.data_feed = self.context.socket(zmq.PULL) - self.data_feed.connect(self.feed_address) + + if(self.bind): + self.logger.info("binding to {feed_address}".format(feed_address=self.feed_address)) + self.data_feed.bind(self.feed_address) + else: + self.data_feed.connect(self.feed_address) + + self.logger.info("synchronizing with data feed") #synchronize with feed sync_socket = self.context.socket(zmq.REQ) @@ -30,6 +38,8 @@ class TestClient(object): sync_socket.recv() sync_socket.close() + self.logger.info("Starting the client loop") + counter = 0 prev_dt = None while True: diff --git a/test.py b/test.py index 8678a29b..7d3c2d7b 100644 --- a/test.py +++ b/test.py @@ -11,7 +11,7 @@ from qbt_client import TestClient def datafeed(): connection, db = connect_db() logger = logging.getLogger() - feed = DataFeed(db, 1) #one moving average, one client + feed = DataFeed(db, 2) #one moving average, one client feed_proc = multiprocessing.Process(target=feed.run) feed_proc.start() @@ -22,16 +22,22 @@ def datafeed(): result_address = "tcp://127.0.0.1:20202" mavg = MovingAverage(feed.feed_address, result_address, feed.sync_address, config['transforms'][0]) - mavg.run() - #mavg_proc = multiprocessing.Process(target=mavg.run()) - #mavg_proc.start() + mavg_proc = multiprocessing.Process(target=mavg.run) + logger.info("about to launch moving average") + mavg_proc.start() #merger = Merge(feed.feed_address, result_address, feed.sync_address, config) #merger_proc = multiprocessing.Process(target=merger.run) #merger_proc.start() + + #subscribe a client directly to the consolidated feed #client = TestClient(feed.feed_address, feed.sync_address) - #client.run() + + logger.info("starting the client") + #subscribe a client to the transformed feed + client = TestClient(result_address, feed.sync_address, bind=True) + client.run() logger.info("feed has {pending} messages".format(pending=feed.pending_messages())) assert(feed.pending_messages() == 0)