I’m trying to subscribe to

rxjava

events and consume them as

flow

, it seems i’m not subscribing correctly with flow or missing to correlating action to rx

subscribe

This is the

flow

usage, i’m expecting to collect each change to the data from the

flowable

i’m listening to:

runBlocking {
            getSubtitleFromProjectDataManager()
        }

private suspend fun getSubtitleFromProjectDataManager() {
        scope.launch(<http://Dispatchers.IO|Dispatchers.IO>) {

            val flow = projectDataManager.getProjectNameFlowable().asFlow()
            flow.collect {
                subtitle = it

            }
            launch(Dispatchers.Main) {
                view?.rebindToolbar()
            }
        }
    }

This is the rxjava code I’m trying to listen to. Indeed the event isn’t firing again when the data is changed, so I suspect as said before there is a subscription issue:

public Flowable<String> getProjectNameFlowable() {
        return RxJavaInterop.toV2Flowable(
                getSelectedProject()
                        .switchMap(projectEntity -> {
                                    Timber.e("fire projectEntity " + projectEntity.title());
                                    return Observable.just(projectEntity.title());
                                }
                        ));
    }

You should use triple baticks for code snippets that are not just referecing a short single line.

edited. thanks.

Also, I think decreasing the indent and making it match standards (4 spaces) would help readability for others that might want to help you.

This line won’t run until your flow completes because

flow.collect { ... }

is blocking.

launch(Dispatchers.Main) {
    view?.rebindToolbar()
}

I would also suggest that you wrap your suspend function body in

coroutineScope { ... }

to prevent leaks

Thanks @reline, any idea how I make the flow listen to changes and emit?

You can still use collect here. I believe your goal is to set the subtitle and rebind the toolbar each time you get a new project? maybe do something like

flow.collect {
    withContext(Dispatchers.Main) {
        subtitle = it
        view?.rebindToolbar()
    }
}

also take a look here if you can use experimental APIs https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flow-on.html

Yes thats what im trying to do, migrate from rx but keep the idea of subscription, will give it a shot, thanks @reline!

This:

flow.collect {
    withContext(Dispatchers.Main) {
        subtitle = it
        view?.rebindToolbar()
    }
}

switches dispatcher on each new value, which is inefficient. Switch dispatcher once for all at collect site (wrapping the whole

collect

call) if you want to do that. For flow source/emitter, there's the

flowOn

operator as said earlier.

thanks for pointing out the differences Louis. If I’m not mistaken, won’t wrapping the

collect

call inside

withContext(Main)

cause it to observe on the main thread as well? how would you go about observing on IO and then consuming events on Main (without using

flowOn

)?

Flow using another dispatcher internally (including using

flowOn

) is an implementation detail. The Flow should be safe to collect on the main thread.