RxScala doOnCompleted не запускается после вызова .take ()

Question

Я новичок в RxScala Observables и испытываю странное поведение при использовании комбинации take (n) и doOnCompleted ().

Ниже приведен пример теста, в котором я считаю, что первая подписка верна (с take (2) в начале) и выводит все журналы для методов подписки, следующего, завершенного и отмены подписки, однако второй тест ( take (2) после doOnComplete) никогда не достигает методов doOnCompleted.

import rx.lang.scala.Observable

object Tester extends App {

    val obs = Observable.from(List(1,2,3,4))

    val obsAddMethodsCorrect = obs.take(2)
        .doOnSubscribe( println("subscribe") )
        .doOnNext( n => println(s"next $n") )
        .doOnError( e => println("error") )
        .doOnCompleted( println("completed") )
        .doOnUnsubscribe( println("unsubscribe") )

    val obsAddMethodsInCorrect = obs
        .doOnError( e => println("error") )
        .doOnCompleted( println("completed") )
        .take(2)
        .doOnNext( n => println(s"next $n") )
        .doOnUnsubscribe( println("unsubscribe") )
        .doOnSubscribe( println("subscribe") )

    obsAddMethodsCorrect.toBlocking.subscribe()
    println("")
    println("The above seems correct. Below seems incorrect")
    println("")
    obsAddMethodsInCorrect.toBlocking.subscribe()

}

токовый выход вышеуказанного теста ниже.

subscribe
next 1
next 2
completed
unsubscribe

The above seems correct. Below seems incorrect

subscribe
next 1
next 2
unsubscribe

Почему во втором примере не запускается doOnCompleted ()?

Answer 1

Оператор

doOnCompleted() вызовет свое действие при возникновении события onCompleted(). Однако, если вы отмените подписку от наблюдателя до того, как он закончит выделять элементы, onCompleted() не будет запущено. Цепочка наблюдателей фактически отменяется.