import scala.compat.java8.FutureConverters; //导入依赖的package包/类
private void onProducerRemoved(final ProducerRemoved message) {
    LOG.debug("Received ProducerRemoved: {}", message);

    final List<CompletableFuture<Object>> futures = new ArrayList<>();

    for (final String address : resolver.getShardingServicePeerActorAddresses()) {
        final ActorSelection selection = actorSystem.actorSelection(address);

        futures.add(FutureConverters.toJava(
                actorContext.executeOperationAsync(selection, new NotifyProducerRemoved(message.getSubtrees())))
                .toCompletableFuture());
    }

    final CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(
            futures.toArray(new CompletableFuture[futures.size()]));

    final ActorRef respondTo = getSender();

    combinedFuture
            .thenRun(() -> respondTo.tell(new Status.Success(null), self()))
            .exceptionally(e -> {
                respondTo.tell(new Status.Failure(null), self());
                return null;
            });

}
 

import scala.compat.java8.FutureConverters; //导入依赖的package包/类
@Override
public CompletionStage<Void> close() {
    // first despawn on the local node
    distributedShardedDOMDataTree.despawnShardFrontend(prefix);
    // update the config so the remote nodes are updated
    final Future<Object> ask =
            Patterns.ask(shardedDataTreeActor, new PrefixShardRemovalLookup(prefix), SHARD_FUTURE_TIMEOUT);

    final Future<Void> closeFuture = ask.transform(
            new Mapper<Object, Void>() {
                @Override
                public Void apply(final Object parameter) {
                    return null;
                }
            },
            new Mapper<Throwable, Throwable>() {
                @Override
                public Throwable apply(final Throwable throwable) {
                    return throwable;
                }
            }, actorSystem.dispatcher());

    return FutureConverters.toJava(closeFuture);
}
 

import scala.compat.java8.FutureConverters; //导入依赖的package包/类
@Bean
public ChannelsRegistry channelsRegistry() {
    ChannelsActors channelsActors = channelsActors();
    return tempo -> {
        Future<Object> result = Patterns.ask(channelsActors.get(tempo), ListChannels.instance(), 1000);
        CompletionStage<Object> stage = FutureConverters.toJava(result);
        return stage.thenCompose(o -> {
            if (o instanceof Channels) {
                return CompletableFuture.completedFuture(((Channels) o).getChannels());
            }
            else {
                CompletableFuture<List<Channel>> f = new CompletableFuture<>();
                f.completeExceptionally(new IllegalStateException());
                return f;
            }
        });
    };
}
 

import scala.compat.java8.FutureConverters; //导入依赖的package包/类
private Route createRoute(ActorRef streamActor) {
    return route(
            path(segment("services").slash(longSegment()), id ->
                    get(() -> {
                        Promise<HttpResponse> promise = Futures.promise();
                        streamActor.tell(new Pair(id, promise), ActorRef.noSender());
                        return completeWithFuture(FutureConverters.toJava(promise.future()));
                    })));
}
 

import scala.compat.java8.FutureConverters; //导入依赖的package包/类
protected final ShardState resolveBackendInfo(final String shardName, final long cookie) {
    LOG.debug("Resolving cookie {} to shard {}", cookie, shardName);

    final CompletableFuture<ShardBackendInfo> future = new CompletableFuture<>();
    FutureConverters.toJava(actorContext.findPrimaryShardAsync(shardName)).whenComplete((info, failure) -> {
        if (failure == null) {
            connectShard(shardName, cookie, info, future);
            return;
        }

        LOG.debug("Shard {} failed to resolve", shardName, failure);
        if (failure instanceof NoShardLeaderException) {
            future.completeExceptionally(wrap("Shard has no current leader", failure));
        } else if (failure instanceof NotInitializedException) {
            // FIXME: this actually is an exception we can retry on
            LOG.info("Shard {} has not initialized yet", shardName);
            future.completeExceptionally(failure);
        } else if (failure instanceof PrimaryNotFoundException) {
            LOG.info("Failed to find primary for shard {}", shardName);
            future.completeExceptionally(failure);
        } else {
            future.completeExceptionally(failure);
        }
    });

    return new ShardState(future);
}
 

import scala.compat.java8.FutureConverters; //导入依赖的package包/类
private void connectShard(final String shardName, final long cookie, final PrimaryShardInfo info,
        final CompletableFuture<ShardBackendInfo> future) {
    LOG.debug("Shard {} resolved to {}, attempting to connect", shardName, info);

    FutureConverters.toJava(ExplicitAsk.ask(info.getPrimaryShardActor(), connectFunction, CONNECT_TIMEOUT))
        .whenComplete((response, failure) -> {
            onConnectResponse(shardName, cookie, future, response, failure);
        });
}
 

import scala.compat.java8.FutureConverters; //导入依赖的package包/类
private void onProducerCreated(final ProducerCreated message) {
    LOG.debug("Received ProducerCreated: {}", message);

    // fastpath if we have no peers
    if (resolver.getShardingServicePeerActorAddresses().isEmpty()) {
        getSender().tell(new Status.Success(null), noSender());
    }

    final ActorRef sender = getSender();
    final Collection<DOMDataTreeIdentifier> subtrees = message.getSubtrees();

    final List<CompletableFuture<Object>> futures = new ArrayList<>();

    for (final String address : resolver.getShardingServicePeerActorAddresses()) {
        final ActorSelection actorSelection = actorSystem.actorSelection(address);
        futures.add(
                FutureConverters.toJava(
                        actorContext.executeOperationAsync(
                                actorSelection, new NotifyProducerCreated(subtrees), DEFAULT_ASK_TIMEOUT))
                .toCompletableFuture());
    }

    final CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(
            futures.toArray(new CompletableFuture[futures.size()]));

    combinedFuture
            .thenRun(() -> sender.tell(new Success(null), noSender()))
            .exceptionally(throwable -> {
                sender.tell(new Status.Failure(throwable), self());
                return null;
            });
}
 

import scala.compat.java8.FutureConverters; //导入依赖的package包/类
private CompletionStage<HttpResponse> dispatchHttpRequest(final HttpRequest request, final String actorName) {
    final Future<ActorRef> refFuture = _actorSystem.actorSelection(actorName)
            .resolveOne(FiniteDuration.create(1, TimeUnit.SECONDS));
    return FutureConverters.toJava(refFuture).thenCompose(
            ref -> {
                final CompletableFuture<HttpResponse> response = new CompletableFuture<>();
                ref.tell(new RequestReply(request, response), ActorRef.noSender());
                return response;
            })
            // We return 404 here since actor startup is controlled by config and
            // the actors may not be running.
            .exceptionally(err -> HttpResponse.create().withStatus(404));
}
 

import scala.compat.java8.FutureConverters; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private <T> CompletionStage<T> ask(final String actorPath, final Object request, final T defaultValue) {
    return FutureConverters.toJava(
            (Future<T>) Patterns.ask(
                    _actorSystem.actorSelection(actorPath),
                    request,
                    Timeout.apply(1, TimeUnit.SECONDS)))
            .exceptionally(throwable -> defaultValue);
}
 

import scala.compat.java8.FutureConverters; //导入依赖的package包/类
public CompletionStage<ActorRef> createUserActor(String id, ActorRef webSocketOut) {
    // Use guice assisted injection to instantiate and configure the child actor.
    long timeoutMillis = 100L;
    return FutureConverters.toJava(
            ask(userParentActor, new UserParentActor.Create(id, webSocketOut), timeoutMillis)
    ).thenApply(stageObj -> (ActorRef) stageObj);
}
 

import scala.compat.java8.FutureConverters; //导入依赖的package包/类
protected static <T> T waitOnAsyncTask(final CompletionStage<T> completionStage, final Duration timeout)
        throws Exception {
    return Await.result(FutureConverters.toScala(completionStage), timeout);
}