Merge pull request #56 from com4/jobmanager-graceful-shutdown0.2.7

Implement cleanup for SIGTERM, SIGQUIT, & SIGINT

3 files changed, 25 insertions, 2 deletions

diff --git a/eventmq/jobmanager.py b/eventmq/jobmanager.py
index f3ce5e9..9c696ff 100644
--- a/eventmq/jobmanager.py
+++ b/eventmq/jobmanager.py
@@ -79,6 +79,9 @@ class JobManager(HeartbeatMixin, EMQPService):
         if not kwargs.pop('skip_signal', False):
             # handle any sighups by reloading config
             signal.signal(signal.SIGHUP, self.sighup_handler)
+            signal.signal(signal.SIGTERM, self.sigterm_handler)
+            signal.signal(signal.SIGINT, self.sigterm_handler)
+            signal.signal(signal.SIGQUIT, self.sigterm_handler)
 
         #: JobManager starts out by INFORMing the router of it's existence,
         #: then telling the router that it is READY. The reply will be the unit
@@ -232,6 +235,13 @@ class JobManager(HeartbeatMixin, EMQPService):
         import_settings(section='jobmanager')
         self.start(addr=conf.WORKER_ADDR)
 
+    def sigterm_handler(self, signum, frame):
+        logger.info('Shutting down..')
+        sendmsg(self.outgoing, KBYE)
+
+        self.awaiting_startup_ack = False
+        self.received_disconnect = True
+
     def jobmanager_main(self, broker_addr=None):
         """
         Kick off jobmanager with logging and settings import
diff --git a/eventmq/tests/test_jobmanager.py b/eventmq/tests/test_jobmanager.py
index 370eb88..4c0d7ae 100644
--- a/eventmq/tests/test_jobmanager.py
+++ b/eventmq/tests/test_jobmanager.py
@@ -134,6 +134,16 @@ class TestCase(unittest.TestCase):
             addr=conf.WORKER_ADDR,
         )
 
+    @mock.patch('eventmq.jobmanager.sendmsg')
+    def test_sigterm_handler(self, sendmsg_mock):
+        jm = jobmanager.JobManager()
+
+        jm.sigterm_handler(13231, "FRAMEY the evil frame")
+
+        sendmsg_mock.assert_called_with(jm.outgoing, constants.KBYE)
+        self.assertFalse(jm.awaiting_startup_ack)
+        self.assertTrue(jm.received_disconnect)
+
     @mock.patch('eventmq.jobmanager.JobManager.start')
     @mock.patch('eventmq.jobmanager.import_settings')
     def test_jobmanager_main(self, import_settings_mock, start_mock):
diff --git a/eventmq/utils/classes.py b/eventmq/utils/classes.py
index be38e80..3cc3b49 100644
--- a/eventmq/utils/classes.py
+++ b/eventmq/utils/classes.py
@@ -157,8 +157,11 @@ class EMQPService(object):
                         self.process_message(msg)
 
             self.status = constants.STATUS.connected
-            logger.info('Starting event loop...')
-            self._start_event_loop()
+
+            if not self.received_disconnect:
+                logger.info('Starting event loop...')
+                self._start_event_loop()
+
             # When we return, soemthing has gone wrong and try to reconnect
             # unless self.received_disconnect is True
             if not self.received_disconnect: