RxJava2 Как дождаться завершения работы нескольких Observer?

Question

[Neonoviiwolf]

Доброго
Есть несколько Maybe и Observable, мне нужно дождаться, когда у всех у них сработает subscribe, как это сделать?

Answer 1

[Денис Загаевский]

toObservable()
toCompletable
Completable#merge

Этого должно хватить, ну и, может быть, ещё скипнуть элементы.

Comments

[Neonoviiwolf]

из Maybe можно Observable, но не Completable, во всяком случаи хз как
Да и merge пробовал, но в нём все потоки объединяются, а мне нужно знать какой поток что вернул

[Денис Загаевский]

Neonoviiwolf, разворачивай тогда вопрос. Сколько их, что должно случиться, если кто-то из них ничего не вернет, и просто завершится? subscribe это не complete, туда может несколько значений прилетать.
maybe.toObservable().toCompletable()

[Neonoviiwolf]

maybe.toObservable() - можно
maybe.toObservable().toCompletable() - нельзя

Нету в выборе

Что хочу: показывать экран загрузки, пока получаю данные с сервера.
Что делаю:
Есть Room, спрашиваю когда было последнее обновление, если бд пустая, то дата по умолчанию

db.getTitleProductDAO().getLastDate()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeOn(Schedulers.io())
                .subscribe(s -> dateWrite = s, throwable -> Log.d(TAG, "DownloadDataServer: " + throwable));
/*когда получаем dateWrite, обращаемся к серверу/серверам*/
        if (dateWrite == null)
            dateWrite = "2000-01-01";
        networkService.getJSONApiTittleProduct()
                .getPostWithID(dateWrite)
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .subscribe(postTittleProducts -> {
                    for (PostTittleProduct i : postTittleProducts) {
                        TitleProduct titleProduct = TitleProduct.builder()
                                .id_product(i.getId_product())
                                .product_name(i.getProduct_name_ru())
                                .product_name_low(i.getProduct_name_ru().toLowerCase())
                                .date_write(i.getDate_write())
                                .build();
                        db.getTitleProductDAO().add(titleProduct);
                    }
                }, throwable -> Log.d(TAG, "getJSONApiTittleProduct: " + throwable));

аналогичных запрос сейчас 5, будет больше

это запрос в бд

/* дата последнейзаписи */
    @Query("Select date_write from title_product where date_write = (select max(date_write) " +
                   "from  title_product)")
    Maybe<String> getLastDate();

таких запросов несколько соответственно. Когда данные будут получены, мне нужно сообщить презентору, что данные загружены и он должен показать новое View. Но вот тут 2 проблемы - не факт, что я получу дату, до того, как будет отправлен запрос на сервер, второе, мне нужно показать, что загрузка завершена или ошибка, когда отработают все.

пока стоит заглушка для модели

public class ModelMainActivity implements iModelMainActivity {
    private static final String TAG = "ModelMainActivity";


    public ModelMainActivity() {
        MainApp.app().appComponent().inject(this);
/**тут качаем с сервера - нужно сделать это нормально**/
        new DownloadDataServer();
    }

    @Override
    public Observable updateDB() {
        return Observable.create(new ObservableOnSubscribe<Object>() {
            @Override
            public void subscribe(ObservableEmitter<Object> emitter) throws Exception {
                Thread.sleep(2000);
                emitter.onNext("");
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
    }
}

по поводу жизни презентера, то стоит Moxy, так что всё должно работать нормально
п.с. знаю, что многое можно сделать лучше, но пока не могу

[Денис Загаевский]

Омг, подписка в подписке. *facepalm*
используй switch/flatMap.
toCompletable я спутал с первой rxjava, кажется. Называется ignoreElements.

[Neonoviiwolf]

Денис Загаевский, не, нету подписки в подписке.

[Neonoviiwolf]

flatMap - похоже на нужное