Do I need to write a custom operator for it? Or we already have some kind of combination to get that...

# rx

[thanh]

Do I need to write a custom operator for it? Or we already have some kind of combination to get that function?

[darkmoon_uk]

I'm interested in a 'Reactive Coroutines' / Flow-based equivalent to the Rx scan function as well. It could be implemented by having a var share some limited scope with a Flow.map function of course; but like you I'd prefer something idiomatic if it exists.

[darkmoon_uk]

@thanh Close to your topic, in RxJava it's very common to work with `Observable`s that provide an initial 'last known' value upon subscription, followed by 'hot' updates. This somewhat merges the concept of hot and cold observables and is (not so intuituively) called a ConflatedBroadcastChannel in Kotlin coroutines.

See: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-conflated-broadcast-channel/

[tseisel]

RxJava Observable has scanWith(Callable<R>, BiFunction<R, T, R>), which is scan with a deferred first value.

[thanh]

@tseisel My scan function is very similar with scanWith, the different is I want to produce the first value from the first emitted item with initialValueProvider.

[ilyagulya]

@thanh Hi!
You simulate such behavior like this:

observable
  .startWith(Observable.fromCallable({YOUR_CALLABLE))
  .scan({YOUR_ACCUMULATOR})

[darkmoon_uk]

Pretty sure @thanh is looking for a coroutines equivalent answer using `Flow`/`Channel`/`ConflatedBroadcastChannel` operators.
People keep responding with RxJava answers... correct if I'm wrong @thanh

[tseisel]

Sorry @thanh, I didn't noticed the difference !
I wrote something like this :

fun <T : Any, R : Any> Observable<T>.scanFirst(firstValueMapper: (T) -> R, accumulator: (R, T) -> R): Observable<R> {
    val first = this.firstElement().map(firstValueMapper)
    val rest = this.skip(1)
    return rest.scanWith({ first.blockingGet() }, accumulator)
}

Not sure it performs well regarding the Rx spec, but that's way simpler than writing a custom operator.

[tseisel]

@darkmoon_uk, does the <T, R> Flow<T>.scan(initial: R, operation: (accumulator: R, value: T) -> R): Flow<R> function available in coroutines-core match your need ? I think it performs the same as Rx's scan.

[darkmoon_uk]

Sure does! Thanks 😄

[ilyagulya]

@darkmoon_uk Then it’s really strange to ask questions about coroutines in #rx channel.

[thanh]

@darkmoon_uk My question is about RxJava 😉

[thanh]

thanks @ilyagulya and @tseisel. Based on your answers I came up with this:

fun <T, R> Observable<T>.scan(initialValueProvider: (T) -> R, accumulator: (R, R) -> R): Observable<R> {
    return this.map(initialValueProvider).scan(accumulator)
}

It is identical with my original function but it works in my case.

[thanh]

Here is the better answer: https://stackoverflow.com/a/57688533/422579