diff --git a/CHANGELOG.md b/CHANGELOG.md index a7805eaca8a53add3889d705b0e58944572cf412..705bbf46f29930cae94c353d72121408278e9417 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ * 0.92: - * TBA + * Added `on_fail` event handler - fired upon connection loss * v0.91: * removed annoying warnings diff --git a/coolamqp/clustering/cluster.py b/coolamqp/clustering/cluster.py index 655a676f998e1e400cf0846f868d97f50f4c77a2..f5216f3cadb1305ff63de1091ab8e9eacec9ee61 100644 --- a/coolamqp/clustering/cluster.py +++ b/coolamqp/clustering/cluster.py @@ -39,7 +39,8 @@ class Cluster(object): """ :param nodes: list of nodes, or a single node. For now, only one is supported. :type nodes: NodeDefinition instance or a list of NodeDefinition instances - :param on_fail: callable/0 to call when connection fails. This is a one-shot + :param on_fail: callable/0 to call when connection fails in an + unclean way. This is a one-shot :type on_fail: callable/0 """ from coolamqp.objects import NodeDefinition @@ -175,7 +176,12 @@ class Cluster(object): self.snr = SingleNodeReconnector(self.node, self.attache_group, self.listener) self.snr.on_fail.add(lambda: self.events.put_nowait(ConnectionLost())) if self.on_fail is not None: - self.snr.on_fail.add(self.on_fail) + + def nice(): + if not self.snr.terminating: + self.on_fail() + + self.snr.on_fail.add(nice) # Spawn a transactional publisher and a noack publisher self.pub_tr = Publisher(Publisher.MODE_CNPUB) diff --git a/tests/test_clustering/test_things.py b/tests/test_clustering/test_things.py index 27a9ac4a85fc100922d30037e84c901aaa55a1ad..51cf2ff28e136970f97a19d9e79cd58da33df375 100644 --- a/tests/test_clustering/test_things.py +++ b/tests/test_clustering/test_things.py @@ -15,13 +15,22 @@ logging.basicConfig(level=logging.DEBUG) class TestConnecting(unittest.TestCase): def test_on_fail(self): - q = {} + q = {'failed': False} c = Cluster([NodeDefinition('127.0.0.1', 'xguest', 'xguest', heartbeat=20)], on_fail=lambda: q.update(failed=True)) c.start() time.sleep(5) c.shutdown() self.assertTrue(q['failed']) + def test_on_clean(self): + q = {} + c = Cluster([NODE], on_fail=lambda: q.update(failed=True)) + c.start() + time.sleep(5) + c.shutdown() + time.sleep(5) + self.assertFalse(q['failed']) + def test_start_called_multiple_times(self): c = Cluster([NODE]) c.start(wait=True)