Как связать два Completable в RxJava2

Вы ищете оператора andThen.

Возвращает Completable, которая сначала запускает эту Completable, а затем другую завершаемую.

firstCompletable
    .andThen(secondCompletable)

В общем, этот оператор является «заменой» flatMap на Completable:

Completable       andThen(CompletableSource next)
<T> Maybe<T>      andThen(MaybeSource<T> next)
<T> Observable<T> andThen(ObservableSource<T> next)
<T> Flowable<T>   andThen(Publisher<T> next)
<T> Single<T>     andThen(SingleSource<T> next)

TL; DR: в других ответах отсутствует тонкость. Используйте doThingA().andThen(doThingB()), если вам нужен эквивалент concat, используйте doThingA().andThen(Completable.defer(() -> doThingB()), если вам нужен эквивалент flatMap.

РЕДАКТИРОВАТЬ: более полная ссылка

• flatMap() - версия отображения merge()
• concatMap() - версия отображения concat()
• Для Completable вам нужно defer(), чтобы сделать вызовы функций ленивыми, как в функциях сопоставления для Single или Observable (или, желательно, чтобы ничего не происходило, пока вы не нажмете подписку - это хорошее соглашение, которому следует следовать, и оно используется в официальных библиотеках Rx как как и любые расширения Rx, с которыми я сталкивался, для продвинутых пользователей это относится только к холодным завершаемым таблицам, но большинство людей могут это игнорировать).
• единственная разница между concat(a, b) и a.andThen(b) - синтаксис

Некоторые примеры:

• foo(a).andThen(bar(b)) будет:

  1. call foo(a)
  2. немедленно вызвать bar(b), даже если завершенная таблица, возвращенная на шаге 1, возвращает ошибку
  3. подписаться на любой возвращаемый шаг 1
  4. После этого подпишется на результат bar(b) , только если последний шаг выполнен успешно

• foo(a).andThen(Completable.defer(() -> bar(b)) будет:

  1. call foo(a)
  2. подписаться на результат шага 1
  3. только в том случае, если завершение, возвращаемое функцией foo(a), выполнено успешно, вызывается bar(b)

Я собираюсь опустить рассмотрение merge(), поскольку он становится немного сложнее, но короче говоря, это то, что нужно называть, если вы хотите «параллелизма».

Приведенные выше ответы в некотором роде верны, но я счел их вводящими в заблуждение, потому что они упускают тонкость, касающуюся нетерпеливой оценки.

doThingA().andThen(doThingB()) вызовет doThingB() немедленно, но подпишется на наблюдаемое, возвращаемое doThingB(), когда завершится наблюдаемое, возвращенное doThingA().

doThingA().andThen(Completable.defer(() -> doThingB()) вызовет doThingB() только после того, как A завершится.

Это важно, только если doThingB() имеет побочные эффекты перед событием подписки. Например. Single.just(sideEffect(1)).toCompletable()

Реализация, которая не имеет побочных эффектов перед событием подписки (истинное холодное наблюдение), может быть Single.just(1).doOnSuccess(i -> sideEffect(i)).toCompletable().

В случае, если меня только что укусило, вещь A - это некоторая логика проверки, и doThingB() немедленно запускает асинхронное обновление базы данных, которое завершает VertX ObservableFuture. Это плохо. Возможно, doThingB() должен быть написан так, чтобы обновлять базу данных только при подписке, и я собираюсь попытаться спроектировать вещи таким образом в будущем.

Пытаться

Completable.concat

Returns a Completable which completes only when all sources complete, one after another.

http://reactivex.io/RxJava/javadoc/io/reactivex/Completable.html#concat(java.lang.Iterable)

У меня была такая же проблема, и мне пришлось использовать оператор .concactWith, чтобы он работал. В моем случае у меня есть два развлечения типа Completable.

fun makeTwoThings(): Completable {
    makeFirstThing().concatWith(makeSecondThing())
}

fun makeFirstThing(): Completable{
     //TODO()
}

fun makeSecondThing(): Completable{
     //TODO()
}

Обратите внимание, что ответ, за который здесь больше голосов, немного вводит в заблуждение. См. Пример ниже, моя идея состоит в том, чтобы показать некоторые сценарии тестирования и показать, как ведет себя завершаемая логика с оператором andThen.

 private fun doThingAFail(): Completable {
        print("Do thingA Called\n")
        return Completable.fromCallable {
            print("calling stream A\n")
            throw(Exception("The excep"))
        }
    }

    private fun doThingB(): Completable {
        print("Do thingB Called\n")
        return Completable.fromCallable {
            print("calling stream B\n")

        }
    }

    private fun doThingA(): Completable {
        print("Do thingA Called\n")
        return Completable.fromCallable {
            print("calling stream A\n")
        }
    }

Обратите внимание на это для теста ниже:

@Test
fun testCallAPlusB() {
    doThingA().andThen(doThingB())
}

Краткое примечание: обратите внимание, что мы не подписываемся на эти Completables в этом фрагменте.

вывод будет:

Do thingA Called
Do thingB Called

Для теста:

@Test
fun theTestSubscribe() {
    doThingA().andThen(doThingB()).subscribe()
}

Результатом будет:

Do thingA Called
Do thingB Called
calling stream A
calling stream B

И, наконец, в случае сбоя первой таблицы завершения вторая таблица завершения не будет выполняться.

@Test
fun theTestFailThingA() {
    doThingAFail().andThen(doThingB()).subscribe()
}

вывод будет:

Do thingA Called
Do thingB Called
calling stream A

Ключевой концепцией здесь является то, что логика внутри метода и внутри наблюдаемого выполняется не одновременно. Строки Do thingA Called и Do thingB Called будут напечатаны после того, как мы вызовем методы doThingA() и doThingB(). В то время как линии вызывающего потока A и вызывающего потока B будут вызываться только тогда, когда кто-то подписывается на методы doThingA и doThingB.

Вторая концепция заключается в том, как оператор andThen будет обрабатывать ошибки. В приведенном выше примере, если doThingA() completable завершается с ошибкой, поток завершится и не будет печатать строку B вызывающего потока.