在Project reactor中处理ListenableFuture

Question

我已经开始玩project reactor并想把我们的一个API移动到反应方式来做事情。我想知道如何处理像ListenableFuture这样的事情。

在我的例子中，我使用的是Cassandra，当我调用session.executeAsync()时，它返回一个扩展ListenableFuture的ResultSetFuture。下面是我现在已经编码的示例，我似乎不太乐意向客户公开ListenableFuture。

public Mono<ListenableFuture<Void>> save(Publisher<AccountDTO> accountPublisher) {
    return Mono.just(accountPublisher)
            .map(accountDTO -> {
                Account accountEntity = modelMapper.map(accountDTO, Account.class);
                return mappingManager.mapper(Account.class).saveAsync(accountEntity);
            })
            .retry(1)
            .doOnError(throwable -> log.error("Unable to create account "))
            .mapError(throwable -> new MyCustomException(""));
}

我的问题是：

公开ListenableFuture是一个好的实践吗，我个人不想把类似这样的东西还给客户机，在那里他们可以阻止。有没有一个更好的方法来处理这个项目反应堆，我可以返回一个单声道？

Answer 1

通过使用mono.create()工厂方法，您可以轻松地桥接ListenableFuture 异步API，以公开Mono 。该方法采用使用者 ，您将其作为lambda提供，该lambda：

1. 向未来添加一个调用sink.success()的成功侦听器（因为没有实际值，也可以使用侦听器接收的void值调用success(avid))
2. 向调用Sink的future添加失败侦听器。错误（失败）

差不多就是这样！请参阅关于create的参考文档（尽管这篇文章提到了flux版本，由于必须处理多个值，该版本稍微复杂一些）:http://projectreactor.io/docs/core/release/reference/docs/index.html#producing.create