Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Flowing Things, not so strange in the MVI world

ragdroid
August 27, 2019

Flowing Things, not so strange in the MVI world

Migrating an Rx MVI pattern completely to coroutines using Flow. In the MVI world, there was a missing piece from the coroutines framework and due to this, it always felt strange to completely adopt coroutines. Recently, with the introduction of Co-routines Flow library, things are not so strange anymore. In this talk we will have a look at Coroutines Flow library, its need and how it compares with the reactive world. We will then learn to migrate an Rx MVI pattern to use coroutines Flow.

ragdroid

August 27, 2019
Tweet

More Decks by ragdroid

Other Decks in Programming

Transcript

  1. What? • Chapter One : Coroutines and Flow • Chapter

    Two : Channels and Flows • Chapter Three : MVI • Chapter Four : Rx to Flow • Summary
  2. Coroutines • Coroutine Scope ★ ViewModel Scope ★ Lifecycle Scope

    • Coroutine Builders ★ launch { } - fire and forget ★ async { } - await() result ★ within a scope, Structured Concurrency
  3. Flow • Cold asynchronous stream that sequentially emits values •

    Utilizes coroutines and channels • Experimental Recently Stable APIs in 1.3.0
  4. Flow Operators • Intermediate : ★ map, filter, take, zip,

    etc. • Terminal : ★ collect, single, reduce, etc.
  5. Flow Operators fun main() = runBlocking { val flow =

    flow { for (i in 1..10) { delay(500L) emit(i) } }.filter { it % 2 == 0 } flow.collect { println(it) } } 2 4 6 8 10
  6. Context Preservation val flowA = flowOf(1, 2, 3) .map {

    it + 1 } .flowOn(ctxA) val filtered = flowA .filter { it == 3 } withContext(Dispatchers.Main) { val result = filtered.single() myUi.text = result } https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-flow/ // Will be executed in ctxA // Changes the upstream context: flowOf and map // Pure operator without a context yet // All non-encapsulated operators will be executed in Main: filter and single
  7. fun main() = runBlocking { val job = launch {

    produce { println(“We are hot") send(1) } } job.invokeOnCompletion { println("Completed") } delay(1000L) println("Done waiting") } Channels Done waiting Delay We are hot
  8. Channels fun main() = runBlocking { val job = launch

    { val hotChannel = produce { println(“We are hot") send(1) } channel.consumeEach { println(it) } } job.invokeOnCompletion { println("Completed") } delay(1000L) println("Done waiting") } 1 Completed Done waiting We are hot
  9. Flows fun main() = runBlocking { val job = launch

    { val coldFlow = flow { println(“We are cold”) emit(1) } } job.invokeOnCompletion { println("Completed") } delay(1000L) println("Done waiting") } Completed Done waiting
  10. Flows fun main() = runBlocking { val job = launch

    { val coldFlow = flow { println("We are cold") emit(1) } coldFlow.collect { println(it) } } job.invokeOnCompletion { println("Completed") } delay(1000L) println("Done waiting") } We are cold 1 Completed Done waiting
  11. Channel Gotchas • Channels are a pathway to a different

    dimension (Upside Down) • If we they are open, another dimension exists • Existence of another dimension could be dangerous • It could use up some heavy resources
  12. View ViewModel Data Intent newState Action Result previousState Reducer State

    View Intention State 4 3 5 1 2 MVI on top of MVVM
  13. State View Intention 1 3 2 4 5 Action Events

    Actions to Results Reduce Results to new State Plugging-in the pieces State stream
  14. State View Intention 1 Action Events 3 2 4 5

    Actions to Results Reduce Results to new State State stream Plugging-in the pieces
  15. State View Intention 1 Action Events 3 2 4 5

    Actions to Results Reduce Results to new State State stream Plugging-in the pieces
  16. State View Intention 1 Action Events 3 2 4 5

    Actions to Results Reduce Results to new State State stream Plugging-in the pieces same
  17. Actions to Results sealed class MainResult { object Loading object

    LoadingError data class LoadingComplete(val characters: List<CharacterMarvel>) object PullToRefreshing: MainResult() object PullToRefreshError data class PullToRefreshComplete(val characters: List<CharacterMarvel>) sealed class DescriptionResult(val characterId: Long) { data class DescriptionLoading(private val id: Long) data class DescriptionError(private val id: Long) data class DescriptionLoadComplete(private val id: Long, val description: String) } }
  18. Actions to Results - LoadData fun loadingResult(loadDataActionStream: Flowable<MainAction.LoadData>) : Flowable<MainResult>

    { return loadDataActionStream .observeOn(Schedulers.io()) .flatMap { mainRepository.fetchCharactersSingle().toFlowable() .map { list -> MainResult.LoadingComplete(list) } .startWith(MainResult.Loading) .onErrorReturn { error -> navigate(MainNavigation.Snackbar(error.message)) MainResult.LoadingError } } } ViewModel
  19. fun loadingResult(actions: Flowable<...>) : Flowable<MainResult> = actions .observeOn(Schedulers.io()) .flatMap {

    mainRepository.fetchCharactersFlowable() .map { MainResult.LoadingComplete } .startWith(MainResult.Loading) .onErrorReturn { error -> navigate(MainNavigation.Snackbar) MainResult.LoadingError } } fun loadingResult(actions: Flowable<MainAction.LoadData>) : Flowable<MainResult> = actions .observeOn(Schedulers.io()) .flatMap { mainRepository.fetchCharactersSingle().toFlowable() .map { states -> MainResult.LoadingComplete(states) } .startWith(MainResult.Loading) .onErrorReturn { error -> navigate(MainNavigation.Snackbar(error.message) MainResult.LoadingError } } Actions to Results - Load Data RxViewModel
  20. fun loadingResult(actions: Flowable<...>) : Flowable<MainResult> = actions .observeOn(Schedulers.io()) .flatMap {

    mainRepository.fetchCharactersFlowable() .map { MainResult.LoadingComplete } .startWith(MainResult.Loading) .onErrorReturn { error -> navigate(MainNavigation.Snackbar) MainResult.LoadingError } } RxViewModel fun loadingResult(actionsFlow: Flow<MainAction.LoadData>) : Flow<MainResult> = actionsFlow.flatMapMerge { flow { emit(MainResult.Loading) val characters = mainRepository.fetchCharacters() emit(MainResult.LoadingComplete(characters)) }.catch { navigate(MainNavigation.Snackbar(it.message) emit(MainResult.LoadingError) } } FlowViewModel Actions to Results - Load Data
  21. fun loadingResult(actions: Flowable<...>) : Flowable<MainResult> = actions .observeOn(Schedulers.io()) .flatMap {

    mainRepository.fetchCharactersFlowable() .map { MainResult.LoadingComplete } .startWith(MainResult.Loading) .onErrorReturn { error -> navigate(MainNavigation.Snackbar) MainResult.LoadingError } } RxViewModel FlowViewModel fun loadingResult(actionsFlow: Flow<MainAction.LoadData>) : Flow<MainResult> = actionsFlow.flatMapMerge { flow { emit(mainRepository.fetchCharacters()) } .map { MainResult.LoadingComplete(characters) } .onStart { emit(MainResult.Loading) } .catch { navigate(MainNavigation.Snackbar(it.message) emit(MainResult.LoadingError) } } Actions to Results - Load Data
  22. Actions to Results fun actionsToResultTransformer(actions: Flowable<MainAction>): Flowable<MainResult> = actions.publish {

    shared -> Flowable.merge(loadingResult(shared.ofType(MainAction.LoadData::class.java)), loadDescriptionResult(shared.ofType(MainAction.LoadDescription::class.java)), pullToRefreshResult(shared.ofType(MainAction.PullToRefresh::class.java))) } RxViewModel
  23. Actions to Results fun Flow<MainAction>.actionsToResultTransformer(): Flow<MainResult> = publish { loadingResult(it.ofType(MainAction.LoadData::class.java))

    .merge(loadDescriptionResult(it.ofType(MainAction.LoadDescription::class.java)), pullToRefreshResult(it.ofType(MainAction.PullToRefresh::class.java))) } Flow
  24. Merge Operator Actions to Results fun Flow<MainAction>.actionsToResultTransformer(): Flow<MainResult> = publish

    { loadingResult(it.ofType(MainAction.LoadData::class.java)) .merge(loadDescriptionResult(it.ofType(MainAction.LoadDescription::class.java)), pullToRefreshResult(it.ofType(MainAction.PullToRefresh::class.java))) } Flow
  25. Merge Operator • No merge operator at the time •

    Creating operators with coroutines is comparatively easy • Let's create one
  26. fun <T> Flow<T>.merge(other: Flow<T>): Flow<T> = channelFlow { launch {

    collect { send(it) } } launch { other.collect { send(it) } } } FlowExtensions.kt // collect from this coroutine and send it // collect and send from this coroutine, too, concurrently Merge Operator
  27. Actions to Results fun Flow<MainAction>.actionsToResultTransformer(): Flow<MainResult> = flatMapMerge { flow

    { when(it) { is MainAction.PullToRefresh -> { ... } is MainAction.LoadData -> { try { emit(MainResult.Loading) val characters = mainRepository.fetchCharacters() emit(MainResult.LoadingComplete(characters)) } catch (exception: Exception) { navigate(MainNavigation.Snackbar(exception.message)) emit(MainResult.LoadingError(exception)) } } is MainAction.LoadDescription -> { … } } } }
  28. View ViewModel Data Intent newState Action Result previousState Reducer State

    View Intention (Co-routines) State Repository Network
  29. View ViewModel Data Intent newState Action Result previousState Reducer State

    View Intention (Co-routines) State Repository Network Room 2.2.0- alpha
  30. State View Intention 3 2 4 5 Actions to Results

    Reduce Results to new State State stream Plugging-in the pieces 1 Action Events same
  31. State View Intention 3 2 4 5 Actions to Results

    Reduce Results to new State State stream Plugging-in the pieces 1 Action Events same
  32. Reduce Results to new State data class MainViewState(val characters: List<CharacterItemState>,

    val emptyState: EmptyState, val loadingState: LoadingState): MviState sealed class EmptyState { object None: EmptyState() object NoData: EmptyState() object NoInternet: EmptyState() } sealed class LoadingState { object None: LoadingState() object Loading: LoadingState() object PullToRefreshing: LoadingState() } } State
  33. Reduce Results to new State fun reduce(result: MainResult) { return

    when (result) { is MainResult.Loading -> copy(loadingState = LoadingState.Loading) is MainResult.LoadingError -> copy(loadingState = LoadingState.None, emptyState = EmptyState.NoData) is MainResult.LoadingComplete -> { val characterStates = reduceCharactersList(null, result.characters, resources) copy(characterStates, loadingState = LoadingState.None, emptyState = EmptyState.None) } ... } RxReducer
  34. Reduce Results to new State FlowReducer fun reduce(result: MainResult) {

    return when (result) { is MainResult.Loading -> copy(loadingState = LoadingState.Loading) is MainResult.LoadingError -> copy(loadingState = LoadingState.None, emptyState = EmptyState.NoData) is MainResult.LoadingComplete -> { val characterStates = reduceCharactersList(null, result.characters, resources) copy(characterStates, loadingState = LoadingState.None, emptyState = EmptyState.None) } ... }
  35. State View Intention 1 Action Events 3 2 4 5

    Actions to Results Reduce Results to new State State stream Plugging-in the pieces same
  36. State View Intention 3 2 4 5 Actions to Results

    Reduce Results to new State Plugging-in the pieces State stream 1 Action Events same
  37. State View Intention 3 2 4 5 Actions to Results

    Reduce Results to new State Plugging-in the pieces State stream 1 Action Events same same
  38. State State stream data class MainViewState(val characters: List<CharacterItemState>, val emptyState:

    EmptyState, val loadingState: LoadingState): MviState sealed class EmptyState { object None: EmptyState() object NoData: EmptyState() object NoInternet: EmptyState() } sealed class LoadingState { object None: LoadingState() object Loading: LoadingState() object PullToRefreshing: LoadingState() } }
  39. State LiveData fun stateLiveData(): LiveData<MainViewState> = stateLiveData private val stateLiveData

    = MutableLiveData<MainViewState>() … stateLiveData.postValue(it) ViewModel
  40. State LiveData fun stateLiveData(): LiveData<MainViewState> = stateLiveData private val stateLiveData

    = MutableLiveData<MainViewState>() … stateLiveData.postValue(it) ViewModel viewModel.stateLiveData() .observe(viewLifecycleOwner, Observer { render(it) }) View
  41. State Flow private val stateChannel = ConflatedBroadcastChannel<MainViewState>() val stateFlow =

    stateChannel.asFlow() … stateChannel.offer(it) ViewModel View viewModel.stateFlow.collect { render(it) }
  42. ConflatedBroadcastChannel State Flow private val stateChannel = ConflatedBroadcastChannel<MainViewState>() val stateFlow

    = stateChannel.asFlow() … stateChannel.offer(it) ViewModel lifecycleScope.launch { viewModel.stateFlow.collect { render(it) } } View 2.2.0- alpha02
  43. ConflatedBroadcastChannel • BroadcastChannel - multiple receivers • Conflate - combine

    • Recent value is emitted • Like RxJava BehaviorSubject
  44. Render override fun render(state: MainViewState) { binding.refreshing = state.loadingState ==

    MainViewState.LoadingState.PullToRefreshing binding.loading = state.loadingState == MainViewState.LoadingState.Loading val characterModelList = state.characters.map { CharacterItem(it, this) } adapter.replaceItems(characterModelList, true) } View
  45. State View Intention 4 5 Plugging-in the pieces State stream

    3 2 Actions to Results Reduce Results to new State 1 Action Events same same
  46. State View Intention 1 Action Events 3 2 4 5

    Actions to Results Reduce Results to new State Plugging-in the pieces State stream 3 2 Actions to Results Reduce Results to new State 1 Action Events same same
  47. State View Intention 1 Action Events 3 2 4 5

    Actions to Results Reduce Results to new State Plugging-in the pieces State stream 3 2 Actions to Results Reduce Results to new State 1 Action Events same same same
  48. FlowViewModel Plugging-in the pieces var broadcastChannel = ConflatedBroadcastChannel<MainAction>() fun onAction(action:

    MainAction) = broadcastChannel.offer(action) var actionsFlow = broadcastChannel.asFlow()
  49. fun processActions() { actionsProcessor .compose(actionToResultTransformer) .scan(initialState) { state, result ->

    reduce(state, result)} .subscribe({ stateLiveData.postValue(it) }, Timber::e) .bindToLifecycle() } fun processActions() { actionsProcessor .compose(actionToResultTransformer) .scan(initialState) { state, result: Result -> reduce(state, result)} .subscribe({ stateLiveData.postValue(it) }, Timber::e) .bindToLifecycle() } Plugging-in the pieces RxViewModel
  50. Plugging-in the pieces fun processActions() { actionsProcessor .compose(actionToResultTransformer) .scan(initialState) {

    state, result -> reduce(state, result)} .subscribe({ stateLiveData.postValue(it) }, Timber::e) .bindToLifecycle() } fun processActions() { viewModelScope.launch { actionsFlow .broadcastIn(viewModelScope).asFlow() .actionToResultTransformer() .scan(initialState) { state, result -> reduce(state, result) } .collect { stateLiveData.postValue(it) } } } RxViewModel FlowViewModel
  51. State View Intention 5 Plugging-in the pieces 1 Action Events

    3 2 4 Actions to Results Reduce Results to new State State stream 3 2 Actions to Results Reduce Results to new State 1 Action Events same same same
  52. State View Intention 5 Plugging-in the pieces 1 Action Events

    3 2 4 Actions to Results Reduce Results to new State State stream 3 2 Actions to Results Reduce Results to new State 1 Action Events same same same
  53. What Did we learn? Rx Coroutines Single / Completable suspend

    function Flowable/Observable Flow BehaviorSubject ConflatedBroadcastChannel (DataFlow proposal) Schedulers Dispatchers Disposables Scopes
  54. What Did we learn? Rx Flow just(“Hello”) flowOf(“Hello”) flatmap() flatmapMerge()

    subscribe() collect() publish() broadcastIn() startWith() onStart() onErrorReturn() catch() map() map() scan() scan() ... ...
  55. Why Flow? • First Party support • Android libraries have

    Flow support now like Room • Reducing complexity of business logic • Hopefully share more business logic
  56. Non coroutine APIs to Coroutine APIs • Yes • suspendCoroutine

    { } • Continuation.resumeWith( ) Continuation.resumeWithException()
  57. What Next? • kotlin-flow-extensions: https://github.com/akarnokd/kotlin-flow-extensions • CoroutineBinding: CoroutineBinding: https://github.com/satoshun/CoroutineBinding •

    Flowing in the Deep: https://www.droidcon.com/media-detail?video=352670453 • DataFlow: https://github.com/Kotlin/kotlinx.coroutines/pull/1354/files • Flow Guide: https://github.com/Kotlin/kotlinx.coroutines/blob/1.3.0/docs/flow.md • Flow vs Channel: https://github.com/manuelvicnt/MathCoroutinesFlow • Demo Project: https://github.com/ragdroid/klayground/tree/kotlin-flow
  58. References & Acknowledgements • https://github.com/brewin/mvi-coroutines • Manuel Vivo @manuelvicnt •

    Hannes Dorfmann @sockeqwe • Android Team @Over • Ritesh Gupta @_riteshhh • https://medium.com/@elizarov - Roman Elizarov • https://medium.com/@objcode - Sean McQuillan • https://www.youtube.com/watch?v=PXBXcHQeDLE - MVI for Android, Benoît Quenaudon