Merge pull request #80 from sideshowdave7/master

EventMQ 0.3

9 files changed, 42 insertions, 31 deletions

diff --git a/bin/send_msg b/bin/send_msg
index b0ea160..19e88de 100755
--- a/bin/send_msg
+++ b/bin/send_msg
@@ -26,7 +26,7 @@ if __name__ == "__main__":
             'kwargs': {}
             }]
 
-        send_request(s, msg, guarantee=True, reply_requested=True, timeout=10)
+        send_request(s, msg, guarantee=True, reply_requested=True, timeout=1)
         print zmq.POLLOUT
         events = dict(poller.poll(500))
         print events
diff --git a/bin/send_publish b/bin/send_publish
index 140be63..50f430e 100755
--- a/bin/send_publish
+++ b/bin/send_publish
@@ -12,13 +12,5 @@ if __name__ == "__main__":
     s.connect(sys.argv[1])
 
     msg = ['update', '1']
-    j = 0
-    while True:
-
-        for i in xrange(10):
-            topic = str(i)
-            send_publish_request(s, topic, msg)
-
-
-        print j
-        j += 1
+    topic = str('topic')
+    send_publish_request(s, topic, msg)
diff --git a/eventmq/__init__.py b/eventmq/__init__.py
index 6b5c4b9..5b8029b 100644
--- a/eventmq/__init__.py
+++ b/eventmq/__init__.py
@@ -1,5 +1,5 @@
 __author__ = 'EventMQ Contributors'
-__version__ = '0.3-rc10'
+__version__ = '0.3'
 
 PROTOCOL_VERSION = 'eMQP/1.0'
 
diff --git a/eventmq/constants.py b/eventmq/constants.py
index 8dbf530..17c6130 100644
--- a/eventmq/constants.py
+++ b/eventmq/constants.py
@@ -7,6 +7,7 @@ class STATUS(object):
     connected = 201
     stopping = 300
     stopped = 301
+    running = 400     # Running and accepting jobs actively
 
 
 class CLIENT_TYPE(object):
diff --git a/eventmq/jobmanager.py b/eventmq/jobmanager.py
index 12629ee..ea5af07 100644
--- a/eventmq/jobmanager.py
+++ b/eventmq/jobmanager.py
@@ -27,7 +27,7 @@ import zmq
 
 from eventmq.log import setup_logger
 from . import conf
-from .constants import KBYE
+from .constants import KBYE, STATUS
 from .poller import Poller, POLLIN
 from .sender import Sender
 from .utils.classes import EMQPService, HeartbeatMixin
@@ -37,6 +37,7 @@ from .utils.messages import send_emqp_message as sendmsg
 from .utils.settings import import_settings
 from .worker import MultiprocessWorker as Worker
 
+
 if sys.version[0] == '2':
     import Queue
 else:
@@ -124,9 +125,12 @@ class JobManager(HeartbeatMixin, EMQPService):
         for i in range(0, len(self.workers)):
             self.send_ready()
 
+        self.status = STATUS.running
+
         while True:
             # Clear any workers if it's time to shut down
             if self.received_disconnect:
+                self.status = STATUS.stopping
                 for _ in range(0, len(self.workers)):
                     logger.debug('Requesting worker death...')
                     self.request_queue.put_nowait('DONE')
@@ -253,7 +257,8 @@ class JobManager(HeartbeatMixin, EMQPService):
         in :meth:`self.process_message` as every message is counted as a
         HEARTBEAT
         """
-        self.check_worker_health()
+        if self.status == STATUS.running:
+            self.check_worker_health()
 
     def check_worker_health(self):
         """
@@ -282,10 +287,11 @@ class JobManager(HeartbeatMixin, EMQPService):
 
     def sighup_handler(self, signum, frame):
         logger.info('Caught signal %s' % signum)
-        self.outgoing.rebuild()
         import_settings()
         import_settings(section='jobmanager')
-        self.start(addr=conf.WORKER_ADDR)
+
+        self.should_reset = True
+        self.received_disconnect = True
 
     def sigterm_handler(self, signum, frame):
         logger.info('Shutting down..')
diff --git a/eventmq/tests/test_jobmanager.py b/eventmq/tests/test_jobmanager.py
index d0497bc..db82843 100644
--- a/eventmq/tests/test_jobmanager.py
+++ b/eventmq/tests/test_jobmanager.py
@@ -111,27 +111,18 @@ class TestCase(unittest.TestCase):
         self.assertTrue(jm.received_disconnect, "Did not receive disconnect.")
 
     # Other Tests
-    @mock.patch('eventmq.jobmanager.JobManager.start')
     @mock.patch('eventmq.jobmanager.import_settings')
-    @mock.patch('eventmq.jobmanager.Sender.rebuild')
-    def test_sighup_handler(self, rebuild_mock, import_settings_mock,
-                            start_mock):
+    def test_sighup_handler(self, import_settings_mock):
         jm = jobmanager.JobManager()
 
         jm.sighup_handler(982374, "FRAMEY the frame")
 
-        self.assertTrue(rebuild_mock.called)
-
         # called once for the default settings, once for the jobmanager
         # settings
         self.assertEqual(2, import_settings_mock.call_count)
         # check to see if the last call was called with the jobmanager section
         import_settings_mock.assert_called_with(section='jobmanager')
 
-        start_mock.assert_called_with(
-            addr=conf.WORKER_ADDR,
-        )
-
     @mock.patch('eventmq.jobmanager.sendmsg')
     def test_sigterm_handler(self, sendmsg_mock):
         jm = jobmanager.JobManager()
diff --git a/eventmq/utils/classes.py b/eventmq/utils/classes.py
index d1ca275..39c600b 100644
--- a/eventmq/utils/classes.py
+++ b/eventmq/utils/classes.py
@@ -119,6 +119,7 @@ class EMQPService(object):
         self.poller.register(self.outgoing, poller.POLLIN)
         self.awaiting_startup_ack = False
         self.received_disconnect = False
+        self.should_reset = False
 
         self.status = constants.STATUS.ready
 
@@ -165,7 +166,7 @@ class EMQPService(object):
 
             # When we return, soemthing has gone wrong and try to reconnect
             # unless self.received_disconnect is True
-            if not self.received_disconnect:
+            if not self.received_disconnect or self.should_reset:
                 self.reset()
 
         logger.info('Death.')
diff --git a/eventmq/worker.py b/eventmq/worker.py
index 7729120..8d25c63 100644
--- a/eventmq/worker.py
+++ b/eventmq/worker.py
@@ -21,13 +21,29 @@ from importlib import import_module
 import logging
 from multiprocessing import Process
 import os
-from threading import Thread
+from threading import Event, Thread
 
 from . import conf
 
 logger = logging.getLogger(__name__)
 
 
+class StoppableThread(Thread):
+    """Thread class with a stop() method. The thread itself has to check
+    regularly for the stopped() condition."""
+
+    def __init__(self, target, name=None, args=()):
+        super(StoppableThread, self).__init__(name=name, target=target,
+                                              args=args)
+        self._stop = Event()
+
+    def stop(self):
+        self._stop.set()
+
+    def stopped(self):
+        return self._stop.isSet()
+
+
 class MultiprocessWorker(Process):
     """
     Defines a worker that spans the job in a multiprocessing task
@@ -60,12 +76,13 @@ class MultiprocessWorker(Process):
 
             try:
                 if timeout:
-                    worker_thread = Thread(target=_run,
-                                           args=(payload['params'], ))
+                    worker_thread = StoppableThread(target=_run,
+                                                    args=(payload['params'], ))
                     worker_thread.start()
                     worker_thread.join(timeout)
 
                     if worker_thread.isAlive():
+                        worker_thread.stop()
                         resp['return'] = 'TimeoutError'
                     else:
                         resp['return'] = 'DONE'
@@ -80,6 +97,9 @@ class MultiprocessWorker(Process):
             if self.job_count > conf.MAX_JOB_COUNT:
                 break
 
+            if resp['return'] == 'TimeoutError':
+                break
+
         logger.debug("Worker death, PID: {}".format(os.getpid()))
 
 
diff --git a/setup.py b/setup.py
index 0419294..24a364b 100644
--- a/setup.py
+++ b/setup.py
@@ -7,7 +7,7 @@ from setuptools import find_packages, setup
 
 setup(
     name='eventmq',
-    version='0.3-rc10',
+    version='0.3',
     description='EventMQ messaging system based on ZeroMQ',
     packages=find_packages(),
     install_requires=['pyzmq==15.4.0',