client is receiving transformed messages

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

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
+        self.state['avg'] = self.average
+        
+        return self.state

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:

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)