Merge branch 'feature/schedule_helpers' into 0.4

12 files changed, 60 insertions, 43 deletions

diff --git a/README.md b/README.md
index 908d7e1..7c1e84a 100644
--- a/README.md
+++ b/README.md
@@ -1,7 +1,7 @@
 EventMQ
 =======
-[![Circle CI](https://circleci.com/gh/enderlabs/eventmq.svg?style=svg&circle-token=312da6ae260c2302baed268d2e052ce5e81cc71f)](https://circleci.com/gh/enderlabs/eventmq)
-[![Coverage Status](https://coveralls.io/repos/github/enderlabs/eventmq/badge.svg?branch=master)](https://coveralls.io/github/enderlabs/eventmq?branch=master)
+[![CircleCI](https://circleci.com/gh/eventmq/eventmq.svg?style=svg)](https://circleci.com/gh/eventmq/eventmq)
+[![Coverage Status](https://coveralls.io/repos/github/eventmq/eventmq/badge.svg)](https://coveralls.io/github/eventmq/eventmq)
 
 # Overview
 EventMQ is a message passing system built on [ZeroMQ](https://zeromq.org)
@@ -17,14 +17,18 @@ pip install eventmq
 # Support
 ## Documenation
 
-[Documentation](https://enderlabs.github.io/eventmq/)
+[Documentation](https://eventmq.github.io/eventmq/)
 
 ## Mailing Lists
 User Support: http://lists.eventmq.io/listinfo.cgi/eventmq-users-eventmq.io
 
 Development Discussion: http://lists.eventmq.io/listinfo.cgi/eventmq-devel-eventmq.io
 
-## Quick Start
+## IRC
+
+ #eventmq on [irc.freenode.net](https://webchat.freenode.net/?channels=#eventmq)
+
+# Quick Start
 
 my_jerbs.py
 ``` python
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/docs/settings_file.rst b/docs/settings_file.rst
index cd37b6a..17d3b67 100644
--- a/docs/settings_file.rst
+++ b/docs/settings_file.rst
@@ -33,13 +33,14 @@ running. Grouping similar jobs in named queues will help you tune this number.
 
 queues
 ======
-Default: (10, default)
+Default: [[10, "default"]]
 
-Semi-colon seperated list of queues to process jobs for with thier
-weights. Example: ``queues=(10, data_process); (15, email)``.  With these
+Comma seperated list of queues to process jobs for with thier weights. This list
+must be valid JSON otherwise an error will be thrown.
+Example: ``queues=[[10, "data_process"], [15, "email"]]``.  With these
 weights and the ``CONCURRENT_JOBS`` setting, you should be able to tune managers
-running jobs locally pretty efficiently. If you have a larger box with a weight
-of 50 on q1 and 8 concurrent jobs and a smaller box with a weight 30 and 4
+running jobs locally efficiently. If you have a larger server with a weight of
+50 on q1 and 8 concurrent jobs and a smaller server with a weight 30 and 4
 concurrent jobs, the q1 jobs will be sent to the large box until it is no longer
 accepting jobs. At this point jobs will start to be sent to the next highest
 number until the large box is ready to accept another q1 job.
diff --git a/eventmq/__init__.py b/eventmq/__init__.py
index 8ec8043..ca9b517 100644
--- a/eventmq/__init__.py
+++ b/eventmq/__init__.py
@@ -1,5 +1,5 @@
 __author__ = 'EventMQ Contributors'
-__version__ = '0.3-rc10'
+__version__ = '0.3.1'
 
 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/scheduler.py b/eventmq/scheduler.py
index 8e0f5ac..4bac781 100644
--- a/eventmq/scheduler.py
+++ b/eventmq/scheduler.py
@@ -401,7 +401,6 @@ class Scheduler(HeartbeatMixin, EMQPService):
         Returns:
             int: unique hash for the job
         """
-
         # Get the job portion of the message
         msg = deserialize(message[3])[1]
 
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 c5315de..067b352 100644
--- a/setup.py
+++ b/setup.py
@@ -7,8 +7,8 @@ from setuptools import find_packages, setup
 
 setup(
     name='eventmq',
-    version='0.3-rc10',
-    description='EventMQ messaging system based on ZeroMQ',
+    version='0.3.1',
+    description='EventMQ job execution and messaging system based on ZeroMQ',
     packages=find_packages(),
     install_requires=['pyzmq==16.0.2',
                       'six==1.10.0',
@@ -32,7 +32,7 @@ setup(
               'mock==1.3.0'],
           },
     author='EventMQ Contributors',
-    url='https://github.com/enderlabs/eventmq/',
+    url='https://github.com/eventmq/eventmq/',
 
     # See https://pypi.python.org/pypi?%3Aaction=list_classifiers
     classifiers=[
@@ -53,6 +53,8 @@ setup(
         # that you indicate whether you support Python 2, Python 3 or both.
         'Programming Language :: Python :: 2.6',
         'Programming Language :: Python :: 2.7',
+        'Programming Language :: Python :: 3.4',
+        'Programming Language :: Python :: 3.5',
     ],
     scripts=[
         'bin/emq-cli',