Удаление переменной мутации из обработки реактивного потока

Question

Следующее работает, но я думаю, что оно может быть лучше, т. Е. Более идиоматическим rx / функциональным / менее изменчивым.

var total = 0
var published = 0

service
    .getThings()
    .doOnNext { total++ }
    .flatMap { shouldPublish(it) }  // Returns empty() or just(it)
    .doOnNext { published++ }
    .subscribe { publishToSqs(it) }

println("Published $publish out of $total")

Есть идеи?

Answer 1

Я обычно использую оператор '. Scan ()' при '. Merge ()' d потоке событий.Таким образом, всегда есть действительный снимок последнего состояния.

Подробнее об этом здесь - Реактивный конвейер агрегатного состояния