Exploring Coroutines

Presentation about architecting a unidirectional data flow Android application using Kotlin's coroutines, with UI drive by state machines.

Presented at

- KotlinSP Meetup #10 (August/2019 - São Paulo, Brazil)

1. Ubiratan Soares August / 2019 EXPLORING COROUTINES
2. How can we architect with Coroutines ?
5. DataSource Use case ViewModel collaborates with
6. DataSource Use case ViewModel External World User Interface Observable<T> Observable<T> Observable<T> data flow format
7. DataSource Use case ViewModel External World User Interface Observable<T> Observable<T> LiveData<T> data flow format
8. DataSource Use case ViewModel External World User Interface suspend fun (T) suspend fun (T) LiveData<T> data flow format
9. DataSource Use case ViewModel External World User Interface suspend fun (T) suspend fun (T) LiveData<T> data flow format
10. Unidirectional Data Flow
11. Activity( ) : ViewProtocol doSomething ( ) Presenter View Protocol fun handle(d : Data) fun showLoading( ) fun showError( ) fun doSomething( ) override fun handle(d : Data) override fun showLoading( ) override fun showError( ) doAnotherThing ( ) fun doAnotherThing( ) method call
12. Activity( ) doSomething ( ) ViewModel fun doSomething( ) : LiveData<Data> fun handle(d : Data) fun showLoading( ) fun showError( ) doAnotherThing ( ) fun doAnotherThing( ) : LiveData<Data> method call LiveData Observer
13. Activity( ) doSomething ( ) ViewModel fun doSomething( ) : LiveData<Data> fun handle(d : Data) fun showLoading( ) fun showError( ) doAnotherThing ( ) fun doAnotherThing( ) : LiveData<Data> method call fun justOneMoreThing( ) : LiveData<Data> justOneMoreThing ( ) LiveData Observer
14. Activity( ) doSomething ( ) ViewModel doAnotherThing ( ) method call justOneMoreThing ( ) LiveData Observer LiveData Observer LiveData Observer LiveData LiveData LiveData
15. Activity( ) doSomething ( ) ViewModel doAnotherThing ( ) method call justOneMoreThing ( ) LiveData Observer LiveData Observer LiveData Observer LiveData LiveData LiveData
17. DataSource Use case ViewModel External World User Interface suspend fun (T) suspend fun (T) suspend fun (T) data flow format
18. DataSource Use case ViewModel External World User Interface suspend fun (T) suspend fun (T) suspend fun (T) data flow format How can we push multiple states to the View level ???
25. DataSource Use case ViewModel External World User Interface suspend fun (T) suspend fun (T) Flow<T>
27. Activity( ) handle(someInteraction) ViewModel method call FlowConsumer StateMachine fun handle(i : UserInteraction) bind( ) fun bind( ) : Flow<ViewState> fun handle(d : Data) fun showLoading( ) fun showError( ) data flow format Flow<ViewState>
28. Started bind( )
29. Started Loading interaction (suspend wrapper) bind( )
30. Started Loading Success interaction (suspend wrapper) bind( ) suspend fun resumed
31. Started Loading Error interaction (suspend wrapper) bind( ) suspend fun resumed
32. Started Loading Success Error interaction (suspend wrapper) bind( ) suspend fun resumed suspend fun resumed
33. Started Loading Success Error interaction (suspend wrapper) interaction (suspend wrapper) interaction (suspend wrapper) bind( ) suspend fun resumed suspend fun resumed
34. sealed class StateTransition<out T> { interface Parameters class Unparametrized<T> internal constructor( val task: suspend () !" T ) : StateTransition<T>() class Parametrized<T> internal constructor( val task: suspend (Parameters) !" T, val parameters: Parameters ) : StateTransition<T>() }
35. sealed class StateTransition<out T> { interface Parameters class Unparametrized<T> internal constructor( val task: suspend () !" T ) : StateTransition<T>() class Parametrized<T> internal constructor( val task: suspend (Parameters) !" T, val parameters: Parameters ) : StateTransition<T>() }
36. class StateMachine<T>( private val container: StateContainer<T>, private val executor: TaskExecutor ) { fun states() = container.observableStates() fun consume(execution: StateTransition<T>) = executor.execute { wrapWithStates(execution) } // Continue on next slide }
37. class StateMachine<T>(#$%) { private suspend fun wrapWithStates(execution: StateTransition<T>) { val first = executionStarted() moveTo(first) val next = executeWith(execution) moveTo(next) } private suspend fun executeWith(transition: StateTransition<T>) = try { val execution = when (transition) { is Unparametrized !" transition.task.invoke() is Parametrized !" with(transition) { this.task.invoke(parameters) } } Success(execution) } catch (error: Throwable) { Failed(error) } }
38. class StateMachine<T>(#$%) { private suspend fun wrapWithStates(execution: StateTransition<T>) { val first = executionStarted() moveTo(first) val next = executeWith(execution) moveTo(next) } private suspend fun executeWith(transition: StateTransition<T>) = try { val execution = when (transition) { is Unparametrized !" transition.task.invoke() is Parametrized !" with(transition) { this.task.invoke(parameters) } } Success(execution) } catch (error: Throwable) { Failed(error) } }
39. class StateMachine<T>(#$%) { private suspend fun wrapWithStates(execution: StateTransition<T>) { val first = executionStarted() moveTo(first) val next = executeWith(execution) moveTo(next) } private suspend fun executeWith(transition: StateTransition<T>) = try { val execution = when (transition) { is Unparametrized !" transition.task.invoke() is Parametrized !" with(transition) { this.task.invoke(parameters) } } Success(execution) } catch (error: Throwable) { Failed(error) } }
40. A complete data flow
41. interface ChuckNorrisDotIO { @GET("/jokes/categories") suspend fun categories(): RawCategories @GET("/jokes/search") suspend fun search(@Query("query") query: String): RawSearch companion object { const val API_URL = "https:&'api.chucknorris.io" } }
42. class FactsInfrastructure( private val rest: ChuckNorrisDotIO ) : RemoteFactsService { override suspend fun availableCategories() = managedExecution { rest.categories().asRelatedCategories() } override suspend fun fetchFacts(searchTerm: String) = managedExecution { rest.search(searchTerm).asChuckNorrisFacts() } }
43. class FetchFacts( private val factsService: RemoteFactsService ) { suspend fun search(term: String) = when { term.isEmpty() !" throw EmptyTerm else !" { val facts = factsService.fetchFacts(term) if (facts.isEmpty()) throw NoResultsFound else facts } } }
44. class FactsViewModel( private val factsFetcher: FetchFacts, private val queryManager: ManageSearchQuery, private val machine: StateMachine<FactsPresentation> ) { fun bind() = machine.states() fun handle(interaction: UserInteraction) = interpret(interaction) .let { transition !" machine.consume(transition) } private fun interpret(interaction: UserInteraction) = when (interaction) { OpenedScreen, RequestedFreshContent !" StateTransition(()showFacts) else !" throw UnsupportedUserInteraction }
45. private fun interpret(interaction: UserInteraction) = when (interaction) { OpenedScreen, RequestedFreshContent !" StateTransition(()showFacts) else !" throw UnsupportedUserInteraction } private suspend fun showFacts() = queryManager.actualQuery().let { query !" factsFetcher .search(query) .map { FactDisplayRow(it) } .let { rows !" FactsPresentation(query, rows) } } }
46. private fun interpret(interaction: UserInteraction) = when (interaction) { OpenedScreen, RequestedFreshContent !" StateTransition(::showFacts) else !" throw UnsupportedUserInteraction } private suspend fun showFacts() = queryManager.actualQuery().let { query !" factsFetcher .search(query) .map { FactDisplayRow(it) } .let { rows !" FactsPresentation(query, rows) } } }
47. class FactsActivity : AppCompatActivity(), KodeinAware { override fun onCreate(savedInstanceState: Bundle?) { super.onCreate(savedInstanceState) setContentView(R.layout.activity_main) setup() } private fun setup() { setSupportActionBar(factsToolbar) factsRecyclerView.layoutManager = LinearLayoutManager(this) factsSwipeToRefresh.setOnRefreshListener { refresh() } lifecycleScope.launch { viewModel.bind().collect { renderState(it) } } }
48. private fun setup() { setSupportActionBar(factsToolbar) factsRecyclerView.layoutManager = LinearLayoutManager(this) factsSwipeToRefresh.setOnRefreshListener { refresh() } lifecycleScope.launch { viewModel.bind().collect { renderState(it) } } } private fun renderState(state: ViewState<FactsPresentation>) = when (state) { is Failed !" handleError(state.reason) is Success !" showFacts(state.value) is Loading.FromEmpty !" startExecution() is Loading.FromPrevious !" showFacts(state.previous) is FirstLaunch !" loadFacts() } }
49. private fun renderState(state: ViewState<FactsPresentation>) = when (state) { is Failed !" handleError(state.reason) is Success !" showFacts(state.value) is Loading.FromEmpty !" startExecution() is Loading.FromPrevious !" showFacts(state.previous) is FirstLaunch !" loadFacts() } } private fun loadFacts() { viewModel.handle(OpenedScreen) } }
50. Solving Lifecycles with Structured Concurrency
51. State Machine TaskExecutor StateContainer collaborates with class StateMachine<T>( private val container: StateContainer<T>, private val executor: TaskExecutor ) { // Implementation }
52. State Machine TaskExecutor StateContainer collaborates with The State Container registers the current state processed by the State Machine. The implementation is CoroutineScope-aware and uses a ConflatedBroadcastChannel as communication engine to propagate state to consumers. We can build a StateContainer instance with a CoroutineScope differente than the consumer’s scope
53. State Machine TaskExecutor StateContainer collaborates with The task executor defines a model of execution for the state transition : either Concurrent or Synchronous
54. interface TaskExecutor { fun execute(block: suspend TaskExecutor.() !" Unit): Job class Concurrent( private val scope: CoroutineScope, private val dispatcher: CoroutineDispatcher ) : TaskExecutor { override fun execute(block: suspend TaskExecutor.() !" Unit) = scope.launch(dispatcher) { block.invoke(this@Concurrent) } } class Synchronous(private val scope: CoroutineScope) : TaskExecutor { override fun execute(block: suspend TaskExecutor.() !" Unit) = runBlocking { scope.launch { block.invoke(this@Synchronous) } } } }
55. interface TaskExecutor { fun execute(block: suspend TaskExecutor.() !" Unit): Job class Concurrent( private val scope: CoroutineScope, private val dispatcher: CoroutineDispatcher ) : TaskExecutor { override fun execute(block: suspend TaskExecutor.() !" Unit) = scope.launch(dispatcher) { block.invoke(this@Concurrent) } } class Synchronous(private val scope: CoroutineScope) : TaskExecutor { override fun execute(block: suspend TaskExecutor.() !" Unit) = runBlocking { scope.launch { block.invoke(this@Synchronous) } } } }
56. Executor Coroutine (runs the suspend function) Consumer Coroutine (consume the states) Channel
57. Executor Coroutine (runs the suspend function Scope = AAC ViewModel Consumer Coroutine (consume the states) Scope = Activity Lifecycle Channel
58. class ConfigChangesAwareStateContainer<T> : StateContainer<T>, ViewModel() { private val broadcaster by lazy { ConflatedBroadcastChannel<ViewState<T*+(ViewState.FirstLaunch) } override val emissionScope = viewModelScope override fun observableStates() = broadcaster.asFlow() override fun current(): ViewState<T> = broadcaster.value override suspend fun store(state: ViewState<T>) { broadcaster.send(state) }
59. class ConfigChangesAwareStateContainer<T> : StateContainer<T>, ViewModel() { private val broadcaster by lazy { ConflatedBroadcastChannel<ViewState<T*+(ViewState.FirstLaunch) } override val emissionScope = viewModelScope override fun observableStates() = broadcaster.asFlow() override fun current(): ViewState<T> = broadcaster.value override suspend fun store(state: ViewState<T>) { broadcaster.send(state) } Cold flow over Hot Channel
60. class ConfigChangesAwareStateContainer<T> : StateContainer<T>, ViewModel() { private val broadcaster by lazy { ConflatedBroadcastChannel<ViewState<T*+(ViewState.FirstLaunch) } override val emissionScope = viewModelScope override fun observableStates() = broadcaster.asFlow() override fun current(): ViewState<T> = broadcaster.value override suspend fun store(state: ViewState<T>) { broadcaster.send(state) }
61. companion object { operator fun <T> invoke(host: FragmentActivity): StateContainer<T> { val f = object : ViewModelProvider.Factory { override fun <Model : ViewModel> create(klass: Class<Model>) = ConfigChangesAwareStateContainer<T>() as Model } val keyClazz = ConfigChangesAwareStateContainer()class.java return ViewModelProviders.of(host, f)[keyClazz] as StateContainer<T> } } }
62. Testing like a pro
63. class FetchCategoriesTests { private val categoriesCache = mock<CategoriesCacheService>() private val remoteFacts = mock<RemoteFactsService>() private lateinit var usecase: FetchCategories private val categories = mutableListOf( RelatedCategory.Available("dev"), RelatedCategory.Available("code") ) @Before fun `before each test`() { usecase = FetchCategories(categoriesCache, remoteFacts) } @Test fun `should fetch from cache from available`() { runBlocking { `given that remote service not available`() `given that cache returns categories`() assertThat(usecase.execute()).isEqualTo(categories) } } }
64. internal class FactsInfrastructureTests { @get:Rule val rule = InfrastructureRule() private lateinit var infrastructure: FactsInfrastructure @Before fun `before each test`() { infrastructure = FactsInfrastructure(rule.api) } @Test fun `should fetch categories`() { rule.defineScenario( status = 200, response = loadFile("200_categories.json") )
65. val expected = listOf( Available("career"), Available("celebrity"), Available("dev") ) val categories = runBlocking { infrastructure.availableCategories() } assertThat(categories).isEqualTo(expected) } private fun simpleSearch() = runBlocking { infrastructure.fetchFacts("Norris") } }
66. class FactsViewModelTests { @get:Rule val helper = CoroutinesTestHelper() private val fetchFacts = mock<FetchFacts>() private val manageQuery = mock<ManageSearchQuery>() private lateinit var viewModel: FactsViewModel @Before fun `before each test`() { val stateMachine = StateMachine<FactsPresentation>( executor = TaskExecutor.Synchronous(helper.scope), container = StateContainer.Unbounded(helper.scope) ) viewModel = FactsViewModel(fetchFacts, manageQuery, stateMachine) } @Test fun `should report failure when fetching from remote`() {
67. @Test fun `should report failure when fetching from remote`() { &' Given flowTest(viewModel.bind()) { triggerEmissions { &' When whenever(fetchFacts.search(anyString())) .thenAnswer { throw UnexpectedResponse } &' And viewModel.handle(UserInteraction.OpenedScreen) } afterCollect { emissions !"
68. afterCollect { emissions !" &' Then val viewStates = listOf( FirstLaunch, Loading.FromEmpty, Failed(UnexpectedResponse) ) assertThat(emissions).isEqualTo(viewStates) } } } }
69. class FlowTest<T>(private val parentJob: Job, private val emissions: List<T>) { fun triggerEmissions(action: suspend () !" Job) { runBlocking { action().join() } } fun afterCollect(verification: (List<T>) !" Unit) { parentJob.invokeOnCompletion { verification.invoke(emissions) } } companion object { fun <T> flowTest( target: Flow<T>, scope: CoroutineScope = GlobalScope, block: FlowTest<T>.() !" Unit ) { target.test(scope, block) } } }
70. class FlowTest<T>(private val parentJob: Job, private val emissions: List<T>) { fun triggerEmissions(action: suspend () !" Job) { runBlocking { action().join() } } fun afterCollect(verification: (List<T>) !" Unit) { parentJob.invokeOnCompletion { verification.invoke(emissions) } } companion object { fun <T> flowTest( target: Flow<T>, scope: CoroutineScope = GlobalScope, block: FlowTest<T>.() !" Unit ) { target.test(scope, block) } } }
72. https://speakerdeck.com/ubiratansoares
73. UBIRATAN SOARES Computer Scientist by ICMC/USP Software Engineer, curious guy Google Developer Expert for Android / Kotlin Teacher, speaker, etc, etc
74. THANK YOU @ubiratanfsoares ubiratansoares.dev https://br.linkedin.com/in/ubiratanfsoares