java类org.apache.zookeeper.AsyncCallback.DataCallback的实例源码

/**
 * The asynchronous version of getData.
 *
 * @see #getData(String, Watcher, Stat)
 */
public void getData(final String path, Watcher watcher,
        DataCallback cb, Object ctx)
{
    final String clientPath = path;
    PathUtils.validatePath(clientPath);

    // the watch contains the un-chroot path
    WatchRegistration wcb = null;
    if (watcher != null) {
        wcb = new DataWatchRegistration(watcher, clientPath);
    }

    final String serverPath = prependChroot(clientPath);

    RequestHeader h = new RequestHeader();
    h.setType(ZooDefs.OpCode.getData);
    GetDataRequest request = new GetDataRequest();
    request.setPath(serverPath);
    request.setWatch(watcher != null);
    GetDataResponse response = new GetDataResponse();
    cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
            clientPath, serverPath, ctx, wcb);
}

/**
 * The asynchronous version of getConfig.
 *
 * @see #getConfig(Watcher, Stat)
 */
public void getConfig(Watcher watcher,
        DataCallback cb, Object ctx)
{
    final String configZnode = ZooDefs.CONFIG_NODE;

    // the watch contains the un-chroot path
    WatchRegistration wcb = null;
    if (watcher != null) {
        wcb = new DataWatchRegistration(watcher, configZnode);
    }

    RequestHeader h = new RequestHeader();
    h.setType(ZooDefs.OpCode.getData);
    GetDataRequest request = new GetDataRequest();
    request.setPath(configZnode);
    request.setWatch(watcher != null);
    GetDataResponse response = new GetDataResponse();
    cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
           configZnode, configZnode, ctx, wcb);
}

/**
 * The Asynchronous version of getData. The request doesn't actually until
 * the asynchronous callback is called.
 *
 * @see #getData(String, Watcher, Stat)
 */
public void getData(final String path, Watcher watcher,
        DataCallback cb, Object ctx)
{
    final String clientPath = path;
    PathUtils.validatePath(clientPath);

    // the watch contains the un-chroot path
    WatchRegistration wcb = null;
    if (watcher != null) {
        wcb = new DataWatchRegistration(watcher, clientPath);
    }

    final String serverPath = prependChroot(clientPath);

    RequestHeader h = new RequestHeader();
    h.setType(ZooDefs.OpCode.getData);
    GetDataRequest request = new GetDataRequest();
    request.setPath(serverPath);
    request.setWatch(watcher != null);
    GetDataResponse response = new GetDataResponse();
    cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
            clientPath, serverPath, ctx, wcb);
}

/**
 * The asynchronous version of getData.
 *
 * @see #getData(String, Watcher, Stat)
 */
public void getData(final String path, Watcher watcher,
        DataCallback cb, Object ctx)
{
    final String clientPath = path;
    PathUtils.validatePath(clientPath);

    // the watch contains the un-chroot path
    WatchRegistration wcb = null;
    if (watcher != null) {
        wcb = new DataWatchRegistration(watcher, clientPath);
    }

    final String serverPath = prependChroot(clientPath);

    RequestHeader h = new RequestHeader();
    h.setType(ZooDefs.OpCode.getData);
    GetDataRequest request = new GetDataRequest();
    request.setPath(serverPath);
    request.setWatch(watcher != null);
    GetDataResponse response = new GetDataResponse();
    cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
            clientPath, serverPath, ctx, wcb);
}

/**
 * The asynchronous version of getConfig.
 *
 * @see #getConfig(Watcher, Stat)
 */
public void getConfig(Watcher watcher,
        DataCallback cb, Object ctx)
{
    final String configZnode = ZooDefs.CONFIG_NODE;

    // the watch contains the un-chroot path
    WatchRegistration wcb = null;
    if (watcher != null) {
        wcb = new DataWatchRegistration(watcher, configZnode);
    }

    RequestHeader h = new RequestHeader();
    h.setType(ZooDefs.OpCode.getData);
    GetDataRequest request = new GetDataRequest();
    request.setPath(configZnode);
    request.setWatch(watcher != null);
    GetDataResponse response = new GetDataResponse();
    cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
           configZnode, configZnode, ctx, wcb);
}

/**
 * The asynchronous version of getData.
 *
 * @see #getData(String, Watcher, Stat)
 */
public void getData(final String path, Watcher watcher,
        DataCallback cb, Object ctx)
{
    final String clientPath = path;
    PathUtils.validatePath(clientPath);

    // the watch contains the un-chroot path
    WatchRegistration wcb = null;
    if (watcher != null) {
        wcb = new DataWatchRegistration(watcher, clientPath);
    }

    final String serverPath = prependChroot(clientPath);

    RequestHeader h = new RequestHeader();
    h.setType(ZooDefs.OpCode.getData);
    GetDataRequest request = new GetDataRequest();
    request.setPath(serverPath);
    request.setWatch(watcher != null);
    GetDataResponse response = new GetDataResponse();
    cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
            clientPath, serverPath, ctx, wcb);
}

/**
 * The asynchronous version of getConfig.
 *
 * @see #getConfig(Watcher, Stat)
 */
public void getConfig(Watcher watcher,
        DataCallback cb, Object ctx)
{
    final String configZnode = ZooDefs.CONFIG_NODE;

    // the watch contains the un-chroot path
    WatchRegistration wcb = null;
    if (watcher != null) {
        wcb = new DataWatchRegistration(watcher, configZnode);
    }

    RequestHeader h = new RequestHeader();
    h.setType(ZooDefs.OpCode.getData);
    GetDataRequest request = new GetDataRequest();
    request.setPath(configZnode);
    request.setWatch(watcher != null);
    GetDataResponse response = new GetDataResponse();
    cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
           configZnode, configZnode, ctx, wcb);
}

private void removeWorkerNodeInJobs(String worker) {
    final String workerPath = ZKDistributedJobEngine.NODES_PATH + "/" + worker;
    zookeeper.getData(workerPath, false, new DataCallback() {
        @Override
        public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
            if (rc == KeeperException.Code.OK.intValue())  {                    
                try {
                    zookeeper.delete(workerPath, stat.getVersion());
                } catch (Exception e) {
                    logger.warn("exception delete node", e);
                }
            }
        }
    }, null);

}

private void notifyWorkerToExecuteJob(final String workerName, final IJavaCCJob jobInfo) {
    final String workerPath = ZKDistributedJobEngine.NODES_PATH + "/" + workerName;
    zookeeper.getData(workerPath, watcher, new DataCallback() {
        @Override
        public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {

            if (logger.isDebugEnabled()) {
                logger.debug("notifing worker [" + workerName + "] to execute job [" + jobInfo.getId() + "],stat ="
                        + stat + "\n rc=" + rc + "\n path=" + path);
            }
            Map<String, Object> context = new HashMap<String, Object>();
            context.put(DATA, data);
            context.put(PATH, path);
            context.put(STAT2, stat);

            Operation operation = Operation.get(rc);
            if (operation != null) {
                operation.apply(workerName, jobInfo, LeaderJobScheduler.this, context);
                return;
            }
            logger.warn("Fail notify worker '" + workerName + "' to execute job '" + jobInfo.getId()
                    + "'   errorCode = " + rc);
        }
    }, null);
}

/**
 * The Asynchronous version of getData. The request doesn't actually until
 * the asynchronous callback is called.
 *
 * @see #getData(String, Watcher, Stat)
 */
public void getData(final String path, Watcher watcher,
        DataCallback cb, Object ctx)
{
    final String clientPath = path;
    PathUtils.validatePath(clientPath);

    // the watch contains the un-chroot path
    WatchRegistration wcb = null;
    if (watcher != null) {
        wcb = new DataWatchRegistration(watcher, clientPath);
    }

    final String serverPath = prependChroot(clientPath);

    RequestHeader h = new RequestHeader();
    h.setType(ZooDefs.OpCode.getData);
    GetDataRequest request = new GetDataRequest();
    request.setPath(serverPath);
    request.setWatch(watcher != null);
    GetDataResponse response = new GetDataResponse();
    cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
            clientPath, serverPath, ctx, wcb);
}

/**
 * The Asynchronous version of reconfig.
 *
 * @see #reconfigure
 *
 **/
public void reconfigure(String joiningServers, String leavingServers,
                        String newMembers, long fromConfig, DataCallback cb, Object ctx) {
    RequestHeader h = new RequestHeader();
    h.setType(ZooDefs.OpCode.reconfig);
    ReconfigRequest request = new ReconfigRequest(joiningServers, leavingServers, newMembers, fromConfig);
    GetDataResponse response = new GetDataResponse();
    cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
           ZooDefs.CONFIG_NODE, ZooDefs.CONFIG_NODE, ctx, null);
}

/**
 * Convenience wrapper around asynchronous reconfig that takes Lists of strings instead of comma-separated servers.
 *
 * @see #reconfigure
 *
 */
public void reconfigure(List<String> joiningServers,
                        List<String> leavingServers, List<String> newMembers, long fromConfig,
                        DataCallback cb, Object ctx) {
    reconfigure(StringUtils.joinStrings(joiningServers, ","),
             StringUtils.joinStrings(leavingServers, ","),
             StringUtils.joinStrings(newMembers, ","),
             fromConfig, cb, ctx);
}

@Override
public void getData(final String path, boolean watch, final DataCallback cb, final Object ctx) {
    executor.execute(() -> {
        checkReadOpDelay();
        if (getProgrammedFailStatus()) {
            cb.processResult(failReturnCode.intValue(), path, ctx, null, null);
            return;
        } else if (stopped) {
            cb.processResult(KeeperException.Code.ConnectionLoss, path, ctx, null, null);
            return;
        }

        Pair<byte[], Integer> value;
        mutex.lock();
        try {
            value = tree.get(path);
        } finally {
            mutex.unlock();
        }

        if (value == null) {
            cb.processResult(KeeperException.Code.NoNode, path, ctx, null, null);
        } else {
            Stat stat = new Stat();
            stat.setVersion(value.second);
            cb.processResult(0, path, ctx, value.first, stat);
        }
    });
}

@Override
public void getData(final String path, final Watcher watcher, final DataCallback cb, final Object ctx) {
    executor.execute(() -> {
        checkReadOpDelay();
        mutex.lock();
        if (getProgrammedFailStatus()) {
            mutex.unlock();
            cb.processResult(failReturnCode.intValue(), path, ctx, null, null);
            return;
        } else if (stopped) {
            mutex.unlock();
            cb.processResult(KeeperException.Code.CONNECTIONLOSS.intValue(), path, ctx, null, null);
            return;
        }

        Pair<byte[], Integer> value = tree.get(path);
        if (value == null) {
            mutex.unlock();
            cb.processResult(KeeperException.Code.NONODE.intValue(), path, ctx, null, null);
        } else {
            if (watcher != null) {
                watchers.put(path, watcher);
            }

            Stat stat = new Stat();
            stat.setVersion(value.second);
            mutex.unlock();
            cb.processResult(0, path, ctx, value.first, stat);
        }
    });
}

/**
 * The Asynchronous version of reconfig. 
 *
 * @see #reconfig
 *      
 **/
public void reconfig(String joiningServers, String leavingServers,
    String newMembers, long fromConfig, DataCallback cb, Object ctx)
{
    RequestHeader h = new RequestHeader();
    h.setType(ZooDefs.OpCode.reconfig);       
    ReconfigRequest request = new ReconfigRequest(joiningServers, leavingServers, newMembers, fromConfig);
    GetDataResponse response = new GetDataResponse();
    cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
           ZooDefs.CONFIG_NODE, ZooDefs.CONFIG_NODE, ctx, null);
}

/**
 * Convenience wrapper around asynchronous reconfig that takes Lists of strings instead of comma-separated servers.
 *
 * @see #reconfig
 *
 */
public void reconfig(List<String> joiningServers,
    List<String> leavingServers, List<String> newMembers, long fromConfig,
    DataCallback cb, Object ctx)
{
    reconfig(StringUtils.joinStrings(joiningServers, ","), 
            StringUtils.joinStrings(leavingServers, ","), 
            StringUtils.joinStrings(newMembers, ","), 
            fromConfig, cb, ctx);
}

/**
 * The Asynchronous version of reconfig. 
 *
 * @see #reconfig
 *      
 **/
public void reconfig(String joiningServers, String leavingServers,
    String newMembers, long fromConfig, DataCallback cb, Object ctx)
{
    RequestHeader h = new RequestHeader();
    h.setType(ZooDefs.OpCode.reconfig);       
    ReconfigRequest request = new ReconfigRequest(joiningServers, leavingServers, newMembers, fromConfig);
    GetDataResponse response = new GetDataResponse();
    cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
           ZooDefs.CONFIG_NODE, ZooDefs.CONFIG_NODE, ctx, null);
}

/**
 * Convenience wrapper around asynchronous reconfig that takes Lists of strings instead of comma-separated servers.
 *
 * @see #reconfig
 *
 */
public void reconfig(List<String> joiningServers,
    List<String> leavingServers, List<String> newMembers, long fromConfig,
    DataCallback cb, Object ctx)
{
    reconfig(StringUtils.joinStrings(joiningServers, ","), 
            StringUtils.joinStrings(leavingServers, ","), 
            StringUtils.joinStrings(newMembers, ","), 
            fromConfig, cb, ctx);
}

@Override
protected void installWatch() {
    connection.getData(getNode(), this, new DataCallback() {
        public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
        }
    }, null);
}

/**
 * The asynchronous version of getData.
 *
 * @see #getData(String, boolean, Stat)
 */
public void getData(String path, boolean watch, DataCallback cb, Object ctx) {
    getData(path, watch ? watchManager.defaultWatcher : null, cb, ctx);
}

/**
 * The Asynchronous version of getConfig. 
 * 
 * @see #getData(String, boolean, Stat)
 */
public void getConfig(boolean watch, DataCallback cb, Object ctx) {
    getConfig(watch ? watchManager.defaultWatcher : null, cb, ctx);
}

/**
 * The Asynchronous version of getData. The request doesn't actually until
 * the asynchronous callback is called.
 *
 * @see #getData(String, boolean, Stat)
 */
public void getData(String path, boolean watch, DataCallback cb, Object ctx) {
    getData(path, watch ? watchManager.defaultWatcher : null, cb, ctx);
}

/**
 * The asynchronous version of getData.
 *
 * @see #getData(String, boolean, Stat)
 */
public void getData(String path, boolean watch, DataCallback cb, Object ctx) {
    getData(path, watch ? watchManager.defaultWatcher : null, cb, ctx);
}

/**
 * The Asynchronous version of getConfig. 
 * 
 * @see #getData(String, boolean, Stat)
 */
public void getConfig(boolean watch, DataCallback cb, Object ctx) {
    getConfig(watch ? watchManager.defaultWatcher : null, cb, ctx);
}

/**
 * The asynchronous version of getData.
 *
 * @see #getData(String, boolean, Stat)
 */
public void getData(String path, boolean watch, DataCallback cb, Object ctx) {
    getData(path, watch ? watchManager.defaultWatcher : null, cb, ctx);
}

/**
 * The Asynchronous version of getConfig. 
 * 
 * @see #getData(String, boolean, Stat)
 */
public void getConfig(boolean watch, DataCallback cb, Object ctx) {
    getConfig(watch ? watchManager.defaultWatcher : null, cb, ctx);
}

/**
 * Passes a callback and a context object to the config/reconfig command.
 * @param callback  The async callback to use.
 * @param ctx       An object that will be passed to the callback.
 * @return this
 */
T usingDataCallback(DataCallback callback, Object ctx);

/**
 * The Asynchronous version of getData. The request doesn't actually until
 * the asynchronous callback is called.
 *
 * @see #getData(String, boolean, Stat)
 */
public void getData(String path, boolean watch, DataCallback cb, Object ctx) {
    getData(path, watch ? watchManager.defaultWatcher : null, cb, ctx);
}