diff --git a/CHANGELOG.md b/CHANGELOG.md index bcb2b0e447b00d7f1da82b15eb57a189fa5a4ea6..4e6d2269f56ad25431bd4f12161a0c2872bbe445 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1 +1,3 @@ # v2.14.42 + +* added `terminating` property to `TerminableThread` diff --git a/satella/__init__.py b/satella/__init__.py index 291cc51dcfca2fb33db89ec6689d1653b1c63d0b..5a7b59359371caea46c32af6d8e2cbfcb22e0142 100644 --- a/satella/__init__.py +++ b/satella/__init__.py @@ -1 +1 @@ -__version__ = '2.14.42a1' +__version__ = '2.14.42a2' diff --git a/satella/coding/concurrent/thread.py b/satella/coding/concurrent/thread.py index 0158013d2f8b934d0795531229789a7b84e81f27..94521e15ccdd9f00dbbb0c07a832a4cebe871902 100644 --- a/satella/coding/concurrent/thread.py +++ b/satella/coding/concurrent/thread.py @@ -185,7 +185,7 @@ class TerminableThread(threading.Thread): >>> a = MeGrimlock().start() >>> a.terminate().join() - Flag whether to terminate is stored in **self.terminating**. + Property to check whether to terminate is stored in **self.terminating**. If you decide to override run(), you got to check periodically for **self._terminating** to become true. If it's true, then a termination request was received, and the thread should @@ -306,6 +306,13 @@ class TerminableThread(threading.Thread): if self._terminating: raise SystemExit() + @property + def terminating(self) -> bool: + """ + :return: Is this thread either alive and trying to terminate or dead and after termination? + """ + return self._terminating + def terminate(self, force: bool = False) -> 'TerminableThread': """ Signal this thread to terminate. diff --git a/tests/test_coding/test_concurrent.py b/tests/test_coding/test_concurrent.py index c3bc04f8f3896e7b6a3f04e90083be7ac246cda8..a12f0906b8a5e3e3183b6300b2309c0143278955 100644 --- a/tests/test_coding/test_concurrent.py +++ b/tests/test_coding/test_concurrent.py @@ -451,6 +451,7 @@ class TestConcurrent(unittest.TestCase): self.assertTrue(t.is_alive()) dct['a'] = True time.sleep(1) + self.assertTrue(t.terminating) self.assertFalse(t.is_alive()) def test_cg_proforma(self):