java - RxJava等待单个事件，RxJava

Question

如何等待一个带有超时的可观察值出现？

我正在寻找类似的东西:

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

Answer 1

首先，将 Observable 转换为 CompletableFuture，如 Converting between Completablefuture and Observable 中所述。 :

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。