observable_在Completablefuture和Observable之间转换

observable

Java 8中的CompletableFuture<T>是对T类型的值将来将可用的承诺的高级抽象。 Observable<T>非常相似，但是它承诺将来会出现任意数量的项，从0到无穷大。异步结果的这两种表示形式与仅使用一项即可使用Observable而不是CompletableFuture情况非常相似，反之亦然。另一方面， CompletableFuture更专业，并且由于它现在是JDK的一部分，因此应该很快流行起来。让我们用简短的文章来庆祝RxJava 1.0的发布，该文章展示了如何在不失去异步和事件驱动性质的情况下在两者之间进行转换。

从

CompletableFuture表示将来的一个值，因此将其变为Observable非常简单。当Future以某个值完成时， Observable也将立即发出该值并关闭流：

class FuturesTest extends Specification {public static final String MSG = "Don't panic"def 'should convert completed Future to completed Observable'() {given:CompletableFuture<String> future = CompletableFuture.completedFuture("Abc")when:Observable<String> observable = Futures.toObservable(future)then:observable.toBlocking().toIterable().toList() == ["Abc"]}def 'should convert failed Future into Observable with failure'() {given:CompletableFuture<String> future = failedFuture(new IllegalStateException(MSG))when:Observable<String> observable = Futures.toObservable(future)then:observable.onErrorReturn({ th -> th.message } as Func1).toBlocking().toIterable().toList() == [MSG]}  CompletableFuture failedFuture(Exception error) {CompletableFuture future = new CompletableFuture()future.completeExceptionally(error)return future}}

尚未执行的 Futures.toObservable()第一个测试会将Future转换为Observable ，并确保正确传播值。第二次测试创建了失败的Future ，将失败替换为异常的消息，并确保传播了异常。实现要短得多：

public static <T> Observable<T> toObservable(CompletableFuture<T> future) {return Observable.create(subscriber ->future.whenComplete((result, error) -> {if (error != null) {subscriber.onError(error);} else {subscriber.onNext(result);subscriber.onCompleted();}}));
}

注意： Observable.fromFuture()存在，但是我们想充分利用ComplatableFuture的异步运算符。

从

实际上，有两种将Observable转换为Future -创建CompletableFuture<List<T>>或CompletableFuture<T> （如果我们假设Observable只有一项）。让我们从前一种情况开始，用以下测试用例进行描述：

def 'should convert Observable with many items to Future of list'() {given:Observable<Integer> observable = Observable>just(1, 2, 3)when:CompletableFuture<List<Integer>> future = Futures>fromObservable(observable)then:future>get() == [1, 2, 3]
}def 'should return failed Future when after few items exception was emitted'() {given:Observable<Integer> observable = Observable>just(1, 2, 3)>concatWith(Observable>error(new IllegalStateException(MSG)))when:Futures>fromObservable(observable)then:def e = thrown(Exception)e>message == MSG
}

显然，直到源Observable信号流结束， Future才完成。因此， Observable.never()将永远不会完成包装Future ，而是用空列表完成它。该实现更短，更甜蜜：

public static <T> CompletableFuture<List<T>> fromObservable(Observable<T> observable) {final CompletableFuture<List<T>> future = new CompletableFuture<>();observable.doOnError(future::completeExceptionally).toList().forEach(future::complete);return future;
}

关键是Observable.toList() ，它可以方便地从Observable<T>和Observable<List<T>> 。当源Observable<T>完成时，后者发出List<T>类型的一项。

从

当我们知道CompletableFuture<T>将恰好返回一项时，就会发生上一次转换的特殊情况。在这种情况下，我们可以将其直接转换为CompletableFuture<T> ，而不是仅包含一项的CompletableFuture<List<T>> 。首先测试：

def 'should convert Observable with single item to Future'() {given:Observable<Integer> observable = Observable.just(1)when:CompletableFuture<Integer> future = Futures.fromSingleObservable(observable)then:future.get() == 1
}def 'should create failed Future when Observable fails'() {given:Observable<String> observable = Observable.<String> error(new IllegalStateException(MSG))when:Futures.fromSingleObservable(observable)then:def e = thrown(Exception)e.message == MSG
}def 'should fail when single Observable produces too many items'() {given:Observable<Integer> observable = Observable.just(1, 2)when:Futures.fromSingleObservable(observable)then:def e = thrown(Exception)e.message.contains("too many elements")
}

同样，实现非常简单并且几乎相同：

public static <T> CompletableFuture<T> fromSingleObservable(Observable<T> observable) {final CompletableFuture<T> future = new CompletableFuture<>();observable.doOnError(future::completeExceptionally).single().forEach(future::complete);return future;
}

上面的Helper方法还不够完善，但是，如果您需要在JDK 8和RxJava风格的异步计算之间进行转换，那么这篇文章应该足以帮助您入门。