Merge pull request #53 from sideshowdave7/master

Backport 0.4 handling of workers after death

2 files changed, 10 insertions, 9 deletions

diff --git a/bin/send_msg b/bin/send_msg
index c3e88fa..33cfb1a 100755
--- a/bin/send_msg
+++ b/bin/send_msg
@@ -22,7 +22,7 @@ if __name__ == "__main__":
             'callable': 'work_job',
             'class_args': ('blurp',),
             'class_kwargs': {'kwarg1': True},
-            'args': (10, ),
+            'args': (1, ),
             'kwargs': {}
             }]
 
diff --git a/eventmq/jobmanager.py b/eventmq/jobmanager.py
index c904d99..c900495 100644
--- a/eventmq/jobmanager.py
+++ b/eventmq/jobmanager.py
@@ -146,14 +146,6 @@ class JobManager(HeartbeatMixin, EMQPService):
         Starts the actual event loop. Usually called by :meth:`start`
         """
         # Acknowledgment has come
-        # When the job manager unexpectedly disconnects from the router and
-        # reconnects it needs to send a ready for each previously available
-        # worker.
-        # Send a READY for each previously available worker
-        if hasattr(self, '_workers'):
-            for _ in self._workers:
-                self.send_ready()
-
         self.status = STATUS.running
 
         try:
@@ -218,6 +210,15 @@ class JobManager(HeartbeatMixin, EMQPService):
         except Exception:
             logger.exception("Unhandled exception in main jobmanager loop")
 
+        # Cleanup
+        if hasattr(self, '_workers'):
+            del self._workers
+
+        # Flush the queues with workers
+        self.request_queue = mp_queue()
+        self.finished_queue = mp_queue()
+        logger.info("Reached end of event loop")
+
     def handle_response(self, resp):
         """
         Handles a response from a worker process to the jobmanager