final ConcurrentLinkedQueue<Operation> ops =
new ConcurrentLinkedQueue<Operation>();
CountDownLatch blatch = broadcastOp(new BroadcastOpFactory() {
public Operation newOp(final MemcachedNode n,
final CountDownLatch latch) {
Operation op = opFact.flush(delay, new OperationCallback() {
public void receivedStatus(OperationStatus s) {
flushResult.set(s.isSuccess());
}
public void complete() {
latch.countDown();
}
});
ops.add(op);
return op;
}
});
return new OperationFuture<Boolean>(null, blatch, flushResult,
operationTimeout) {
@Override
public boolean cancel(boolean ign) {
boolean rv = false;
for (Operation op : ops) {
op.cancel();
rv |= op.getState() == OperationState.WRITE_QUEUED;
}
return rv;
}
@Override
public Boolean get(long duration, TimeUnit units)
throws InterruptedException, TimeoutException, ExecutionException {
status = new OperationStatus(true, "OK");
return super.get(duration, units);
}
@Override
public boolean isCancelled() {
boolean rv = false;
for (Operation op : ops) {
rv |= op.isCancelled();
}
return rv;
}
@Override
public boolean isDone() {
boolean rv = true;
for (Operation op : ops) {
rv &= op.getState() == OperationState.COMPLETE;
}
return rv || isCancelled();
}
};
}