Merge remote-tracking branch 'com4/master'

5 files changed, 19 insertions, 17 deletions

diff --git a/bin/send_msg b/bin/send_msg
index 33cfb1a..c3e88fa 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': (1, ),
+            'args': (10, ),
             'kwargs': {}
             }]
 
diff --git a/eventmq/__init__.py b/eventmq/__init__.py
index a02a091..ef87218 100644
--- a/eventmq/__init__.py
+++ b/eventmq/__init__.py
@@ -1,5 +1,5 @@
 __author__ = 'EventMQ Contributors'
-__version__ = '0.3.4.9'
+__version__ = '0.3.4.11'
 
 PROTOCOL_VERSION = 'eMQP/1.0'
 
diff --git a/eventmq/jobmanager.py b/eventmq/jobmanager.py
index c900495..78a080b 100644
--- a/eventmq/jobmanager.py
+++ b/eventmq/jobmanager.py
@@ -20,7 +20,7 @@ Ensures things about jobs and spawns the actual tasks
 from json import dumps as serializer, loads as deserializer
 
 import logging
-from multiprocessing import Queue as mp_queue
+from multiprocessing import Manager as MPManager
 import os
 import signal
 import sys
@@ -121,8 +121,9 @@ class JobManager(HeartbeatMixin, EMQPService):
         self.pid_distribution = {}
 
         #: Setup worker queues
-        self.request_queue = mp_queue()
-        self.finished_queue = mp_queue()
+        self._mp_manager = MPManager()
+        self.request_queue = self._mp_manager.Queue()
+        self.finished_queue = self._mp_manager.Queue()
         self._setup()
 
     def handle_pdb(self, sig, frame):
@@ -146,6 +147,14 @@ 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:
@@ -210,15 +219,6 @@ 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
diff --git a/eventmq/log.py b/eventmq/log.py
index 1864c8d..3d62165 100644
--- a/eventmq/log.py
+++ b/eventmq/log.py
@@ -27,9 +27,11 @@ import zmq.log.handlers
 
 
 FORMAT_STANDARD = logging.Formatter(
-    '%(asctime)s - %(name)s  %(levelname)s - %(message)s')
+    '%(asctime)s - %(name)s  %(levelname)s - %(message)s',
+    datefmt='%Y-%m-%dT%H:%M:%S%z')
 FORMAT_NAMELESS = logging.Formatter(
-    '%(asctime)s - %(levelname)s - %(message)s')
+    '%(asctime)s - %(levelname)s - %(message)s',
+    datefmt='%Y-%m-%dT%H:%M:%S%z')
 
 
 class PUBHandler(zmq.log.handlers.PUBHandler):
diff --git a/setup.py b/setup.py
index 553c86b..6faa4c1 100644
--- a/setup.py
+++ b/setup.py
@@ -7,7 +7,7 @@ from setuptools import find_packages, setup
 
 setup(
     name='eventmq',
-    version='0.3.4.9',
+    version='0.3.4.11',
     description='EventMQ job execution and messaging system based on ZeroMQ',
     packages=find_packages(),
     install_requires=['pyzmq==15.4.0',