Repository: liampauling/flumine
Pull request #300 (2020-10-26 19:05:31)

graceful worker shutdown added


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index e985ddb..604e15e 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,20 +3,6 @@
 Release History
 ---------------
 
-1.14.1 (2020-10-29)
-+++++++++++++++++++
-
-**Improvements**
-
-- #297 add violation msg to order on violation
-- Graceful worker shutdown
-- Terminate worker example added
-
-**Libraries**
-
-- betfairlightweight upgraded to 2.9.2
-- python-json-logger upgraded to 2.0.1
-
 1.14.0 (2020-10-12)
 +++++++++++++++++++
 
diff --git a/flumine/__version__.py b/flumine/__version__.py
index d8848ae..303d19c 100644
--- a/flumine/__version__.py
+++ b/flumine/__version__.py
@@ -1,6 +1,6 @@
 __title__ = "flumine"
 __description__ = "Betfair trading framework"
 __url__ = "https://github.com/liampauling/flumine"
-__version__ = "1.14.1"
+__version__ = "1.14.0"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/baseflumine.py b/flumine/baseflumine.py
index ef3c260..555fe83 100644
--- a/flumine/baseflumine.py
+++ b/flumine/baseflumine.py
@@ -370,7 +370,9 @@ class BaseFlumine:
         # shutdown thread pools
         self.simulated_execution.shutdown()
         self.betfair_execution.shutdown()
-        # todo shutdown workers
+        # shutdown workers
+        for w in self._workers:
+            w.shutdown()
         # shutdown logging controls
         self.log_control(events.TerminationEvent(None))
         for c in self._logging_controls:
diff --git a/flumine/worker.py b/flumine/worker.py
index 69143a9..892d786 100644
--- a/flumine/worker.py
+++ b/flumine/worker.py
@@ -33,6 +33,7 @@ class BackgroundWorker(threading.Thread):
         self.func_kwargs = func_kwargs if func_kwargs is not None else {}
         self.start_delay = start_delay
         self.context = context or {}
+        self._running = False
 
     def run(self) -> None:
         logger.info(
@@ -48,7 +49,8 @@ class BackgroundWorker(threading.Thread):
             },
         )
         time.sleep(self.start_delay)
-        while self.is_alive():
+        self._running = True
+        while self._running:
             logger.debug(
                 "BackgroundWorker {0} executing".format(self.name),
                 extra={
@@ -73,6 +75,17 @@ class BackgroundWorker(threading.Thread):
                 )
             time.sleep(self.interval)
 
+    def shutdown(self, timeout: int = 4) -> None:
+        logger.info(
+            "BackgroundWorker {0} shutting down".format(self.name),
+            extra={
+                "worker_name": self.name,
+                "function": self.function,
+            },
+        )
+        self._running = False
+        self.join(timeout)
+
 
 def keep_alive(context: dict, flumine) -> None:
     """Attempt keep alive if required or
diff --git a/requirements.txt b/requirements.txt
index 9a938fb..6c09bdd 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,4 @@
-betfairlightweight==2.9.2
+betfairlightweight==2.9.0
 tenacity==5.0.3
-python-json-logger==2.0.1
+python-json-logger==2.0.0
 requests


Test patch:
diff --git a/tests/test_worker.py b/tests/test_worker.py
index d0e42ba..31dace9 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -13,26 +13,34 @@ class BackgroundWorkerTest(unittest.TestCase):
         self.worker = worker.BackgroundWorker(
             self.mock_flumine,
             self.mock_function,
-            123,
+            0,
             (1, 2),
             {"hello": "world"},
-            5,
+            0,
             {1: 2},
         )
 
     def test_init(self):
-        self.assertEqual(self.worker.interval, 123)
+        self.assertEqual(self.worker.interval, 0)
         self.assertEqual(self.worker.function, self.mock_function)
         self.assertEqual(self.worker.flumine, self.mock_flumine)
         self.assertEqual(self.worker.func_args, (1, 2))
         self.assertEqual(self.worker.func_kwargs, {"hello": "world"})
-        self.assertEqual(self.worker.start_delay, 5)
+        self.assertEqual(self.worker.start_delay, 0)
         self.assertEqual(self.worker.context, {1: 2})
         self.assertEqual(self.worker.name, "test")
+        self.assertFalse(self.worker._running)
 
     # def test_run(self):
     #     self.worker.run()
 
+    def test_shutdown(self):
+        self.worker.start()
+        self.assertTrue(self.worker.is_alive())
+        self.worker.shutdown()
+        self.assertFalse(self.worker._running)
+        self.assertFalse(self.worker.is_alive())
+
 
 class WorkersTest(unittest.TestCase):
     def setUp(self) -> None: