7

RxJava と Java 8 の CompletableFuture クラスを試していますが、タイムアウト条件を処理する方法がよくわかりません。

import static net.javacrumbs.futureconverter.java8rx.FutureConverter.toObservable;

// ...

    Observable<String> doSomethingSlowly() {
        CompletableFuture<PaymentResult> task = CompletableFuture.supplyAsync(() -> {
            // this call may be very slow - if it takes too long, 
            // we want to time out and cancel it.
            return processor.slowExternalCall();

        });

        return toObservable(task);
    }

    // ...

    doSomethingSlowly()
        .single()
        .timeout(3, TimeUnit.SECONDS, Observable.just("timeout"));

これは基本的に機能します (3 秒のタイムアウトに達すると、「タイムアウト」が発行されます)。ただし、ラップした将来のタスクをキャンセルしたいObservable場合もあります-RxJava中心のアプローチでそれは可能ですか?

1つのオプションは、を使用して自分でタイムアウトを処理することであることは知っていますがtask.get(3, TimeUnit.SECONDS)、RxJavaですべてのタスク処理を行うことが可能かどうか疑問に思います。

4

1 に答える 1

11

はい、できます。に を追加しSubscriptionますSubscriber。

subscribe().unsubscribe()これにより、明示的に呼び出した場合、Observableまたは が正常に完了した場合、またはエラーが発生した場合に発生するサブスクリプション解除をリッスンできます。

future が完了する前にサブスクリプション解除が表示された場合は、明示的unsubscribeまたはタイムアウトのいずれかが原因であると推測できます。

public class FutureTest {
    public static void main(String[] args) throws IOException {
        doSomethingSlowly()
                .timeout(1, TimeUnit.SECONDS, Observable.just("timeout"))
                .subscribe(System.out::println);
        System.in.read(); // keep process alive
    }

    private static Observable<String> doSomethingSlowly() {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
            }
            return "Something";

        });
        return toObservable(future);
    }

    private static <T> Observable<T> toObservable(CompletableFuture<T> future) {
        return Observable.create(subscriber -> {
            subscriber.add(new Subscription() {
                private boolean unsubscribed = false;
                @Override
                public void unsubscribe() {
                    if (!future.isDone()){
                        future.cancel(true);
                    }
                    unsubscribed = true;
                }

                @Override
                public boolean isUnsubscribed() {
                    return unsubscribed;
                }
            });

            future.thenAccept(value -> {
                if (!subscriber.isUnsubscribed()){
                    subscriber.onNext(value);
                    subscriber.onCompleted();
                }
            }).exceptionally(throwable -> {
                if (!subscriber.isUnsubscribed()) {
                    subscriber.onError(throwable);
                }
                return null;
            });
        });
    }
}
于 2015-04-18T13:55:37.903 に答える