RxJava等待单个事件，RxJava

Question

你如何等待一个值出现在一个可以观察到的时间点上？

我在找这样的东西：

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

Answer 1

可以通过添加.timeout()和.onErrorComplete() .blockingGet()来做到这一点。

因此，在这个例子中，应该是：

    Acknowledgement ack = acknowledgementObservable
                              .firstElement()
                              .timeout(3, TimeUnit.SECONDS)
                              .onErrorComplete()
                              .blockingGet();

如果超时被击中，ack将为null。

Answer 2

首先，将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。