java - RxJava等待单个事件，RxJava

您如何等待带有timout的可观察值出现在单个值上？

我正在寻找类似的东西：

Observable<Acknowledgement> acknowledgementObservable;
port.send(new Message());
Optional<Acknowledgement> ack = acknowledgementObservable.getFirst(100, TimeUnit.MILLISECONDS);

最佳答案

首先，按照Converting between Completablefuture and Observable中的说明将Observable转换为CompletableFuture：

Observable<Acknowledgement> acknowledgementObservable;
port.send(new Message());
CompletableFuture<T> future = new CompletableFuture<>();
acknowledgementObservable
    .doOnError(future::completeExceptionally)
    .single()
    .forEach(future::complete);

然后，使用超时等待事件：

Acknowledgement ack = future.get(100, TimeUnit.MILLISECONDS);

如果发生超时，它将抛出TimeoutException。