From a5b914c53f402a4616da1cd7339da24c29725aab Mon Sep 17 00:00:00 2001
From: Ashley Winters <ashley.winters@gmail.com>
Date: Thu, 15 Jan 2015 18:41:53 +0000
Subject: [PATCH] Add sync() function to enable consistent cross-client views

Per the ZooKeeper documentation, "ZooKeeper by itself doesn't guarantee
that changes occur synchronously across all servers". In order to
ensure a write that was committed by another client is visible to
another client, it's necessary to have the sync() function from the
ZooKeeper API available.
---
 README.md                   | 40 +++++++++++++++++++++++++++++++++++++
 index.js                    | 36 +++++++++++++++++++++++++++++++++
 lib/ConnectionManager.js    |  3 +++
 lib/jute/specification.json |  6 ++++++
 4 files changed, 85 insertions(+)

diff --git a/README.md b/README.md
index a12765e..8a45e9a 100644
--- a/README.md
+++ b/README.md
@@ -413,6 +413,46 @@ zookeeper.setData('/test/demo', null, 2, function (error, stat) {
 
 ---
 
+#### void sync(path, callback)
+
+Synchronize the client's view of the given path with the ZooKeeper leader.
+
+It is possible for a client to read a stale value for a given path, even
+after another client has received confirmation that the new value has been
+committed. Only a quorum of servers has to agree to a given change for a
+write to be confirmed, and we might be connected to a server that hasn't
+seen the update yet.
+
+The callback function will be called once we confirm that any read of the
+given path will return a value at least as new as when sync() was called.
+
+**Arguments**
+
+* path `String` - Path of the node.
+* callback(error) `Function` - The callback function.
+
+**Example**
+
+```javascript
+zookeeper.sync('/test/demo', function (error) {
+    if (error) {
+        console.log(error.stack);
+        return;
+    }
+
+    zookeeper.getData('/test/demo', function (error, data, stat) {
+        if (error) {
+            console.log(error.stack);
+            return;
+        }
+
+        console.log('Fresh value read.');
+    });
+});
+```
+
+---
+
 #### void getACL(path, callback)
 
 Retrieve the list of [ACL](#acl) and stat of the node of the given path.
diff --git a/index.js b/index.js
index 49bc5bd..16e41b9 100644
--- a/index.js
+++ b/index.js
@@ -422,6 +422,42 @@ Client.prototype.create = function (path, data, acls, mode, callback) {
     );
 };
 
+/**
+ * Synchronize this client's view of the given path with the
+ * ZooKeeper leader. From the callback, any read will see
+ * data at least as new as when this function was called.
+ *
+ * @method sync
+ * @param path {String} The node path.
+ * @param callback {Function} The callback function.
+ */
+Client.prototype.sync = function (path, callback) {
+    Path.validate(path);
+
+    assert(typeof callback === 'function', 'callback must be a function.');
+
+    var self = this,
+        header = new jute.protocol.RequestHeader(),
+        payload = new jute.protocol.SyncRequest(),
+        request;
+
+    header.type = jute.OP_CODES.SYNC;
+
+    payload.path = path;
+
+    request = new jute.Request(header, payload);
+
+    attempt(
+        self,
+        function (attempts, next) {
+            self.connectionManager.queue(request, function (error, response) {
+                next(error);
+            });
+        },
+        callback
+    );
+};
+
 /**
  * Delete a node with the given path. If version is not -1, the request will
  * fail when the provided version does not match the server version.
diff --git a/lib/ConnectionManager.js b/lib/ConnectionManager.js
index 0ae6086..140b18d 100644
--- a/lib/ConnectionManager.js
+++ b/lib/ConnectionManager.js
@@ -528,6 +528,9 @@ ConnectionManager.prototype.onSocketData = function (buffer) {
                 case jute.OP_CODES.GET_DATA:
                     responsePayload = new jute.protocol.GetDataResponse();
                     break;
+                case jute.OP_CODES.SYNC:
+                    responsePayload = new jute.protocol.SyncResponse();
+                    break;
                 case jute.OP_CODES.SET_ACL:
                     responsePayload = new jute.protocol.SetACLResponse();
                     break;
diff --git a/lib/jute/specification.json b/lib/jute/specification.json
index b13feb0..e2c5252 100644
--- a/lib/jute/specification.json
+++ b/lib/jute/specification.json
@@ -94,6 +94,12 @@
             { "name" : "data", "type" : "buffer" },
             { "name" : "stat", "type" : "data.Stat" }
         ],
+        "SyncRequest" : [
+            { "name" : "path", "type" : "ustring" }
+        ],
+        "SyncResponse" : [
+            { "name" : "path", "type" : "ustring" }
+        ],
         "GetACLRequest" : [
             { "name" : "path", "type" : "ustring" }
         ],