Add CompletableFuture converters to armeria-rxjava

[anuraaga]

I wrote my first code using armeria-rxjava and I was surprised that despite us having official integration, I still needed another library, future-converter to actually use it with Armeria, which heavily uses CompletableFuture. Should we expose these converters in armeria-rxjava itself? We could probably just shade that library in, or reimplement since it's not a lot of code.

[ikhoon]

The future-converter is an interesting library. https://github.com/lukas-krecan/future-converter

[ikhoon]

I'm working on this issue.

I tested FutureConverter.toSingle(cf) to reimplemente it in Armeria. It emits a ProtocolViolationException exception when I specify a Scheduler to Single.

ExecutorService pool = Executors.newCachedThreadPool();

// Works
Single.just(10)
      .flatMap(i -> Single.just(i + 10))
      .subscribeOn(Schedulers.from(pool))
      .subscribe(i -> {}, ex -> logger.error("ex", ex));

// Not work
final CompletableFuture<Integer> completableFuture = new CompletableFuture<>();
FutureConverter.toSingle(completableFuture)
               .flatMap(i -> Single.just(i + 10))
               .subscribeOn(Schedulers.from(pool))
               .subscribe(i -> {}, ex -> logger.error("ex", ex));
completableFuture.complete(20);

// Works
final CompletableFuture<Integer> completableFuture = new CompletableFuture<>();
FutureConverter.toSingle(completableFuture)
               .flatMap(i -> Single.just(i + 10))
               .subscribe(i -> {}, ex -> logger.error("ex", ex));
completableFuture.complete(20);

The error messages:

io.reactivex.exceptions.ProtocolViolationException: Disposable already set!
	at io.reactivex.internal.disposables.DisposableHelper.reportDisposableSet(DisposableHelper.java:156)
	at io.reactivex.internal.disposables.DisposableHelper.setOnce(DisposableHelper.java:82)
	at io.reactivex.internal.operators.single.SingleFlatMap$SingleFlatMapCallback.onSubscribe(SingleFlatMap.java:66)
	at net.javacrumbs.futureconverter.rxjava2common.RxJava2FutureUtils$ValueSourceBackedSingle.subscribeActual(RxJava2FutureUtils.java:93)
	at io.reactivex.Single.subscribe(Single.java:3666)
	at io.reactivex.internal.operators.single.SingleFlatMap.subscribeActual(SingleFlatMap.java:36)
	at io.reactivex.Single.subscribe(Single.java:3666)
	at io.reactivex.internal.operators.single.SingleSubscribeOn$SubscribeOnObserver.run(SingleSubscribeOn.java:89)
	at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:38)
	at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:26)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:835)
Exception in thread "pool-1-thread-1" io.reactivex.exceptions.ProtocolViolationException: Disposable already set!
	at io.reactivex.internal.disposables.DisposableHelper.reportDisposableSet(DisposableHelper.java:156)
	at io.reactivex.internal.disposables.DisposableHelper.setOnce(DisposableHelper.java:82)
	at io.reactivex.internal.operators.single.SingleFlatMap$SingleFlatMapCallback.onSubscribe(SingleFlatMap.java:66)
	at net.javacrumbs.futureconverter.rxjava2common.RxJava2FutureUtils$ValueSourceBackedSingle.subscribeActual(RxJava2FutureUtils.java:93)
	at io.reactivex.Single.subscribe(Single.java:3666)
	at io.reactivex.internal.operators.single.SingleFlatMap.subscribeActual(SingleFlatMap.java:36)
	at io.reactivex.Single.subscribe(Single.java:3666)
	at io.reactivex.internal.operators.single.SingleSubscribeOn$SubscribeOnObserver.run(SingleSubscribeOn.java:89)
	at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:38)
	at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:26)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:835)

[ikhoon]

While implementing the CompletableFuture converter, I found the the RxJava3 plans to raise the base line to Java8. https://twitter.com/rxjava/status/1211936600017555456?s=21
It has built-in Java8 interop features. https://github.com/ReactiveX/RxJava/blob/3.x/src/main/java/io/reactivex/rxjava3/internal/jdk8/SingleFromCompletionStage.java

[ikhoon]

I think we don't need to add our own SingleFromCompletionStage now.
RxJava 3 will be released at the end of January 2020. The builtin SingleFromCompletionStage will be exposed to armeria-rxjava with RxJava 3.

These are useful links for RxJava 3:

• RxJavaBridge - Bridge between RxJava 2 and RxJava 3
• What's different in 3.0
• 3.0.0-RC7 release note
• RxJava 3 release plan
• RxJava 3 jdk8 module

Let me resume this issue after releasing RxJava 3. 😀

[anuraaga]

Thanks for the investigation @ikhoon! I wonder if armeria-rxjava will be able to support both 2/3 at the same time or will we have two modules. But either way, I agree we don't have to worry about it now.

[ikhoon]

I think armeria-rxjava could rename to armeria-rxjava2 and armeria-rxjava can support RxJava 3. Maybe it will happen before releasing Armeria 1.0.0.

[ikhoon]

RxJava 3 has been released! https://github.com/ReactiveX/RxJava/releases/tag/v3.0.0

Good time to resume this. 😀