From ceb90133636b4a74a6d796fcaeacd6c2faccb738 Mon Sep 17 00:00:00 2001
From: fawce <fawce@quantopian.com>
Date: Fri, 13 Jul 2012 14:54:35 -0400
Subject: [PATCH] heartbeats working, however the zipline is now fully
 sequential.

---
 zipline/components/aggregator.py |  1 +
 zipline/core/component.py        |  6 +++---
 zipline/core/monitor.py          | 10 +++++++---
 3 files changed, 11 insertions(+), 6 deletions(-)

diff --git a/zipline/components/aggregator.py b/zipline/components/aggregator.py
index e4a6fba4..cc589789 100644
--- a/zipline/components/aggregator.py
+++ b/zipline/components/aggregator.py
@@ -114,6 +114,7 @@ class Aggregate(Component):
         """
         while self.pending_messages() > 0:
             event = self.next()
+            self.heartbeat()
             if event:
                 self.send(event)
 
diff --git a/zipline/core/component.py b/zipline/core/component.py
index b2353cc2..99e56184 100644
--- a/zipline/core/component.py
+++ b/zipline/core/component.py
@@ -334,17 +334,17 @@ class Component(object):
         # Just exit.
         elif self.last_ping and time.time() - self.last_ping > 1:
             # send a ping ahead of schedule
+            pre_pong = time.time()
             heartbeat_frame = CONTROL_FRAME(
                     CONTROL_PROTOCOL.OK,
-                    str(self.last_ping)
+                    str(pre_pong)
                 )
 
             # Echo back the heartbeat identifier to tell the
             # controller that this component is still alive and
             # doing work
             self.control_out.send(heartbeat_frame)
-
-
+            self.last_ping = pre_pong
 
     # ----------------------------
     #  Cleanup & Modes of Failure
diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py
index acf012b3..a9b66b4f 100644
--- a/zipline/core/monitor.py
+++ b/zipline/core/monitor.py
@@ -312,7 +312,7 @@ class Controller(object):
                         if not self.router.getsockopt(self.zmq.RCVMORE):
                             self.handle_recv(buffer[:])
                             buffer = []
-                            checktime = time.time()
+                            #checktime = time.time()
 
                     except INVALID_CONTROL_FRAME:
                         log.error('Invalid frame', rawmessage)
@@ -563,9 +563,13 @@ class Controller(object):
                 # Go to your bosom; knock there, and ask your heart what
                 # it doth know...
                 self.responses.add(identity)
-            elif status < self.ctime:
+            elif float(status) < self.ctime:
                 # False face must hide what the false heart doth know.
-                log.warning('Delayed heartbeat received.')
+                log.warning('Delayed heartbeat received: %s' % msg)
+            elif float(status) > self.ctime:
+                # Pre-emptive heartbeat from the component
+                # log.info("pre-emptive pong: %s" % msg)
+                self.responses.add(identity)
             else:
                 # Otherwise its something weird and we don't know
                 # what to do so just say so, probably line noise