[ZOOKEEPER-1416] Persistent Recursive Watch

src/docs/src/documentation/content/xdocs/zookeeperOver.xml

@@ -172,6 +172,11 @@
       connection between the client and one of the Zoo Keeper servers is
       broken, the client will receive a local notification. These can be used
       to <emphasis>[tbd]</emphasis>.</para>
+
+      <para><emphasis role="bold">New in 3.6.0: </emphasis>Clients can also set
+      permanent, recursive watches on a znode that are not removed when triggered
+      and that trigger for changes on the registered znode as well as any children
+      znodes recursively.</para>
     </section>
 
     <section>

src/docs/src/documentation/content/xdocs/zookeeperProgrammers.xml

@@ -192,6 +192,11 @@
         sends the client a notification. More information about watches can be
         found in the section 
 	    <ulink url="#ch_zkWatches">ZooKeeper Watches</ulink>.</para>
+
+        <para><emphasis role="bold">New in 3.6.0: </emphasis>Clients can also set
+        permanent, recursive watches on a znode that are not removed when triggered
+        and that trigger for changes on the registered znode as well as any children
+        znodes recursively.</para>
       </section>
 
       <section>
@@ -712,6 +717,24 @@
       </itemizedlist>
 	</section>
 
+    <section id="sc_PersistentWatches">
+      <title>Persistent Watches</title>
+      <para><emphasis role="bold">New in 3.6.0: </emphasis> There is now a variation on the standard
+      watch described above whereby you can set a watch that does not get removed when triggered.
+      Additionally, these watches trigger the event types <emphasis role="bold">NodeCreated</emphasis>,
+      <emphasis role="bold">NodeDeleted</emphasis>, and <emphasis role="bold">NodeDataChanged</emphasis>
+      and, optionally, recursively for all znodes starting at the znode that the watch is registered for. Note that
+      <emphasis role="bold">NodeChildrenChanged</emphasis> are not triggered for persistent recursive watches as
+      it is redundant.</para>
+
+      <para>Persistent watches are set using the method <emphasis
+      role="bold">addPersistentWatch()</emphasis>. The triggering semantics and guarantees
+      (other than one-time triggering) are the same as standard watches. The only exception regarding events is that
+      recursive persistent watchers never trigger child changed events as they are redundant.
+      Persistent watches are removed using <emphasis role="bold">removeWatches()</emphasis> with watcher type
+      <emphasis role="bold">WatcherType.Any</emphasis>.</para>
+    </section>
+
     <section id="sc_WatchRemoval">
       <title>Remove Watches</title>
       <para>We can remove the watches registered on a znode with a call to 
@@ -730,6 +753,11 @@
           <para><emphasis role="bold">Data Remove event:</emphasis></para>
           <para>Watcher which was added with a call to exists or getData.</para>
         </listitem>
+
+        <listitem>
+          <para><emphasis role="bold">Persistent Remove event:</emphasis></para>
+          <para>Watcher which was added with a call to add a persistent watch.</para>
+        </listitem>
       </itemizedlist>
    </section>
 
@@ -767,15 +795,15 @@
 
       <itemizedlist>
         <listitem>
-          <para>Watches are one time triggers; if you get a watch event and
+          <para>Standard watches are one time triggers; if you get a watch event and
           you want to get notified of future changes, you must set another
           watch.</para>
         </listitem>
       </itemizedlist>
 
       <itemizedlist>
         <listitem>
-          <para>Because watches are one time triggers and there is latency
+          <para>Because standard watches are one time triggers and there is latency
           between getting the event and sending a new request to get a watch
           you cannot reliably see every change that happens to a node in
           ZooKeeper. Be prepared to handle the case where the znode changes

src/java/main/org/apache/zookeeper/ClientCnxn.java

@@ -82,6 +82,7 @@
 import org.apache.zookeeper.proto.SetACLResponse;
 import org.apache.zookeeper.proto.SetDataResponse;
 import org.apache.zookeeper.proto.SetWatches;
+import org.apache.zookeeper.proto.SetWatches2;
 import org.apache.zookeeper.proto.WatcherEvent;
 import org.apache.zookeeper.server.ByteBufferInputStream;
 import org.apache.zookeeper.server.ZooKeeperThread;
@@ -958,18 +959,26 @@ void primeConnection() throws IOException {
                 List<String> dataWatches = zooKeeper.getDataWatches();
                 List<String> existWatches = zooKeeper.getExistWatches();
                 List<String> childWatches = zooKeeper.getChildWatches();
+                List<String> persistentWatches = zooKeeper.getPersistentWatches();
+                List<String> persistentRecursiveWatches = zooKeeper.getPersistentRecursiveWatches();
                 if (!dataWatches.isEmpty()
-                        || !existWatches.isEmpty() || !childWatches.isEmpty()) {
+                        || !existWatches.isEmpty() || !childWatches.isEmpty()
+                        || !persistentWatches.isEmpty() || !persistentRecursiveWatches.isEmpty()) {
                     Iterator<String> dataWatchesIter = prependChroot(dataWatches).iterator();
                     Iterator<String> existWatchesIter = prependChroot(existWatches).iterator();
                     Iterator<String> childWatchesIter = prependChroot(childWatches).iterator();
+                    Iterator<String> persistentWatchesIter = prependChroot(persistentWatches).iterator();
+                    Iterator<String> persistentRecursiveWatchesIter = prependChroot(persistentRecursiveWatches).iterator();
                     long setWatchesLastZxid = lastZxid;
 
                     while (dataWatchesIter.hasNext()
-                           || existWatchesIter.hasNext() || childWatchesIter.hasNext()) {
+                           || existWatchesIter.hasNext() || childWatchesIter.hasNext()
+                           || persistentWatchesIter.hasNext() || persistentRecursiveWatchesIter.hasNext()) {
                         List<String> dataWatchesBatch = new ArrayList<String>();
                         List<String> existWatchesBatch = new ArrayList<String>();
                         List<String> childWatchesBatch = new ArrayList<String>();
+                        List<String> persistentWatchesBatch = new ArrayList<String>();
+                        List<String> persistentRecursiveWatchesBatch = new ArrayList<String>();
                         int batchLength = 0;
 
                         // Note, we may exceed our max length by a bit when we add the last
@@ -985,17 +994,25 @@ void primeConnection() throws IOException {
                             } else if (childWatchesIter.hasNext()) {
                                 watch = childWatchesIter.next();
                                 childWatchesBatch.add(watch);
+                            } else if (persistentWatchesIter.hasNext()) {
+                                watch = persistentWatchesIter.next();
+                                persistentWatchesBatch.add(watch);
+                            } else if (persistentRecursiveWatchesIter.hasNext()) {
+                                watch = persistentRecursiveWatchesIter.next();
+                                persistentRecursiveWatchesBatch.add(watch);
                             } else {
                                 break;
                             }
                             batchLength += watch.length();
                         }
 
-                        SetWatches sw = new SetWatches(setWatchesLastZxid,
+                        SetWatches2 sw = new SetWatches2(setWatchesLastZxid,
                                                        dataWatchesBatch,
                                                        existWatchesBatch,
-                                                       childWatchesBatch);
-                        RequestHeader header = new RequestHeader(-8, OpCode.setWatches);
+                                                       childWatchesBatch,
+                                                       persistentWatchesBatch,
+                                                       persistentRecursiveWatchesBatch);
+                        RequestHeader header = new RequestHeader(-8, OpCode.setWatches2);
                         Packet packet = new Packet(header, new ReplyHeader(), sw, null, null);
                         outgoingQueue.addFirst(packet);
                     }

src/java/main/org/apache/zookeeper/Watcher.java

@@ -125,7 +125,8 @@ public enum EventType {
             NodeDataChanged (3),
             NodeChildrenChanged (4),
             DataWatchRemoved (5),
-            ChildWatchRemoved (6);
+            ChildWatchRemoved (6),
+            PersistentWatchRemoved (7);
 
             private final int intValue;     // Integer representation of value
                                             // for sending over wire
@@ -147,6 +148,7 @@ public static EventType fromInt(int intValue) {
                     case  4: return EventType.NodeChildrenChanged;
                     case  5: return EventType.DataWatchRemoved;
                     case  6: return EventType.ChildWatchRemoved;
+                    case  7: return EventType.PersistentWatchRemoved;
 
                     default:
                         throw new RuntimeException("Invalid integer value for conversion to EventType");

src/java/main/org/apache/zookeeper/ZooDefs.java

@@ -74,12 +74,16 @@ public interface OpCode {
 
         public final int createTTL = 21;
 
+        public final int addPersistentWatch = 22;
+
         public final int auth = 100;
 
         public final int setWatches = 101;
 
         public final int sasl = 102;
 
+        public final int setWatches2 = 103;
+
         public final int createSession = -10;
 
         public final int closeSession = -11;