RxJava Zip два наблюдаемых без их завершения

У меня есть две наблюдаемые: первая из библиотеки RxAndroidBle:

Observable<RxBleConnection> bluetoothObservable = RxBleClient.create(getBaseContext()).getBleDevice(macAddress)
.establishConnection(false)

Который подключается к устройству и поддерживает соединение, пока у него есть подписчики, а другой

Observable<Response> serverObservable = Observable.fromCallable(() -> callServer())

Затем я застегиваю их вместе

bluetoothObservable.zipWith(serverObservable , (rxBleConnection, s) -> {
                                Log.d(TAG, "zip done");
                                return "mock result";
                            }).subscribe((s) -> {},
                                    Throwable::printStackTrace);

Но после зипа bluetoothObservable отписывается и связь сразу падает. Что я должен сделать, чтобы заархивировать эти наблюдаемые данные и сохранить bluetoothObservable живым/подписанным?


rx-java rxandroidble
person Nikita Khlebushkin    schedule 05.07.2017    source источник

Ответы (1)


arrow_upward
2
arrow_downward

Вместо .zip() вы можете использовать:

Observable.combineLatest(
  bluetoothObservable, 
  serverObservable,
  (rxBleConnection, s) -> {
    Log.d(TAG, "combined");
    return "mock result";
  }
)
  .subscribe(
    (s) -> {},
    Throwable::printStackTrace
  )

Объяснение: zip пытается объединить выходные данные обоих Observable один за другим. Если одна из них завершится и все предыдущие эмиссии из этого Observable совпадут — нет смысла сохранять подписку на другие Observable, так как последующие эмиссии использоваться не будут. combineLatest просто пытается объединить все выбросы от обоих Observable в пары, которые являются последними.

person Dariusz Seweryn    schedule 05.07.2017
comment
Спасибо, попробую, потом отчитаюсь - person Nikita Khlebushkin; 05.07.2017
comment
Нет, боюсь, bluetoothObservable все еще отписывается - person Nikita Khlebushkin; 05.07.2017
comment
Я только что проверил точно такой же код, и он у меня работает, поэтому возможны две ситуации: поток генерирует исключение или ваш bluetoothObservable завершается сам по себе. - person Dariusz Seweryn; 05.07.2017
comment
Да, вы правы, выбрасывается исключение. Спасибо за ваш ответ! - person Nikita Khlebushkin; 05.07.2017