Как дождаться завершения async Observable

Я пытаюсь создать образец с помощью rxjava. Образец должен организовать ReactiveWareService и ReactiveReviewService для повторного запуска композита WareAndReview.

ReactiveWareService
        public Observable<Ware> findWares() {
        return Observable.from(wareService.findWares());
    }

ReactiveReviewService: reviewService.findReviewsByItem does a ThreadSleep to simulate a latency!

public Observable<Review> findReviewsByItem(final String item) {
return Observable.create((Observable.OnSubscribe<Review>) observer -> executor.execute(() -> {
    try {
        List<Review> reviews = reviewService.findReviewsByItem(item);
        reviews.forEach(observer::onNext);
        observer.onCompleted();
    } catch (Exception e) {
        observer.onError(e);
    }
}));
}

public List<WareAndReview> findWaresWithReviews() throws RuntimeException {
final List<WareAndReview> wareAndReviews = new ArrayList<>();

wareService.findWares()
    .map(WareAndReview::new)
.subscribe(wr -> {
        wareAndReviews.add(wr);
        //Async!!!!
        reviewService.findReviewsByItem(wr.getWare().getItem())
            .subscribe(wr::addReview,
                throwable -> System.out.println("Error while trying to find reviews for " + wr)
            );
    }
);

//TODO: There should be a better way to wait for async reviewService.findReviewsByItem completion!
try {
    Thread.sleep(3000);
} catch (InterruptedException e) {}

return wareAndReviews;
}

Учитывая тот факт, что я не хочу возвращать Observable, как я могу дождаться завершения асинхронного Observable (findReviewsByItem)?


java rx-java
person user3753094    schedule 18.06.2014    source источник

Ответы (3)


arrow_upward
18
arrow_downward

Большую часть вашего примера можно переписать с помощью стандартных операторов RxJava, которые хорошо работают вместе:

public class Example {

    Scheduler scheduler = Schedulers.from(executor);

    public Observable<Review> findReviewsByItem(final String item) {
        return Observable.just(item)
               .subscribeOn(scheduler)
               .flatMapIterable(reviewService::findReviewsByItem);
    }
    public List<WareAndReview> findWaresWithReviews() {
        return wareService
               .findWares()
               .map(WareAndReview::new)
               .flatMap(wr -> {
                   return reviewService
                          .findReviewsByItem(wr.getWare().getItem())
                          .doOnNext(wr::addReview)
                          .lastOrDefault(null)
                          .map(v -> wr);
               })
               .toList()
               .toBlocking()
               .first();
    }
}

Всякий раз, когда вы хотите создавать сервисы, подобные этому, сначала подумайте о flatMap. Вам не нужно блокировать для каждого sub-Observable, но только в самом конце с toBlocking(), если это действительно необходимо.

person akarnokd    schedule 13.08.2015

arrow_upward
13
arrow_downward

Вы можете использовать методы из BlockingObservable, см. https://github.com/Netflix/RxJava/wiki/Blocking-Observable-Operators. например

BlockingObservable.from(reviewService.findReviewsByItem(..)).toIterable()
person Marek Hawrylczak    schedule 23.06.2014

arrow_upward
-5
arrow_downward

Другой способ — объявить CountdownLatch перед запуском. Затем вызовите countDown() для этой защелки в вашем onCompleted(). Затем вы можете заменить свой Thread.sleep() на await() на этой защелке.

person theBuckWheat    schedule 12.08.2015