fix(recipe): No more memory leak when ChildrenWatch was stopped (#543)

This ensures that the watcher is removed from the client listener when the func given to ChildrenWatch returns False.
Previously, the watcher was never removed so the ChildrenWatch object would endlessly grow in memory. A unit test is added to ensure this case never happen again.

Fix #542

kazoo/recipe/watchers.py

@@ -341,6 +341,8 @@ def _get_children(self, event=None):
                 if result is False:
                     self._stopped = True
                     self._func = None
+                    if self._allow_session_lost:
+                        self._client.remove_listener(self._session_watcher)
             except Exception as exc:
                 log.exception(exc)
                 raise

kazoo/tests/test_watchers.py

@@ -418,6 +418,40 @@ def changed(children):
         update.wait(0.5)
         eq_(all_children, ['smith'])
 
+    def test_child_watcher_remove_session_watcher(self):
+        update = threading.Event()
+        all_children = ['fred']
+
+        fail_through = []
+
+        def changed(children):
+            while all_children:
+                all_children.pop()
+            all_children.extend(children)
+            update.set()
+            if fail_through:
+                return False
+
+        children_watch = self.client.ChildrenWatch(self.path, changed)
+        session_watcher = children_watch._session_watcher
+
+        update.wait(10)
+        eq_(session_watcher in self.client.state_listeners, True)
+        eq_(all_children, [])
+        update.clear()
+
+        fail_through.append(True)
+        self.client.create(self.path + '/' + 'smith')
+        update.wait(10)
+        eq_(session_watcher not in self.client.state_listeners, True)
+        eq_(all_children, ['smith'])
+        update.clear()
+
+        self.client.create(self.path + '/' + 'george')
+        update.wait(10)
+        eq_(session_watcher not in self.client.state_listeners, True)
+        eq_(all_children, ['smith'])
+
     def test_child_watch_session_loss(self):
         update = threading.Event()
         all_children = ['fred']