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

Question

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

Ниже у меня есть пример теста, в котором я считаю, что первая подписка правильная (с take(2) в начале) и выводит все журналы для методов подписки, next, complete и unsubscribe, однако второй тест (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() не уволен. Цепочка наблюдателей фактически отменяется.