RxJava2: onComplete не вызывается с flatMapIterable

Question

Вот короткий фрагмент кода:

    val subject = BehaviorSubject.createDefault(emptyList<Int>())
    subject.onNext(Arrays.asList(1, 2, 3))
    subject.flatMapIterable { list: List<Int> -> list }
            .subscribeBy(
                    onNext = { l("on next", it) },
                    onComplete = { l("on complete") }
            )

Зачем onComplete не звонит сюда? Что я должен сделать для работы этого кода? Потому что в оригинальном коде я не могу использовать .toList() метод.

Answer 1

BehaviorSubject Вы имеете в своей форме бесконечный источник, если onComplete называется на это. Следовательно flatMapIterable не будет завершена, и ваш onComplete обработчик никогда не будет вызван.

Так что либо вы завершите BehaviorSubject:

val subject = BehaviorSubject.createDefault(emptyList<Int>())
subject.onNext(Arrays.asList(1, 2, 3))
subject.flatMapIterable { list: List<Int> -> list }
       .subscribeBy(
                onNext = { l("on next", it) },
                onComplete = { l("on complete") }
       )

subject.onComplete() // <-----------------------------------------------------

или вы берете не более одного предмета

val subject = BehaviorSubject.createDefault(emptyList<Int>())
subject.onNext(Arrays.asList(1, 2, 3))
subject.take(1) // <----------------------------------------------------------
       .flatMapIterable { list: List<Int> -> list }
       .subscribeBy(
                onNext = { l("on next", it) },
                onComplete = { l("on complete") }
       )