heartbeats working, however the zipline is now fully sequential.

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)
 

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

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