/*
Set the watch in ZK on the node that represents a user
request for a snapshot
969716
void userSnapshotRequestExistenceCheck(boolean deleteExistingRequest) throws Exception {
        if (deleteExistingRequest) {
            m_zk.delete(VoltZK.user_snapshot_request, -1, null, null);
        }
        if (m_zk.exists(VoltZK.user_snapshot_request, m_userSnapshotRequestExistenceWatcher) != null) {
            processUserSnapshotRequestEvent(new WatchedEvent(
                    EventType.NodeCreated,
                    KeeperState.SyncConnected,
                    VoltZK.user_snapshot_request));
        }
    }
1005500
@Nonnull
  public ListenerRegistration addSnapshotListener(
      @Nonnull Executor executor, @Nonnull EventListener<QuerySnapshot> listener) {
    return Watch.forQuery(this).runWatch(executor, listener);
  }
162556
function createRegisterWatcher(curatedDevice) {
	this.watchers = {};

	this.getWatchers = function() {
		var defered = q.defer();
		var watcherKeys = Object.keys(self.watchers);
		defered.resolve(watcherKeys);
		return defered.promise;
	};
	
	var isValidWatcher = function(watcherName) {
		var isValid = true;
		if(typeof(self.watchers[watcherName]) === 'undefined') {
			isValid = false;
			return isValid;
		}
		return isValid;
	};
	// createWatcher is the function that creates a new watcher if one doesn't
	// already exist.
	this.createWatcher = function(watcherName, callback, options) {
		var defered = q.defer();

		var collectionFunction = 'iReadMany';
		if(options) {
			if(options.collectionFunction) {
				collectionFunction = options.collectionFunction;
			}
		}
		// If there aren't any watchers
		if(!isValidWatcher(watcherName)) {
			self.watchers[watcherName] = new createWatcherObject(
				watcherName,
				collectionFunction,
				callback,
				curatedDevice
			);
			defered.resolve(watcherName);
		} else {
			defered.reject({'err': 'Watcher already exists'});
		}
		return defered.promise;
	};

	this.configureWatcher = function(watcherName, newConfig) {
		if(isValidWatcher(watcherName)) {
			return self.watchers[watcherName].configureWatcher(newConfig);
		} else {
			var defered = q.defer();
			defered.reject({'err': 'Watcher does not exist'});
			return defered.promise;
		}
	};

	var deleteWatcher = function(watcherName) {
		var defered = q.defer();
		var statistics = {};
		var keys = Object.keys(self.watchers[watcherName].statistics);
		keys.forEach(function(key) {
			statistics[key] = self.watchers[watcherName].statistics[key];
		});
		self.watchers[watcherName] = null;
		self.watchers[watcherName] = undefined;
		delete self.watchers[watcherName];
		defered.resolve({'name': watcherName, 'statistics': statistics});
		return defered.promise;
	};
	this.stopWatcher = function(watcherName) {
		var defered = q.defer();
		if(isValidWatcher(watcherName)) {
			self.watchers[watcherName].stopWatcher()
			.then(deleteWatcher)
			.then(defered.resolve, defered.reject);
		} else {
			defered.reject({'err': 'Watcher does not exist'});
		}
		return defered.promise;
	};

	this.stopAllWatchers = function() {
		var defered = q.defer();
		var watcherKeys = Object.keys(self.watchers);
		if(watcherKeys.length > 0) {
			var promises = watcherKeys.map(self.stopWatcher);

			q.AllSettled(promises)
			.then(function(results) {
				defered.resolve(watcherKeys);
			});
		} else {
			defered.resolve([]);
		}
		return defered.promise;
	};
	
	var self = this;
}
691910
private void sendSnapshotTxnId(SnapshotInfo toRestore) {
        long txnId = toRestore != null ? toRestore.txnId : 0;
        String jsonData = toRestore != null ? toRestore.toJSONObject().toString() : "{}";
        LOG.debug("Sending snapshot ID " + txnId + " for restore to other nodes");
        try {
            m_zk.create(VoltZK.restore_snapshot_id, jsonData.getBytes(Constants.UTF8ENCODING),
                        Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        } catch (Exception e) {
            VoltDB.crashGlobalVoltDB("Failed to create Zookeeper node: " + e.getMessage(),
                                     false, e);
        }
    }
368988
function _watch(vm, vars, update) {
    var watchKeys = []
    function _handler (kp) {
        if (watchKeys.some(function(key) {
            if (_relative(kp, key)) {
                return true
            }
        })) update.apply(null, arguments)
    }

    if (vars && vars.length) {
        vars.forEach(function (k) {
            if (~keywords.indexOf(k)) return
            while (k) {
                if (!~watchKeys.indexOf(k)) watchKeys.push(k)
                k = util.digest(k)
            }
        })
        if (!watchKeys.length) return noop
        return vm.$watch(_handler)
    }
    return noop
}
1094605
function ZkCxn(urls, timeout) {
  this._locks = {}; // lockname -> lock node
  this._urls = urls;
  this._options = {}; // passed to this._zk on creation
  if (timeout) {
    this._options.sessionTimeout = timeout;
  }
  this._timeout = timeout || DEFAULT_TIMEOUT; // used for onConnection callbacks. TODO: break out to distinct parameter?
  this._cxnState = this.cxnStates.CLOSED; // initial state
  this._stateEmitter = undefined; // created on connect
  this._zk = undefined; // instance of node-zookeeper-client
}
1310637
def ack_watch(self, watch_id, action_id=None, params=None):
        
        if watch_id in SKIP_IN_PATH:
            raise ValueError("Empty value passed for a required argument 'watch_id'.")
        return self.transport.perform_request(
            "PUT",
            _make_path("_watcher", "watch", watch_id, "_ack", action_id),
            params=params,
        )
49246
private void watchFilesTask() {
        while (!watcherExecutor.isShutdown()) {
            try {
                WatchKey key = watchService.take();
                Path parent = watchKeyDirs.get(key);
                if (parent == null) {
                    key.reset();
                    continue;
                }

                Set<Path> updates = new TreeSet<>();
                for (WatchEvent<?> ev : key.pollEvents()) {
                    WatchEvent.Kind<?> kind = ev.kind();

                    // Only file modification, creation, deletion is interesting.
                    if (kind != ENTRY_MODIFY &&
                        kind != ENTRY_CREATE &&
                        kind != ENTRY_DELETE) {
                        if (kind == OVERFLOW) {
                            LOGGER.warn("Overflow event, file updates may have been lost");
                        }
                        continue;
                    }

                    @SuppressWarnings("unchecked")
                    WatchEvent<Path> event = (WatchEvent<Path>) ev;

                    Path file = parent.resolve(event.context());
                    if (targetToRequests.containsKey(file)) {
                        LOGGER.trace("Watched file " + file + " event " + kind);
                        updates.add(file);
                    }
                }
                // Ready the key again so it wil signal more events.
                key.reset();

                if (updates.size() > 0) {
                    List<Supplier<Listener>>           watcherList;
                    Map<Path,List<Supplier<Listener>>> watcherMap;
                    Set<Path>                          updatedRequests = new TreeSet<>();
                    synchronized (mutex) {
                        watcherList = new ArrayList<>(this.watchers);
                        watcherMap = deepCopy(watchedFiles);

                        for (final Path file : updates) {
                            updatedRequests.addAll(targetToRequests.getOrDefault(file, Collections.emptySet()));
                        }
                        for (final Path file : updates) {
                            try {
                                updateLinkTargetting(file);
                            } catch (IOException e) {
                                LOGGER.warn("Failed to update link targeting: {}", e.getMessage(), e);
                            }
                        }
                    }

                    callbackExecutor.submit(() -> {
                        for (final Path file : updates) {
                            for (final Supplier<Listener> supplier : watcherList) {
                                Listener watcher = supplier.get();
                                if (watcher != null) {
                                    try {
                                        watcher.onPathUpdate(file);
                                    } catch (RuntimeException e) {
                                        LOGGER.error("Exception when notifying update on " + file, e);
                                    }
                                }
                            }
                        }
                        for (Path request : updatedRequests) {
                            Optional.ofNullable(watcherMap.get(request))
                                    .ifPresent(list -> {
                                        for (final Supplier<Listener> supplier : list) {
                                            Listener watcher = supplier.get();
                                            if (watcher != null) {
                                                try {
                                                    watcher.onPathUpdate(request);
                                                } catch (RuntimeException e) {
                                                    LOGGER.error("Exception when notifying update on " + request, e);
                                                }
                                            }
                                        }
                                    });
                        }
                    });
                }
            } catch (InterruptedException interruptedEx) {
                LOGGER.error("Interrupted in service file watch thread: " + interruptedEx.getMessage(), interruptedEx);
                return;
            }
        }
    }
875977
function watch(glob, taskName)
 {
    var watcher = chokidar.watch(glob, {persistent: true,});
    watcher.on('ready', ()=>
    {
        watcher.on('all', (event, path)=>
        {
            invokeTask(taskName);
        });
    });
 }
793237
function(self, agent, id, cb) {
  let msg = {action: 'subscribe', id: id};
  agent.request(Constants.KEYWORDS.MASTER_WATCHER, msg, function(err, servers) {
    if(err) {
      logger.error('subscribeRequest request to master with error: %j', err.stack);
      utils.invokeCallback(cb, err);
    }
    let res = [];
    for(let id in servers) {
      res.push(servers[id]);
    }
    addServers(self, res);
    utils.invokeCallback(cb);
  });
}
1107719
def handle_zk_event(event)
      update_znode_timestamp
      if event.node_created? || event.node_changed?
        build_clients
      elsif event.node_deleted?
        purge_clients
        @zk.stat(redis_nodes_path, :watch => true)
      else
        logger.error("Unknown ZK node event: #{event.inspect}")
      end
    ensure
      @zk.stat(redis_nodes_path, :watch => true)
    end
214765
private void _watchNode(String path) throws ZooKeeperException {
        try {
            cacheNodeWatcher.get(path);
        } catch (Exception e) {
            if (e instanceof ZooKeeperException) {
                throw (ZooKeeperException) e;
            } else {
                throw new ZooKeeperException(e);
            }
        }
    }
736044
def watch(
        self,
        username,
        watch={
            "friend":True,
            "deviations":True,
            "journals":True,
            "forum_threads":True,
            "critiques":True,
            "scraps":True,
            "activity":True,
            "collections":True
        }
    ):

        

        if self.standard_grant_type is not "authorization_code":
            raise DeviantartError("Authentication through Authorization Code (Grant Type) is required in order to connect to this endpoint.")

        response = self._req('/user/friends/watch/{}'.format(username), post_data={
            "watch[friend]": watch['friend'],
            "watch[deviations]": watch['deviations'],
            "watch[journals]": watch['journals'],
            "watch[forum_threads]": watch['forum_threads'],
            "watch[critiques]": watch['critiques'],
            "watch[scraps]": watch['scraps'],
            "watch[activity]": watch['activity'],
            "watch[collections]": watch['collections'],
        })

        return response['success']
1129629
@Override
  protected void initZnode() { 
    try {
      ZooKeeperUtil.createAndFailSilent(this, "/chronos");
      super.initZnode();
      ZooKeeperUtil.createAndFailSilent(this, baseZnode + "/persistent-timestamp");
    } catch (Exception e) {
      LOG.fatal("Error to create znode of chronos, exit immediately");
      System.exit(0);
    }
  }
1084709
public void requestUserSnapshot(final StoredProcedureInvocation invocation, final Connection c) {
        m_es.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    submitUserSnapshotRequest(invocation, c);
                } catch (Exception e) {
                    VoltDB.crashLocalVoltDB("Exception submitting user snapshot request", true, e);
                }
            }
        });
    }
238378
function addWatcher(file, handler, callback)
{
  fs.stat(file, function(err, stat)
  {
    if (err) return callback(false);

    fs.watch(file, handler(file));

    callback(true, stat.mtime.getTime());
  });
}
536000
protected void processParentEvent() throws Exception {
        // get current children snapshot and reset this watch.
        Set<String> children = new TreeSet<String>(m_zk.getChildren(m_rootNode, m_parentWatch));
        // intersect to get newChildren and update m_lastChildren to the current set.
        Set<String> newChildren = new HashSet<String>(children);
        newChildren.removeAll(m_lastChildren);
        m_lastChildren = children;

        List<ByteArrayCallback> callbacks = new ArrayList<ByteArrayCallback>();
        for (String child : children) {
            ByteArrayCallback cb = new ByteArrayCallback();
            // set watches on new children.
            if(newChildren.contains(child)) {
                m_zk.getData(ZKUtil.joinZKPath(m_rootNode, child), m_childWatch, cb, null);
            } else {
                m_zk.getData(ZKUtil.joinZKPath(m_rootNode, child), false, cb, null);
            }

            callbacks.add(cb);
        }

        HashMap<Integer, LeaderCallBackInfo> cache = new HashMap<Integer, LeaderCallBackInfo>();
        for (ByteArrayCallback callback : callbacks) {
            try {
                byte payload[] = callback.get();
                // During initialization children node may contain no data.
                if (payload == null) {
                    continue;
                }
                String data = new String(payload, "UTF-8");
                LeaderCallBackInfo info = LeaderCache.buildLeaderCallbackFromString(data);
                Integer partitionId = getPartitionIdFromZKPath(callback.getPath());
                cache.put(partitionId, info);
            } catch (KeeperException.NoNodeException e) {
                // child may have been deleted between the parent trigger and getData.
            }
        }

        m_publicCache = ImmutableMap.copyOf(cache);
        if (m_cb != null) {
            m_cb.run(m_publicCache);
        }
    }