从返回未来的服务创建Mono/Flux的正确方法

Question

如何正确处理由期货构建的Monos？

我试着让我的头脑围绕着Spring Reactive（和Spring 5），观看所有的视频，阅读所有我能找到的博客，但他们似乎都没有做一些事情，而不仅仅是查询数据库或其他琐碎的事情。

我正在使用新的AWS 2.0开发工具包，它使用CompletableFuture的用于大多数事情。使用服务创建新实例，我的方法如下所示

public Mono<RunInstancesResponse> create(Instance instance) {
    RunInstancesRequest runInstancesRequest = RunInstancesRequest.builder()
            .instanceType(instance.getInstanceType())
            .imageId(instance.getImageId())
            .securityGroupIds(instance.getSecurityGroupIds())
            .keyName(instance.getKeyName())
            .minCount(1)
            .maxCount(1)
            .tagSpecifications(createTags(instance))
            .build();

    CompletableFuture<RunInstancesResponse> future = client.runInstances(runInstancesRequest);

    future.whenComplete((response, error) -> {
        response.reservation().instances().stream().map(aws -> Instance.builder()
                .imageId(aws.imageId())
                .build()
        ).forEach(instanceRepository::save);

    });

    return Mono.fromFuture(future);
}

我在这里的理解是，我几乎立即返回RunInstancesACK类型的Mono，而future.whenComplete将随时执行它的操作。

我从我的路由处理器调用它，它看起来像

public Mono<ServerResponse> create(ServerRequest request) {
    return request.bodyToMono(Instance.class)
            .flatMap(createService::create)
            .flatMap(i -> ServerResponse.accepted().build());
}

现在这几乎像我预期的那样工作，但是有几个关键的事情是错误的，我不知道如何解决。

1.）当Complete从未被调用过，我相信这是因为我没有订阅它。

2.）服务器在完成之前不会响应客户端（大约2.5秒），这并不理想，因为我希望它立即响应，然后在调用当完成时更新客户端。

我感觉我的整个服务和处理程序完全是错误的。

我想举一些例子，说明我应该如何处理从＜code＞Mono＜code＞或＜code＞Flux＜code＞类型的路由处理程序调用的服务中的未来。

Answer 1

我编写了一个包装SQS、SNS和DynamoDb的开源库，以使其更容易一些。为了避免仅链接的答案，您可以对其应用以下内容：

public Mono<ChangeMessageVisibilityResult> changeMessageVisibility(String queueUrl, String receiptHandle, Integer visibilityTimeout) {
    return Mono.create(subscriber -> amazonClient.changeMessageVisibilityAsync(queueUrl, receiptHandle, visibilityTimeout, AmazonWebServiceRequestAsyncHandler.valueEmittingHandlerFor(subscriber)));
  }

传递的处理程序在 2 个世界之间进行转换：

public class AmazonWebServiceRequestAsyncHandler<RQ extends AmazonWebServiceRequest, RS> implements AsyncHandler<RQ, RS> {
    private final MonoSink<? super RS> subscriber;
    private boolean shouldEmitValue;

    private AmazonWebServiceRequestAsyncHandler(MonoSink<? super RS> subscriber, boolean shouldEmitValue) {
        this.subscriber = subscriber;
        this.shouldEmitValue = shouldEmitValue;
    }

    @Override
    public void onError(Exception exception) {
        subscriber.error(exception);
    }

    @Override
    public void onSuccess(RQ request, RS response) {
        if (shouldEmitValue) {
            subscriber.success(response);
        } else {
            subscriber.success();
        }
    }

    public static <RQ extends AmazonWebServiceRequest, RS> AsyncHandler<RQ, RS> valueEmittingHandlerFor(final MonoSink<? super RS> subscriber) {
        return new AmazonWebServiceRequestAsyncHandler<>(subscriber, true);
    }

    public static <RQ extends AmazonWebServiceRequest> AsyncHandler<RQ, Void> voidHandlerFor(MonoSink<? super Void> subscriber) {
        return new AmazonWebServiceRequestAsyncHandler<>(subscriber, false);
    }
}