Merge pull request #391 from michielbaird/watcher_changes

Ensure pending watches are not dropped when connection is lost

Author: bbangert

kazoo/client.py

@@ -44,8 +44,11 @@
     Sync,
     Transaction
 )
+from kazoo.protocol.states import Callback
+from kazoo.protocol.states import EventType
 from kazoo.protocol.states import KazooState
 from kazoo.protocol.states import KeeperState
+from kazoo.protocol.states import WatchedEvent
 from kazoo.retry import KazooRetry
 from kazoo.security import ACL
 from kazoo.security import OPEN_ACL_UNSAFE
@@ -192,7 +195,8 @@ def __init__(self, hosts='127.0.0.1:2181',
         self._state = KeeperState.CLOSED
         self.state = KazooState.LOST
         self.state_listeners = set()
-
+        self._child_watchers = defaultdict(set)
+        self._data_watchers = defaultdict(set)
         self._reset()
         self.read_only = read_only
 
@@ -309,9 +313,20 @@ def _reset(self):
         self._protocol_version = None
 
     def _reset_watchers(self):
+        watchers = []
+        for child_watchers in six.itervalues(self._child_watchers):
+            watchers.extend(child_watchers)
+
+        for data_watchers in six.itervalues(self._data_watchers):
+            watchers.extend(data_watchers)
+
         self._child_watchers = defaultdict(set)
         self._data_watchers = defaultdict(set)
 
+        ev = WatchedEvent(EventType.NONE, self._state, None)
+        for watch in watchers:
+            self.handler.dispatch_callback(Callback("watch", watch, (ev,)))
+
     def _reset_session(self):
         self._session_id = None
         self._session_passwd = b'\x00' * 16

kazoo/protocol/states.py

@@ -93,13 +93,19 @@ class EventType(object):
         removed). This event does not indicate the data for a child
         node has changed, which must have its own watch established.
 
+    .. attribute:: NONE
+
+        The connection state has been altered.
+
     """
     CREATED = 'CREATED'
     DELETED = 'DELETED'
     CHANGED = 'CHANGED'
     CHILD = 'CHILD'
+    NONE = 'NONE'
 
 EVENT_TYPE_MAP = {
+    -1:EventType.NONE,
     1: EventType.CREATED,
     2: EventType.DELETED,
     3: EventType.CHANGED,

kazoo/recipe/watchers.py

@@ -345,7 +345,8 @@ def _get_children(self, event=None):
                 raise
 
     def _watcher(self, event):
-        self._get_children(event)
+        if event.type != "NONE":
+            self._get_children(event)
 
     def _session_watcher(self, state):
         if state in (KazooState.LOST, KazooState.SUSPENDED):

kazoo/tests/test_client.py

@@ -414,6 +414,22 @@ def test_closed(self):
             client._state = oldstate
             client._connection._write_sock = None
 
+    def test_watch_trigger_expire(self):
+        client = self.client
+        cv = self.make_event()
+
+        client.create("/test", b"")
+
+        def test_watch(event):
+            cv.set()
+
+        client.get("/test/", watch=test_watch)
+        self.expire_session(self.make_event)
+
+
+        cv.wait(3)
+        assert cv.is_set()
+
 
 class TestClient(KazooTestCase):
     def _makeOne(self, *args):