Java 类com.google.common.util.concurrent.ListenableFutureTask 实例源码

public <V> ListenableFuture<V> callFromMainThread(Callable<V> callable)
{
    Validate.notNull(callable);

    if (!this.isCallingFromMinecraftThread() && !this.isServerStopped())
    {
        ListenableFutureTask<V> listenablefuturetask = ListenableFutureTask.<V>create(callable);

        synchronized (this.futureTaskQueue)
        {
            this.futureTaskQueue.add(listenablefuturetask);
            return listenablefuturetask;
        }
    }
    else
    {
        try
        {
            return Futures.<V>immediateFuture(callable.call());
        }
        catch (Exception exception)
        {
            return Futures.immediateFailedCheckedFuture(exception);
        }
    }
}

public <V> ListenableFuture<V> addScheduledTask(Callable<V> callableToSchedule)
{
    Validate.notNull(callableToSchedule);

    if (!this.isCallingFromMinecraftThread())
    {
        ListenableFutureTask<V> listenablefuturetask = ListenableFutureTask.<V>create(callableToSchedule);

        synchronized (this.scheduledTasks)
        {
            this.scheduledTasks.add(listenablefuturetask);
            return listenablefuturetask;
        }
    }
    else
    {
        try
        {
            return Futures.<V>immediateFuture(callableToSchedule.call());
        }
        catch (Exception exception)
        {
            return Futures.immediateFailedCheckedFuture(exception);
        }
    }
}

public <V> ListenableFuture<V> callFromMainThread(Callable<V> callable)
{
    Validate.notNull(callable);

    if (!this.isCallingFromMinecraftThread() && !this.isServerStopped())
    {
        ListenableFutureTask<V> listenablefuturetask = ListenableFutureTask.<V>create(callable);

        synchronized (this.futureTaskQueue)
        {
            this.futureTaskQueue.add(listenablefuturetask);
            return listenablefuturetask;
        }
    }
    else
    {
        try
        {
            return Futures.<V>immediateFuture(callable.call());
        }
        catch (Exception exception)
        {
            return Futures.immediateFailedCheckedFuture(exception);
        }
    }
}

public <V> ListenableFuture<V> addScheduledTask(Callable<V> callableToSchedule)
{
    Validate.notNull(callableToSchedule);

    if (!this.isCallingFromMinecraftThread())
    {
        ListenableFutureTask<V> listenablefuturetask = ListenableFutureTask.<V>create(callableToSchedule);

        synchronized (this.scheduledTasks)
        {
            this.scheduledTasks.add(listenablefuturetask);
            return listenablefuturetask;
        }
    }
    else
    {
        try
        {
            return Futures.<V>immediateFuture(callableToSchedule.call());
        }
        catch (Exception exception)
        {
            return Futures.immediateFailedCheckedFuture(exception);
        }
    }
}

public <V> ListenableFuture<V> callFromMainThread(Callable<V> callable)
{
    Validate.notNull(callable);

    if (!this.isCallingFromMinecraftThread() && !this.isServerStopped())
    {
        ListenableFutureTask<V> listenablefuturetask = ListenableFutureTask.<V>create(callable);

        synchronized (this.futureTaskQueue)
        {
            this.futureTaskQueue.add(listenablefuturetask);
            return listenablefuturetask;
        }
    }
    else
    {
        try
        {
            return Futures.<V>immediateFuture(callable.call());
        }
        catch (Exception exception)
        {
            return Futures.immediateFailedCheckedFuture(exception);
        }
    }
}

public <V> ListenableFuture<V> addScheduledTask(Callable<V> callableToSchedule) {
    Validate.notNull(callableToSchedule);

    if (!this.isCallingFromMinecraftThread()) {
        ListenableFutureTask<V> listenablefuturetask = ListenableFutureTask.<V>create(callableToSchedule);

        synchronized (this.scheduledTasks) {
            this.scheduledTasks.add(listenablefuturetask);
            return listenablefuturetask;
        }
    } else {
        try {
            return Futures.<V>immediateFuture(callableToSchedule.call());
        } catch (Exception exception) {
            return Futures.immediateFailedCheckedFuture(exception);
        }
    }
}

public <V> ListenableFuture<V> callFromMainThread(Callable<V> callable)
{
    Validate.notNull(callable);

    if (!this.isCallingFromMinecraftThread() && !this.isServerStopped())
    {
        ListenableFutureTask<V> listenablefuturetask = ListenableFutureTask.<V>create(callable);

        synchronized (this.futureTaskQueue)
        {
            this.futureTaskQueue.add(listenablefuturetask);
            return listenablefuturetask;
        }
    }
    else
    {
        try
        {
            return Futures.<V>immediateFuture(callable.call());
        }
        catch (Exception exception)
        {
            return Futures.immediateFailedCheckedFuture(exception);
        }
    }
}

public <V> ListenableFuture<V> addScheduledTask(Callable<V> callableToSchedule)
{
    Validate.notNull(callableToSchedule);

    if (this.isCallingFromMinecraftThread())
    {
        try
        {
            return Futures.<V>immediateFuture(callableToSchedule.call());
        }
        catch (Exception exception)
        {
            return Futures.immediateFailedCheckedFuture(exception);
        }
    }
    else
    {
        ListenableFutureTask<V> listenablefuturetask = ListenableFutureTask.<V>create(callableToSchedule);

        synchronized (this.scheduledTasks)
        {
            this.scheduledTasks.add(listenablefuturetask);
            return listenablefuturetask;
        }
    }
}

public <V> ListenableFuture<V> addScheduledTask(Callable<V> callableToSchedule)
{
    Validate.notNull(callableToSchedule);

    if (this.isCallingFromMinecraftThread())
    {
        try
        {
            return Futures.<V>immediateFuture(callableToSchedule.call());
        }
        catch (Exception exception)
        {
            return Futures.immediateFailedCheckedFuture(exception);
        }
    }
    else
    {
        ListenableFutureTask<V> listenablefuturetask = ListenableFutureTask.<V>create(callableToSchedule);

        synchronized (this.scheduledTasks)
        {
            this.scheduledTasks.add(listenablefuturetask);
            return listenablefuturetask;
        }
    }
}

public <V> ListenableFuture<V> callFromMainThread(Callable<V> callable)
{
    Validate.notNull(callable);

    if (!this.isCallingFromMinecraftThread() && !this.isServerStopped())
    {
        ListenableFutureTask<V> listenablefuturetask = ListenableFutureTask.<V>create(callable);

        synchronized (this.futureTaskQueue)
        {
            this.futureTaskQueue.add(listenablefuturetask);
            return listenablefuturetask;
        }
    }
    else
    {
        try
        {
            return Futures.<V>immediateFuture(callable.call());
        }
        catch (Exception exception)
        {
            return Futures.immediateFailedCheckedFuture(exception);
        }
    }
}

public <V> ListenableFuture<V> addScheduledTask(Callable<V> callableToSchedule)
{
    Validate.notNull(callableToSchedule);

    if (this.isCallingFromMinecraftThread())
    {
        try
        {
            return Futures.<V>immediateFuture(callableToSchedule.call());
        }
        catch (Exception exception)
        {
            return Futures.immediateFailedCheckedFuture(exception);
        }
    }
    else
    {
        ListenableFutureTask<V> listenablefuturetask = ListenableFutureTask.<V>create(callableToSchedule);

        synchronized (this.scheduledTasks)
        {
            this.scheduledTasks.add(listenablefuturetask);
            return listenablefuturetask;
        }
    }
}

public <V> ListenableFuture<V> callFromMainThread(Callable<V> callable)
{
    Validate.notNull(callable);

    if (!this.isCallingFromMinecraftThread() && !this.isServerStopped())
    {
        ListenableFutureTask<V> listenablefuturetask = ListenableFutureTask.<V>create(callable);

        synchronized (this.futureTaskQueue)
        {
            this.futureTaskQueue.add(listenablefuturetask);
            return listenablefuturetask;
        }
    }
    else
    {
        try
        {
            return Futures.<V>immediateFuture(callable.call());
        }
        catch (Exception exception)
        {
            return Futures.immediateFailedCheckedFuture(exception);
        }
    }
}

public static void main(String[] args) {
    ExecutorService executorService = Executors.newSingleThreadExecutor();
    // create ListenableFutureTask instance
    ListenableFutureTask<String> listenableFutureTask = ListenableFutureTask.create(() -> {
        log.info("listenable future begin to execute");
        TimeUnit.SECONDS.sleep(5);  // monitor
        return "listenable future task done";
    });
    // add three listeners, use MoreExecutors.directExecutor() to execute Runnable in ExecutorServcie
    listenableFutureTask.addListener(()-> log.info("listener one execute"), MoreExecutors.directExecutor());
    listenableFutureTask.addListener(() -> log.info("listener two execute"), MoreExecutors.directExecutor());
    listenableFutureTask.addListener(() -> log.info("listener three execute"), MoreExecutors.directExecutor());
    // execute listenable future task
    executorService.execute(listenableFutureTask);
    // ExecutorService.shutdown()
    executorService.shutdown();
}

public static void main(String[] args) throws BrokenBarrierException, InterruptedException {
    final CyclicBarrier barrier = new CyclicBarrier(4);
    ExecutorService executorService = Executors.newSingleThreadExecutor();
    ListenableFutureTask<String> listenableFutureTask = ListenableFutureTask.create(() -> {
        log.info("listenable future begin to execute");
        TimeUnit.SECONDS.sleep(5);
        return "listenable future task done";
    });
    // !!!!! barrier貌似并不能实现，对比下CountDownLatch，两种控制的应用场景
    Futures.addCallback(listenableFutureTask, new BarrieredSimpleFutureCallback(0, barrier), executorService);
    Futures.addCallback(listenableFutureTask, new BarrieredSimpleFutureCallback(1, barrier), executorService);
    Futures.addCallback(listenableFutureTask, new BarrieredSimpleFutureCallback(2, barrier), executorService);
    executorService.execute(listenableFutureTask);

    barrier.await();
    executorService.shutdown();
}

public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
    final CountDownLatch latch = new CountDownLatch(3);
    //ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
    ExecutorService executorService = Executors.newSingleThreadExecutor();
    ListenableFutureTask<String> listenableFutureTask = ListenableFutureTask.create(() -> {
        log.info("listenable future begin to execute");
        TimeUnit.SECONDS.sleep(5);
        return "listenable future task done";
    });
    // add three callbacks, use executorService instance to execute
    Futures.addCallback(listenableFutureTask, new LatchedSimpleFutureCallback(0, latch), executorService);
    Futures.addCallback(listenableFutureTask, new LatchedSimpleFutureCallback(1, latch), executorService);
    Futures.addCallback(listenableFutureTask, new LatchedSimpleFutureCallback(2, latch), executorService);
    // execute listenable future task
    executorService.execute(listenableFutureTask);
    // ensure listener execution before ExecutorService#shutdown
    if (latch.getCount() > 0) {
        latch.await();
    }
    // ExecutorService.shutdown()
    executorService.shutdown();
}

public AbstractJobInstance(Config config) {
    Assert.notNull(config.parameters, "parameters is null");
    Assert.notNull(config.job, "job is null");
    Assert.notNull(config.jobsManager, "jobsManager is null");
    Assert.notNull(config.info, "info is null");
    this.jobContext = new JobContext(this, config.parameters);
    this.infoRef = new AtomicReference<>(config.info);
    this.manager = config.jobsManager;
    // create wait future with stub
    this.atEndFuture = SettableFuture.create();
    this.job = config.job;
    this.authentication = config.authentication;
    this.watcher = config.watcher;
    this.cancelFuture = ListenableFutureTask.create(this::innerCancel);
    this.startFuture = ListenableFutureTask.create(this::innerStart);
}

private ListenableFuture<Void> sheduleUpdateHtml(final WaveId waveId) {
  ListenableFutureTask<Void> task = processingWaves.get(waveId);
  if (task == null) {
    task = ListenableFutureTask.create(new Callable<Void>() {

      @Override
      public Void call() throws Exception {
        processingWaves.remove(waveId);
        try {
          updateHtml(waveId);
        } catch (Exception e) {
          LOG.log(Level.SEVERE, "Failed to make HTML for " + waveId.serialise(), e);
          throw e;
        }
        return null;
      }
    });
    processingWaves.put(waveId, task);
    executor.schedule(task, WAVE_UPDATE_HTML_DELAY_MIN, TimeUnit.MINUTES);
  }
  return task;
}

@Override
public ListenableFuture close() {
  checkOpened();
  writeLock.lock();
  try {
    closingFutureTask = ListenableFutureTask.<Void>create(closingTask);
    if (latestVersionToPersist != null) {
      deferredClosing = true;
    } else {
      persistExecutor.execute(closingFutureTask);
    }
    return closingFutureTask;
  } finally {
    writeLock.unlock();
  }
}

private ListenableFutureTask<Void> sheduleCommitIndex() {
  ListenableFutureTask<Void> task = commitTask.get();
  if (task == null) {
    task = ListenableFutureTask.create(new Callable<Void>() {

      @Override
      public Void call() throws Exception {
        commitTask.set(null);
        try {
          LOG.info("Commiting indexes...");
          indexWriter.commit();
          LOG.info("Commiting indexes is complete");
        } catch (IndexException e) {
          LOG.log(Level.SEVERE, "Index commit failed", e);
          throw e;
        }
        return null;
      }
    });
    commitTask.set(task);
    indexExecutor.schedule(task, WAVE_COMMIT_DELAY_SEC, TimeUnit.SECONDS);
  }
  return task;
}

@VisibleForTesting
static ListenableFuture<SegmentWaveletState> loadSegmentWaveletState(
  final WaveletName waveletName,
  final WaveletStateFactory waveletStateFactory,
  Executor waveletLoadExecutor) {
  ListenableFutureTask<SegmentWaveletState> task
    = ListenableFutureTask.<SegmentWaveletState>create(
      new Callable<SegmentWaveletState>() {

        @Override
        public SegmentWaveletState call()
        throws PersistenceException, OperationException, WaveletStateException, InterruptedException, ExecutionException {
          BlockWaveletState blockState = waveletStateFactory.createBlockWaveletState(waveletName);
          blockState.open();
          SegmentWaveletState state = waveletStateFactory.createSegmentWaveletState(waveletName, blockState);
          state.open();
          return state;
        }
      });
  waveletLoadExecutor.execute(task);
  return task;
}

@Override
public ListenableFuture<ArrayChunk> load(final Pair<Integer, Integer> key) throws Exception {
  final PyDebugValue value = myProvider.getDebugValue();
  final PyDebugValue slicedValue =
    new PyDebugValue(myProvider.getSliceText(), value.getType(), value.getValue(), value.isContainer(), value.isErrorOnEval(),
                     value.getParent(), value.getFrameAccessor());

  ListenableFutureTask<ArrayChunk> task = ListenableFutureTask.create(new Callable<ArrayChunk>() {
    @Override
    public ArrayChunk call() throws Exception {
      return value.getFrameAccessor()
        .getArrayItems(slicedValue, key.first, key.second, Math.min(CHUNK_ROW_SIZE, getRowCount() - key.first),
                       Math.min(CHUNK_COL_SIZE, getColumnCount() - key.second),
                       myProvider.getFormat());
    }
  });

  myExecutorService.execute(task);

  return task;
}

private static void testOneSlice(final String id, final NavigableSet<Integer> test, final NavigableSet<Integer> canon, List<ListenableFuture<?>> results)
{
    ListenableFutureTask<?> f = ListenableFutureTask.create(new Runnable()
    {

        @Override
        public void run()
        {
            test(id + " Count", test.size(), canon.size());
            testEqual(id, test.iterator(), canon.iterator());
            testEqual(id + "->DSCI", test.descendingIterator(), canon.descendingIterator());
            testEqual(id + "->DSCS", test.descendingSet().iterator(), canon.descendingSet().iterator());
            testEqual(id + "->DSCS->DSCI", test.descendingSet().descendingIterator(), canon.descendingSet().descendingIterator());
        }
    }, null);
    results.add(f);
    COMPARE.execute(f);
}

public <V> ListenableFuture<V> a(Callable<V> callable) {
    Validate.notNull(callable);
    if (!this.isMainThread() && !this.isStopped()) {
        ListenableFutureTask listenablefuturetask = ListenableFutureTask.create(callable);
        Queue queue = this.j;

        synchronized (this.j) {
            this.j.add(listenablefuturetask);
            return listenablefuturetask;
        }
    } else {
        try {
            return Futures.immediateFuture(callable.call());
        } catch (Exception exception) {
            return Futures.immediateFailedCheckedFuture(exception);
        }
    }
}

private static void testOneSlice(final String id, final NavigableSet<Integer> test, final NavigableSet<Integer> canon, List<ListenableFuture<?>> results)
{
    ListenableFutureTask<?> f = ListenableFutureTask.create(new Runnable()
    {

        @Override
        public void run()
        {
            test(id + " Count", test.size(), canon.size());
            testEqual(id, test.iterator(), canon.iterator());
            testEqual(id + "->DSCI", test.descendingIterator(), canon.descendingIterator());
            testEqual(id + "->DSCS", test.descendingSet().iterator(), canon.descendingSet().iterator());
            testEqual(id + "->DSCS->DSCI", test.descendingSet().descendingIterator(), canon.descendingSet().descendingIterator());
        }
    }, null);
    results.add(f);
    if (DEBUG)
        f.run();
    else
        COMPARE.execute(f);
}

private static void testOneSlice(final String id, final NavigableSet<Integer> test, final NavigableSet<Integer> canon, List<ListenableFuture<?>> results)
{
    ListenableFutureTask<?> f = ListenableFutureTask.create(new Runnable()
    {

        @Override
        public void run()
        {
            test(id + " Count", test.size(), canon.size());
            testEqual(id, test.iterator(), canon.iterator());
            testEqual(id + "->DSCI", test.descendingIterator(), canon.descendingIterator());
            testEqual(id + "->DSCS", test.descendingSet().iterator(), canon.descendingSet().iterator());
            testEqual(id + "->DSCS->DSCI", test.descendingSet().descendingIterator(), canon.descendingSet().descendingIterator());
        }
    }, null);
    results.add(f);
    COMPARE.execute(f);
}

/**
 * Mark phase of the GC.
 */
private void mark() throws IOException, InterruptedException {
    state = State.MARKING;
    LOG.debug("Starting mark phase of the garbage collector");

    // Find all blobs available in the blob store
    ListenableFutureTask<Integer> blobIdRetriever = ListenableFutureTask.create(new BlobIdRetriever());
    if (runConcurrently) {
        executor.execute(blobIdRetriever);
    } else {
        MoreExecutors.sameThreadExecutor().execute(blobIdRetriever);
    }

    // Find all blob references after iterating over the whole repository
    iterateNodeTree();

    try {
        blobIdRetriever.get();
    } catch (ExecutionException e) {
       LOG.warn("Error occurred while fetching all the blobIds from the BlobStore. GC would " +
               "continue with the blobIds retrieved so far", e.getCause());
    }

    difference();
    LOG.debug("Ending mark phase of the garbage collector");
}

public <V> ListenableFuture<V> addScheduledTask(Callable<V> callableToSchedule)
{
    Validate.notNull(callableToSchedule);

    if (this.isCallingFromMinecraftThread())
    {
        try
        {
            return Futures.<V>immediateFuture(callableToSchedule.call());
        }
        catch (Exception exception)
        {
            return Futures.immediateFailedCheckedFuture(exception);
        }
    }
    else
    {
        ListenableFutureTask<V> listenablefuturetask = ListenableFutureTask.<V>create(callableToSchedule);

        synchronized (this.scheduledTasks)
        {
            this.scheduledTasks.add(listenablefuturetask);
            return listenablefuturetask;
        }
    }
}

private static void testOneSlice(final String id, final NavigableSet<Integer> test, final NavigableSet<Integer> canon, List<ListenableFuture<?>> results)
{
    ListenableFutureTask<?> f = ListenableFutureTask.create(new Runnable()
    {

        @Override
        public void run()
        {
            test(id + " Count", test.size(), canon.size());
            testEqual(id, test.iterator(), canon.iterator());
            testEqual(id + "->DSCI", test.descendingIterator(), canon.descendingIterator());
            testEqual(id + "->DSCS", test.descendingSet().iterator(), canon.descendingSet().iterator());
            testEqual(id + "->DSCS->DSCI", test.descendingSet().descendingIterator(), canon.descendingSet().descendingIterator());
        }
    }, null);
    results.add(f);
    COMPARE.execute(f);
}

private static void testOneSlice(final String id, final NavigableSet<Integer> test, final NavigableSet<Integer> canon, List<ListenableFuture<?>> results)
{
    ListenableFutureTask<?> f = ListenableFutureTask.create(new Runnable()
    {

        @Override
        public void run()
        {
            test(id + " Count", test.size(), canon.size());
            testEqual(id, test.iterator(), canon.iterator());
            testEqual(id + "->DSCI", test.descendingIterator(), canon.descendingIterator());
            testEqual(id + "->DSCS", test.descendingSet().iterator(), canon.descendingSet().iterator());
            testEqual(id + "->DSCS->DSCI", test.descendingSet().descendingIterator(), canon.descendingSet().descendingIterator());
        }
    }, null);
    results.add(f);
    COMPARE.execute(f);
}

private static void testOneSlice(final String id, final NavigableSet<Integer> test, final NavigableSet<Integer> canon, List<ListenableFuture<?>> results)
{
    ListenableFutureTask<?> f = ListenableFutureTask.create(new Runnable()
    {

        @Override
        public void run()
        {
            test(id + " Count", test.size(), canon.size());
            testEqual(id, test.iterator(), canon.iterator());
            testEqual(id + "->DSCI", test.descendingIterator(), canon.descendingIterator());
            testEqual(id + "->DSCS", test.descendingSet().iterator(), canon.descendingSet().iterator());
            testEqual(id + "->DSCS->DSCI", test.descendingSet().descendingIterator(), canon.descendingSet().descendingIterator());
        }
    }, null);
    results.add(f);
    COMPARE.execute(f);
}

public <V> ListenableFuture<V> a(Callable<V> callable) {
    Validate.notNull(callable);
    if (!this.isMainThread()) { // CraftBukkit && !this.isStopped()) {
        ListenableFutureTask listenablefuturetask = ListenableFutureTask.create(callable);
        Queue queue = this.j;

        // Spigot start
        this.j.add(listenablefuturetask);
        return listenablefuturetask;
        // Spigot end
    } else {
        try {
            return Futures.immediateFuture(callable.call());
        } catch (Exception exception) {
            return Futures.immediateFailedCheckedFuture(exception);
        }
    }
}

/**
 * Returns a future whose result is the state of the wavelet after it has been
 * loaded from storage. Any failure is reported as a
 * {@link PersistenceException}.
 */
@VisibleForTesting
static ListenableFuture<DeltaStoreBasedWaveletState> loadWaveletState(Executor executor,
    final DeltaStore deltaStore, final WaveletName waveletName, final Executor persistExecutor) {
  ListenableFutureTask<DeltaStoreBasedWaveletState> task =
      ListenableFutureTask.create(
         new Callable<DeltaStoreBasedWaveletState>() {
           @Override
           public DeltaStoreBasedWaveletState call() throws PersistenceException {
             return DeltaStoreBasedWaveletState.create(deltaStore.open(waveletName),
                                                              persistExecutor);
           }
         });
  executor.execute(task);
  return task;
}

@Override
public ListenableFuture<Void> onParticipantAdded(final WaveletName waveletName,
    ParticipantId participant) {
  Preconditions.checkNotNull(waveletName);
  Preconditions.checkNotNull(participant);

  ListenableFutureTask<Void> task = ListenableFutureTask.create(new Callable<Void>() {

    @Override
    public Void call() throws Exception {
      ReadableWaveletData waveletData;
      try {
        waveletData = waveletProvider.getReadableWaveletData(waveletName);
        updateIndex(waveletData);
      } catch (WaveServerException e) {
        LOG.log(Level.SEVERE, "Failed to update index for " + waveletName, e);
        throw e;
      }
      return null;
    }
  });
  executor.execute(task);
  return task;
}

@Override
public ListenableFuture<Void> onWaveInit(final WaveletName waveletName) {
  Preconditions.checkNotNull(waveletName);

  ListenableFutureTask<Void> task = ListenableFutureTask.create(new Callable<Void>() {

    @Override
    public Void call() throws Exception {
      ReadableWaveletData waveletData;
      try {
        waveletData = waveletProvider.getReadableWaveletData(waveletName);
        updateIndex(waveletData);
      } catch (WaveServerException e) {
        LOG.log(Level.SEVERE, "Failed to initialize index for " + waveletName, e);
        throw e;
      }
      return null;
    }
  });
  executor.execute(task);
  return task;
}

@Override
public ListenableFuture<Void> onWaveInit(final WaveletName waveletName) {

  ListenableFutureTask<Void> task = ListenableFutureTask.create(new Callable<Void>() {

    @Override
    public Void call() throws Exception {
      ReadableWaveletData waveletData;
      try {
        waveletData = waveletDataProvider.getReadableWaveletData(waveletName);
        updateIndex(waveletData);
      } catch (WaveServerException e) {
        LOG.log(Level.SEVERE, "Failed to initialize index for " + waveletName, e);
        throw e;
      }
      return null;
    }
  });
  executor.execute(task);
  return task;
}

@Override
public void waveletCommitted(final WaveletName waveletName, final HashedVersion version) {

  Preconditions.checkNotNull(waveletName);

  ListenableFutureTask<Void> task = ListenableFutureTask.create(new Callable<Void>() {

    @Override
    public Void call() throws Exception {
      ReadableWaveletData waveletData;
      try {
        waveletData = waveletDataProvider.getReadableWaveletData(waveletName);
        LOG.fine("commit " + version + " " + waveletData.getVersion());
        if (waveletData.getVersion() == version.getVersion()) {
          updateIndex(waveletData);
        }
      } catch (WaveServerException e) {
        LOG.log(Level.SEVERE, "Failed to update index for " + waveletName, e);
        throw e;
      }
      return null;
    }
  });
  executor.execute(task);
}

@Override
@Nonnull
public RequestVoteResponse requestVote(@Nonnull final RequestVote request) {

  checkNotNull(request);

  ListenableFutureTask<RequestVoteResponse> response = ListenableFutureTask.create(new Callable<RequestVoteResponse>() {
    @Override
    public RequestVoteResponse call() throws Exception {
      return delegate.requestVote(RaftStateContext.this, request);
    }
  });

  executor.execute(response);

  try {
    return response.get();
  } catch (Exception e) {
    throw Throwables.propagate(e);
  } finally {
    notifyRequestVote(request);
  }

}

@Override
@Nonnull
public AppendEntriesResponse appendEntries(@Nonnull final AppendEntries request) {

  checkNotNull(request);

  ListenableFutureTask<AppendEntriesResponse> response = ListenableFutureTask.create(new Callable<AppendEntriesResponse>() {
    @Override
    public AppendEntriesResponse call() throws Exception {
      return delegate.appendEntries(RaftStateContext.this, request);
    }
  });

  executor.execute(response);

  try {
    return response.get();
  } catch (Exception e) {
    throw Throwables.propagate(e);
  } finally {
    notifyAppendEntries(request);
  }

}

@Override
@Nonnull
public ListenableFuture<Object> commitOperation(@Nonnull final byte[] op) throws RaftException {

  checkNotNull(op);

  ListenableFutureTask<Object> response = ListenableFutureTask.create(new Callable<Object>() {
    @Override
    public Object call() throws Exception {
      return delegate.commitOperation(RaftStateContext.this, op);
    }
  });

  executor.execute(response);

  notifyCommit(op);

  return response;
}

@Override public ListenableFuture<FeatureRecord> reload(String key, FeatureRecord oldValue)
    throws Exception {

  final String[] storageKeySplit = FeatureStoreKeys.storageKeySplit(key);
  ListenableFutureTask<FeatureRecord>
      t = ListenableFutureTask.create(
      () -> httpLoad(storageKeySplit[0], storageKeySplit[1], oldValue));
  reloadExecutor.execute(t);
  return t;
}