public QueueEvent offer(byte[] data,long timeout) throws KeeperException, InterruptedException {
  TimerContext time=stats.time(dir + "_offer");
  try {
    String path=createData(dir + "/" + prefix,data,CreateMode.PERSISTENT_SEQUENTIAL);
    String watchID=createData(dir + "/" + response_prefix+ path.substring(path.lastIndexOf("-") + 1),null,CreateMode.EPHEMERAL);
    Object lock=new Object();
    LatchChildWatcher watcher=new LatchChildWatcher(lock);
synchronized (lock) {
      if (zookeeper.exists(watchID,watcher,true) != null) {
        watcher.await(timeout);
      }
    }
    byte[] bytes=zookeeper.getData(watchID,null,null,true);
    zookeeper.delete(watchID,-1,true);
    return new QueueEvent(watchID,bytes,watcher.getWatchedEvent());
  }
  finally {
    time.stop();
  }
}

public QueueEvent offer ( byte [ ] data , long timeout ) throws KeeperException , InterruptedException { TimerContext time = stats . time ( dir + "_offer" ) ; try { String path = createData ( dir + "/" + prefix , data , CreateMode . PERSISTENT_SEQUENTIAL ) ; String watchID = createData ( dir + "/" + response_prefix + path . substring ( path . lastIndexOf ( "-" ) + 1 ) , null , CreateMode . EPHEMERAL ) ; Object lock = new Object ( ) ; LatchChildWatcher watcher = new LatchChildWatcher ( lock ) ; synchronized ( lock ) { if ( zookeeper . exists ( watchID , watcher , true ) != null ) { watcher . await ( timeout ) ; } } byte [ ] bytes = zookeeper . getData ( watchID , null , null , true ) ; zookeeper . delete ( watchID , - 1 , true ) ; return new QueueEvent ( watchID , bytes , watcher . getWatchedEvent ( ) ) ; } finally { time . stop ( ) ; } }
offer the data and wait for the response
