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

Fueled Reactive apps with Asynchronous Flow & S...

Fueled Reactive apps with Asynchronous Flow & StateFlow to Sync with the UI

Reactive Extensions are widely used on large scale successful Android applications, and for this, the most popular library is an adaptation of these Reactive Extensions, the very well known RxJava.
What if we could use the existing Kotlin suspending functions asynchronously to return more than a single value?
How could we return multiple asynchronously backed results in Kotlin?
Kotlin Flows to the rescue!
Would Kotlin Flows replace RxJava in the long term on the Android platform?
How could we start migrating smoothly to Flow on hybrid (Java & Kotlin) languages apps?
Can we fully avoid using callbacks now? StateFlow to the rescue!

Conferences or meetups:
- Droidcon EMEA #dcEMEA #droidconEMEA (October the 9th 2020)
- Madrid Android Developer Group (MADG) meetup - Spanish language (November the 10th 2020)
- Kotlin London (KDU) meetup (December the 2nd 2020)
- Droidcon APAC #dcAPAC #droidconAPAC (December the 14th 2020)

Raul Hernandez Lopez

October 09, 2020
Tweet

More Decks by Raul Hernandez Lopez

Other Decks in Programming

Transcript

  1. 1. Use case 2. Migration strategy 3. Basics 4. Implementation

    5. Lessons learned 6. Next steps @raulhernandezl AGENDA
  2. Use Case Repository Network data source DB data source Business

    Layer @raulhernandezl requests results results
  3. Presenter Use Case Repository Network data source DB data source

    Model View Presenter (MVP) + Clean Architecture @raulhernandezl requests executes results results results
  4. Presenter Use Case Repository View / Callbacks Network data source

    DB data source Model View Presenter (MVP) + Clean Architecture @raulhernandezl View Delegate View Listener starts injects requests executes types results results results
  5. Presenter Use Case Repository View / Callbacks Network data source

    DB data source requests View Delegate View Listener JAVA + RXJAVA @raulhernandezl executes starts injects types results results results
  6. Presenter Use Case Repository View / Callbacks Network data source

    DB data source requests executes View / Callbacks View Delegate View Listener MIGRATION of DATASOURCES: KOTLIN + RXJAVA @raulhernandezl starts injects types results results results
  7. Presenter Use Case Repository Network data source DB data source

    requests executes View / Callbacks View Delegate View Listener MIGRATION of DATASOURCES: KOTLIN + RXJAVA + COROUTINES @raulhernandezl starts injects types results results results
  8. Presenter Use Case Repository Network data source DB data source

    requests executes View / Callbacks View Delegate View Listener MIGRATION of REPOSITORY: KOTLIN + RXJAVA @raulhernandezl starts injects types results results results
  9. Presenter Use Case Repository View / Callbacks Network data source

    DB data source requests executes View / Callbacks View Delegate View Listener MIGRATION of REPOSITORY: KOTLIN + RXJAVA + FLOW @raulhernandezl starts injects types results results results
  10. Presenter Use Case Repository View / Callbacks Network data source

    DB data source requests executes View / Callbacks View Delegate View Listener MIGRATION of REPOSITORY: KOTLIN + FLOW / SUSPEND @raulhernandezl starts injects types results results results Flow
  11. Presenter Use Case Repository Network data source DB data source

    requests executes View / Callbacks View Delegate View Listener MIGRATION of USE CASE: KOTLIN + FLOW / SUSPEND @raulhernandezl starts injects types results results results Flow Flow
  12. Presenter Use Case Repository Network data source DB data source

    requests executes Flow Flow View / Callbacks View Delegate View Listener @raulhernandezl CAN WE CHANGE VIEWDELEGATE’S RXJAVA? starts injects types results results results
  13. Presenter Use Case Repository View / Callbacks Network data source

    DB data source View Delegate starts executes requests View Listener Channels as Flows injects VIEWDELEGATE: CHANNELS @raulhernandezl Flow Flow types results results results
  14. Presenter Use Case Repository View / Callbacks Network data source

    DB data source View Delegate starts executes requests View Listener Channels as Flows injects CAN WE REMOVE CALLBACKS to COMMUNICATE? @raulhernandezl Flow Flow types results results results
  15. Presenter Use Case Repository View Network data source DB data

    source View Delegate starts executes requests View Listener Channels as Flows injects CALLBACKS MIGRATION: STATEFLOW @raulhernandezl Flow Flow results types StateFlow StateFlow
  16. Presenter Use Case Repository View Network data source DB data

    source View Delegate starts executes requests View Listener Channels as Flows injects CALLBACKS MIGRATION: STATEFLOW HANDLER @raulhernandezl Flow Flow StateFlow StateFlow Handler results types StateFlow initializes processes
  17. ONE SHOT OPERATIONS @raulhernandezl SINGLE Single<T> SUSPEND FUNCTION OBJECT suspend

    () -> T MAYBE Maybe<T> SUSPEND FUNCTION NULLABLE suspend () -> T?
  18. ONE SHOT OPERATIONS @raulhernandezl SINGLE Single<T> SUSPEND FUNCTION OBJECT suspend

    () -> T MAYBE Maybe<T> SUSPEND FUNCTION NULLABLE suspend () -> T? COMPLETABLE Completable SUSPEND FUNCTION UNIT suspend () -> Unit
  19. Gradle dependencies @raulhernandezl build.gradle ... // Kotlin Coroutines implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:$kotlin_coroutines_version"

    implementation "org.jetbrains.kotlinx:kotlinx-coroutines-android:$kotlin_coroutines_version" // Kotlin Standard Library implementation "org.jetbrains.kotlin:kotlin-stdlib:$kotlin_stdlib_version" kotlin_coroutines_version = '1.4.0' kotlin_stdlib_version = '1.4.0'
  20. Presenter Use Case Repository Network data source DB data source

    requests executes View / Callbacks View Delegate View Listener SUSPEND to SINGLE @raulhernandezl starts injects results results types results
  21. @Singleton class NetworkDataSourceImpl @Inject constructor( private val twitterApi: TwitterApi, private

    val connectionHandler: ConnectionHandler, private val requestsIOHandler: RequestsIOHandler ) : NetworkDataSource NetworkDataSourceImpl constructor dependencies @raulhernandezl
  22. @NotNull public Single<List<TweetApiModel>> search( @NotNull String token, @NotNull String query

    ) { return twitterApi.search( requestsIOHandler.getTokenFormatted(token),query) ... } NetworkDataSource w/ Single (Java) @raulhernandezl
  23. @NotNull public Single<List<TweetApiModel>> search( @NotNull String token, @NotNull String query

    ) { return twitterApi.search( requestsIOHandler.getTokenFormatted(token),query) .filter(requestsIOHandler::searchHasNoErrorResponse) .map(requestsIOHandler::getTweets) .flatMapSingle(Single::just)); } NetworkDataSource w/ Single (Java) @raulhernandezl
  24. override suspend fun search(token: String, query: String) : Either<Throwable, List<TweetApiModel>>

    { val response = twitterApi.search( requestsIOHandler.getTokenFormatted(token), query) return if (requestsIOHandler.searchIsSuccessful(response)) { val tweets = requestsIOHandler.getTweets(response) Either.right(tweets) } else { Either.left(EmptyResponseException(response.msg())) } } NetworkDataSource w/ suspend @raulhernandezl
  25. override fun search(token: String, query: String) : Single<Either<Throwable, List<TweetApiModel>>> {

    return rxSingle { val response = twitterApi.search( requestsIOHandler.getTokenFormatted(token), query) if (requestsIOHandler.searchIsSuccessful(response)) { val tweets = requestsIOHandler.getTweets(response) Either.right(tweets) } else { Either.left(EmptyResponseException()) } } } NetworkDataSource with... rxSingle @raulhernandezl kotlinx-coroutines-rx2
  26. @Query("SELECT * FROM ${Tweet.TABLE_NAME} WHERE tweet_id IN(:tweetIds) ORDER BY created_at

    DESC") fun retrieveAllTweetsRxForTweetsIds(tweetIds: List<String>): Single<List<Tweet>> TweetDao: Database datasource (DAO) @raulhernandezl
  27. Presenter Use Case Repository View / Callbacks Network data source

    DB data source requests executes View / Callbacks View Delegate View Listener REPOSITORY: KOTLIN + FLOW + RXJAVA @raulhernandezl starts injects types results results types results
  28. @Singleton class TweetsRepositoryImpl @Inject constructor( private val networkDataSource: NetworkDataSource, private

    val tweetsDataSource: TweetDao, private val mapperTweets: TweetsNetworkToDBMapperList, private val tokenDataSource: TokenDao, private val queryDataSource: QueryDao, private val tweetQueryJoinDataSource: TweetQueryJoinDao, private val mapperToken: TokenNetworkToDBMapper, private val taskThreading: TaskThreading ) : TweetsRepository { TweetsRepository constructor dependencies @raulhernandezl
  29. fun getSearchTweets(token: String, query: String): Observable<List<Tweet>> { return networkDataSource.search(token, query)

    .subscribeOn(taskThreading.io()) .observeOn(taskThreading.computation()) ... } TweetsRepository: Kotlin + RxJava threading
  30. fun getSearchTweets(token: String, query: String): Observable<List<Tweet>> { return networkDataSource.search(token, query)

    .subscribeOn(taskThreading.io()) .observeOn(taskThreading.computation()) .map { either -> mapperTweets.map(either) } ... } TweetsRepository: Kotlin + RxJava map
  31. fun getSearchTweets(token: String, query: String): Observable<List<Tweet>> { return networkDataSource.search(token, query)

    .subscribeOn(taskThreading.io()) .observeOn(taskThreading.computation()) .map { either -> mapperTweets.map(either) } .doOnNext { tweetsToAdd -> tweetQueryJoinDataSource.deleteTweets(query) tweetsDataSource.insertTweets(tweetsToAdd) insertQuery(query) tweetQueryJoinDataSource.insertQueryAndTweetsId( convert(tweetsToAdd, query)) ... } TweetsRepository: Kotlin + RxJava doOnNext
  32. fun getSearchTweets(token: String, query: String): Observable<List<Tweet>> { return networkDataSource.search(token, query)

    .subscribeOn(taskThreading.io()) .observeOn(taskThreading.computation()) .map { either -> mapperTweets.map(either) } .doOnNext { tweetsToAdd -> tweetQueryJoinDataSource.deleteTweets(query) tweetsDataSource.insertTweets(tweetsToAdd) insertQuery(query) tweetQueryJoinDataSource.insertQueryAndTweetsId( convert(tweetsToAdd, query)) }.flatMap { tweetQueryJoinDataSource.retrieveAllTweetsIdRxAQuery(query) .flatMap { tweetIds -> tweetsDataSource .retrieveAllTweetsRxForTweetsIds(tweetIds) } } } TweetsRepository: Kotlin + RxJava flatMap
  33. fun getSearchTweets(token: String, query: String): Flow<List<Tweet>> { return networkDataSource .search(token,

    query) .map { either -> mapperTweets.map(either) } ... } TweetsRepository w/ Flow @raulhernandezl
  34. fun getSearchTweets(token: String, query: String): Flow<List<Tweet>> { return networkDataSource .search(token,

    query) .map { either -> mapperTweets.map(either) } .onEach { tweetQueryJoinDataSource.deleteTweets(query) tweetsDataSource.insertTweets(it) insertQuery(query) tweetQueryJoinDataSource.insertQueryAndTweetsId( convert(it,query)) } ... } TweetsRepository w/ onEach @raulhernandezl
  35. fun getSearchTweets(token: String, query: String): Flow<List<Tweet>> { return networkDataSource .search(token,

    query) .map { either -> mapperTweets.map(either) } .onEach { tweetQueryJoinDataSource.deleteTweets(query) tweetsDataSource.insertTweets(it) insertQuery(query) tweetQueryJoinDataSource.insertQueryAndTweetsId( convert(it,query)) }.flatMapConcat { val tweetIds = tweetQueryJoinDataSource.retrieveAllTweetsIdAQuery(query) return@flatMapConcat tweetsDataSource .retrieveAllTweetsForTweetsIdsFlow(tweetIds) } } TweetsRepository w/ flatMapConcat @raulhernandezl
  36. @Query("SELECT * FROM ${Tweet.TABLE_NAME} WHERE tweet_id IN(:tweetIds) ORDER BY created_at

    DESC") fun retrieveAllTweetsForTweetsIdsFlow(tweetIds: List<String>): Flow<List<Tweet>> Database datasource (DAO) w/ Flow @raulhernandezl
  37. fun getSearchTweets(token: String, query: String): Flow<List<Tweet>> { return networkDataSource .search(token,

    query) .map { either -> mapperTweets.map(either) } .onEach { tweetQueryJoinDataSource.deleteTweets(query) tweetsDataSource.insertTweets(it) insertQuery(query) tweetQueryJoinDataSource.insertQueryAndTweetsId( convert(it,query)) }.flatMapConcat { val tweetIds = tweetQueryJoinDataSource.retrieveAllTweetsIdAQuery(query) return@flatMapConcat tweetsDataSource .retrieveAllTweetsForTweetsIdsFlow(tweetIds) }.flowOn(taskThreading.ioDispatcher()) } TweetsRepository w/ flowOn & IO dispatcher @raulhernandezl
  38. class TaskThreadingImpl @Inject constructor() : TaskThreading { private val computationScheduler

    = Schedulers.computation() private val ioScheduler = Schedulers.io() private val mainScheduler = AndroidSchedulers.mainThread() override fun ui(): Scheduler = mainScheduler override fun io(): Scheduler = ioScheduler override fun computation(): Scheduler = computationScheduler } TaskThreading with RxJava Schedulers @raulhernandezl
  39. class TaskThreadingImpl @Inject constructor() : TaskThreading { private val computationCoroutineDispatcher:

    CoroutineDispatcher = Default private val ioCoroutineDispatcher: CoroutineDispatcher = IO private val mainCoroutineDispatcher: CoroutineDispatcher = Main ... override fun uiDispatcher(): CoroutineDispatcher = mainCoroutineDispatcher override fun ioDispatcher(): CoroutineDispatcher = ioCoroutineDispatcher override fun computationDispatcher(): CoroutineDispatcher = computationCoroutineDispatcher } TaskThreading with Coroutines Dispatchers @raulhernandezl
  40. @Singleton class TaskThreadingImpl @Inject constructor() : TaskThreading { private val

    computationScheduler = Schedulers.computation() private val ioScheduler = Schedulers.io() private val mainScheduler = AndroidSchedulers.mainThread() ... } Can we re-use same RxJava schedulers / dispatchers? @raulhernandezl
  41. @Singleton class TaskThreadingImpl @Inject constructor() : TaskThreading { private val

    computationScheduler = Schedulers.computation() private val ioScheduler = Schedulers.io() private val mainScheduler = AndroidSchedulers.mainThread() ... } Yes, we can re-use existing RxJava schedulers YES! @raulhernandezl
  42. @Singleton class TaskThreadingImpl @Inject constructor() : TaskThreading { private val

    computationScheduler = Schedulers.computation() private val ioScheduler = Schedulers.io() private val mainScheduler = AndroidSchedulers.mainThread() override fun uiDispatcher(): CoroutineDispatcher = mainScheduler.asCoroutineDispatcher() override fun ioDispatcher(): CoroutineDispatcher = ioScheduler.asCoroutineDispatcher() override fun computationDispatcher(): CoroutineDispatcher = computationScheduler.asCoroutineDispatcher() TaskThreading: reuse Schedulers transformed into CoroutineDispatchers kotlinx-coroutines-rx2 @raulhernandezl
  43. fun getSearchTweets(token: String, query: String): Observable<List<Tweet>> { return networkDataSource .search(token,

    query) .map { either -> mapperTweets.map(either) } .onEach { tweetQueryJoinDataSource.deleteTweets(query) tweetsDataSource.insertTweets(it) insertQuery(query) tweetQueryJoinDataSource.insertQueryAndTweetsId( convert(it,query)) }.flatMapConcat { val tweetIds = tweetQueryJoinDataSource.retrieveAllTweetsIdAQuery(query) return@flatMapConcat tweetsDataSource .retrieveAllTweetsForTweetsIdsFlow(tweetIds) }.flowOn(taskThreading.ioDispatcher()).asObservable() } TweetsRepository w/ Flow to Observable @raulhernandezl kotlinx-coroutines-rx2
  44. Presenter Use Case Repository View / Callbacks Network data source

    DB data source requests executes View / Callbacks View Delegate View Listener REPOSITORY: KOTLIN + FLOW / SUSPEND @raulhernandezl starts injects results results types results
  45. suspend fun getSearchTweets(query: String): List<Tweet> { val accessToken = tokenDataSource.getAccessToken()

    ?: throw BadTokenException("No token was found") val token = accessToken.token ... } Case 1) TweetsRepository: suspend @raulhernandezl
  46. suspend fun getSearchTweets(query: String): List<Tweet> { val accessToken = tokenDataSource.getAccessToken()

    ?: throw BadTokenException("No token was found") val token = accessToken.token val eitherResult = networkDataSource .search(token, query) val tweets = mapperTweets.map(eitherResult) ... } Case 1) TweetsRepository: suspend @raulhernandezl
  47. suspend fun getSearchTweets(query: String): List<Tweet> { val accessToken = tokenDataSource.getAccessToken()

    ?: throw BadTokenException("No token was found") val token = accessToken.token val eitherResult = networkDataSource .search(token, query) val tweets = mapperTweets.map(eitherResult) tweetQueryJoinDataSource.deleteTweets(query) tweetsDataSource.insertTweets(tweets) insertQuery(query) tweetQueryJoinDataSource.insertQueryAndTweetsId( tweetQueryJoinDataSource.convert(tweets, query)) ... } Case 1) TweetsRepository: suspend @raulhernandezl
  48. suspend fun getSearchTweets(query: String): List<Tweet> { val accessToken = tokenDataSource.getAccessToken()

    ?: throw BadTokenException("No token was found") val token = accessToken.token val eitherResult = networkDataSource .search(token, query) val tweets = mapperTweets.map(eitherResult) tweetQueryJoinDataSource.deleteTweets(query) tweetsDataSource.insertTweets(tweets) insertQuery(query) tweetQueryJoinDataSource.insertQueryAndTweetsId( tweetQueryJoinDataSource.convert(tweets, query)) val tweetIds = tweetQueryJoinDataSource.retrieveAllTweetsIdAQuery(query) return tweetsDataSource.retrieveAllTweetsForTweetsIdsFlow(tweetIds) } @raulhernandezl Case 1) TweetsRepository: Can Flow be returned into a suspend function? ???
  49. suspend fun getSearchTweets(query: String): List<Tweet> { val accessToken = tokenDataSource.getAccessToken()

    ?: throw BadTokenException("No token was found") val token = accessToken.token val eitherResult = networkDataSource .search(token, query) val tweets = mapperTweets.map(eitherResult) tweetQueryJoinDataSource.deleteTweets(query) tweetsDataSource.insertTweets(tweets) insertQuery(query) tweetQueryJoinDataSource.insertQueryAndTweetsId( tweetQueryJoinDataSource.convert(tweets, query)) val tweetIds = tweetQueryJoinDataSource.retrieveAllTweetsIdAQuery(query) return tweetsDataSource.retrieveAllTweetsForTweetsIdsFlow(tweetIds) } Case 1) TweetsRepository: Flow cannot be returned into suspend function! @raulhernandezl NO!
  50. suspend fun getSearchTweets(query: String): List<Tweet> { val accessToken = tokenDataSource.getAccessToken()

    ?: throw BadTokenException("No token was found") val token = accessToken.token val eitherResult = networkDataSource .search(token, query) val tweets = mapperTweets.map(eitherResult) tweetQueryJoinDataSource.deleteTweets(query) tweetsDataSource.insertTweets(tweets) insertQuery(query) tweetQueryJoinDataSource.insertQueryAndTweetsId( tweetQueryJoinDataSource.convert(tweets, query)) val tweetIds = tweetQueryJoinDataSource.retrieveAllTweetsIdAQuery(query) return tweetsDataSource.retrieveAllTweetsForTweetsIds(tweetIds) } Case 1) TweetsRepository: a suspend function is needed @raulhernandezl YES!
  51. @Query("SELECT * FROM ${Tweet.TABLE_NAME} WHERE tweet_id IN(:tweetIds) ORDER BY created_at

    DESC") suspend fun retrieveAllTweetsForTweetsIds(tweetIds: List<String>): List<Tweet> Database datasource (DAO) with suspend @raulhernandezl
  52. fun getSearchTweets(query: String): Flow<List<Tweet>> = flow { ... val tweetIds

    = tweetQueryJoinDataSource.retrieveAllTweetsIdAQuery(query) emitAll(tweetsDataSource.retrieveAllTweetsForTweetsIdsFlow(tweetIds)) }.flowOn(taskThreading.ioDispatcher()) Case 2) TweetsRepository: flow builder & emitAll values from DB (Flow) @raulhernandezl YES!
  53. fun getSearchTweets(query: String): Flow<List<Tweet>> = flow { ... val tweetIds

    = tweetQueryJoinDataSource.retrieveAllTweetsIdAQuery(query) emit(tweetsDataSource.retrieveAllTweetsForTweetsIds(tweetIds)) }.flowOn(taskThreading.ioDispatcher()) Case 2) TweetsRepository: flow builder & emit a suspend fun returning a List @raulhernandezl YES!
  54. fun getSearchTweets(query: String): Flow<List<Tweet>> = flow { ... // retrieve

    old values from DB emit(tweetsDataSource.retrieveAllTweetsForTweetsIds(tweetIds)) // get fresh values from network & saved them into DB val eitherResult = networkDataSource.search(token, query) ... // saved network into DB & emit fresh values from DB val tweetIds = tweetQueryJoinDataSource.retrieveAllTweetsIdAQuery(query) emit(tweetsDataSource.retrieveAllTweetsForTweetsIds(tweetIds)) }.flowOn(taskThreading.ioDispatcher()) Case 2) TweetsRepository: flow builder & emit more than once @raulhernandezl YES!
  55. Presenter Use Case Repository Network data source DB data source

    requests executes View / Callbacks View Delegate View Listener RxSearch USE CASE: KOTLIN + FLOW / SUSPEND @raulhernandezl starts injects results results types results
  56. @RetainedScope class SearchTweetUseCase @Inject constructor( private val tweetsRepository: TweetsRepository, private

    val taskThreading: TaskThreading ) : UseCase<SearchCallback> { SearchTweetUseCase: constructor dependencies @raulhernandezl
  57. UseCase contract w/ callback interface UseCase<T> { fun execute(param: String,

    callbackInput: T?) fun cancel() } @raulhernandezl
  58. @Override void execute(@NotNull String query, @Nullable SearchCallback callback) { this.callback

    = callback; if(callback != null) { callback.onShowLoader(); } } SearchTweetUseCase: RxJava + Java @raulhernandezl
  59. @Override void execute(@NotNull String query, @Nullable SearchCallback callback) { this.callback

    = callback; if(callback != null) { callback.onShowLoader(); } repository.searchTweet(query) .subscribe( callback::onSuccess,// action for each stream callback::onError); // error handling } SearchTweetUseCase: RxJava + Java @raulhernandezl
  60. @Override void execute(@NotNull String query, @Nullable SearchCallback callback) { this.callback

    = callback; if(callback != null) { callback.onShowLoader(); } disposable = repository.searchTweet(query) .subscribeOn(taskThreading.computation()) // subscribed to .observeOn(taskThreading.ui()) // action performed at .map( // some computation ) .subscribe( callback::onSuccess,// action for each stream callback::onError); // error handling } SearchTweetUseCase: Rx threading & map downstream @raulhernandezl
  61. override fun execute(query: String, callbackInput: SearchCallback?) { callback = callbackInput

    callback?.onShowLoader() ... } SearchTweetUseCase: Kotlin @raulhernandezl
  62. override fun execute(query: String, callbackInput: SearchCallback?) { callback = callbackInput

    callback?.onShowLoader() scope.launch { ... } } SearchTweetUseCase: Kotlin Flow w/ scope.launch @raulhernandezl private val scope = CoroutineScope(taskThreading.uiDispatcher() + SupervisorJob())
  63. override fun execute(query: String, callbackInput: SearchCallback?) { callback = callbackInput

    callback?.onShowLoader() scope.launch { repository.searchTweet(query) .map { // some computation } .flowOn(taskThreading.computationDispatcher()) ... } } SearchTweetUseCase: Kotlin Flow map in another thread w/ flowOn upstream @raulhernandezl private val scope = CoroutineScope(taskThreading.uiDispatcher() + SupervisorJob())
  64. override fun execute(query: String, callbackInput: SearchCallback?) { callback = callbackInput

    callback?.onShowLoader() scope.launch { repository.searchTweet(query) .map { // some computation } .flowOn(taskThreading.computationDispatcher()) .catch { callback::onError }.collect { tweets ->// UI actions for each stream callback.onSuccess(tweets) } } } SearchTweetUseCase: Flow + Kotlin using collect @raulhernandezl private val scope = CoroutineScope(taskThreading.uiDispatcher() + SupervisorJob())
  65. override fun execute(query: String, callbackInput: SearchCallback?) { callback = callbackInput

    callback?.onShowLoader() repository.searchTweet(query) .map { // some computation } .flowOn(taskThreading.computationDispatcher()) .onEach { tweets -> // UI actions for each stream callback.onSuccess(tweets) }.catch { callback::onError }.launchIn(scope) } SearchTweetUseCase: Flow + Kotlin using launchIn @raulhernandezl private val scope = CoroutineScope(taskThreading.uiDispatcher() + SupervisorJob())
  66. override fun cancel() { ... } private val scope =

    CoroutineScope(taskThreading.uiDispatcher() + SupervisorJob()) SearchTweetUseCase: How to cancel with Structured concurrency? @raulhernandezl
  67. override fun cancel() { callback = null scope.cancel() } private

    val scope = CoroutineScope(taskThreading.uiDispatcher() + SupervisorJob()) SearchTweetUseCase: cancellation with Structured concurrency @raulhernandezl
  68. @Override public void cancel() { if (disposable != null &&

    !disposable.isDisposed()) { disposable.dispose(); disposable = null; } callback = null; } SearchTweetUseCase: observables disposal .subscribeOn(taskThreading.computation()) .observeOn(taskThreading.ui()) @raulhernandezl
  69. fun getSearchTweets(query: String): Flow<List<Tweet>> = flow { ... }.flowOn(taskThreading.ioDispatcher()).asObserv.. Reminder:

    TweetsRepository doesn’t need to return Observable anymore, just Flow @raulhernandezl
  70. Presenter Use Case Repository View / Callbacks Network data source

    DB data source View Delegate starts executes requests View Listener Flows Flows Flows Channels as Flows injects VIEW DELEGATE with CHANNELS @raulhernandezl starts injects results results types results
  71. @ActivityScope class SearchViewDelegate constructor( private val presenter: SearchTweetPresenter ) {

    private val publishSubject: PublishSubject<String> = PublishSubject.create() ... } SearchViewDelegate: RxJava + Kotlin with PublishSubject @raulhernandezl
  72. @ActivityScope class SearchViewDelegate constructor( private val presenter: SearchTweetPresenter ) {

    private val publishSubject: PublishSubject<String> = PublishSubject.create() fun prepareViewDelegateListener(val view: SearchView) { val listener = SearchViewListener(publishSubject) view.setOnQueryTextListener(listener) } } SearchViewDelegate: RxJava + Kotlin with PublishSubject @raulhernandezl
  73. @ActivityScope class SearchViewDelegate constructor( private val presenter: SearchTweetPresenter ) {

    fun prepareViewDelegateListener(val view: SearchView): Flow<String>= ... } SearchViewDelegate: Flows with channels as flows @raulhernandezl
  74. @ActivityScope class SearchViewDelegate constructor( private val presenter: SearchTweetPresenter ) {

    @ExperimentalCoroutinesApi fun prepareViewDelegateListener(val view: SearchView): Flow<String> = (channelFlow<String> { // define listener val listener = SearchViewListener(channel) ... }).flowOn(taskThreading.ioDispatcher()) } @raulhernandezl SearchViewDelegate: Flows with channelFlow builder
  75. @ExperimentalCoroutinesApi fun prepareViewDelegateListener(val view: SearchView): Flow<String> = (channelFlow<String> { val

    listener = SearchViewListener(channel) view.setOnQueryTextListener(listener) awaitClose { view.setOnQueryTextListener(null) } }).flowOn(taskThreading.ioDispatcher()) SearchViewDelegate: Flows setting/closing the query listener @raulhernandezl
  76. @ActivityScope class SearchViewDelegate constructor( private val presenter: SearchTweetPresenter ) {

    fun observeSubject(): Disposable { ... } fun cancel() { view.setOnQueryTextListener(null) if (disposable != null && !disposable.isDisposed()) { disposable.dispose() disposable = null } presenter.cancel() } } SearchViewDelegate: RxJava needs to manually dispose the view and disposable @raulhernandezl
  77. class SearchViewListener constructor( private val publishSubject: PublishSubject<String> ): SearchView.OnQueryTextListener {

    override fun onQueryTextChange(query: String): Boolean { ... return true } override fun onQueryTextSubmit(query: String) = false } SearchViewListener w/ injected PublishSubject @raulhernandezl
  78. class SearchViewListener constructor( private val publishSubject: PublishSubject<String> ): SearchView.OnQueryTextListener {

    override fun onQueryTextChange(query: String): Boolean { publishSubject.onNext(query) return true } override fun onQueryTextSubmit(query: String) = false } SearchViewListener: RxJava with publishSubject @raulhernandezl
  79. class SearchViewListener constructor( private val queryChannel: SendChannel<String> ): SearchView.OnQueryTextListener {

    override fun onQueryTextChange(query: String): Boolean { ... return true } override fun onQueryTextSubmit(query: String) = false } SearchViewListener: Channels with SendChannel @raulhernandezl
  80. class SearchViewListener constructor( private val queryChannel: SendChannel<String> ): SearchView.OnQueryTextListener {

    override fun onQueryTextChange(query: String): Boolean { queryChannel.offer(query) return true } override fun onQueryTextSubmit(query: String) = false } SearchViewListener: Channels with SendChannel @raulhernandezl
  81. @ActivityScope class SearchViewDelegate constructor( private val presenter: SearchTweetPresenter ) {

    fun observeSubject(): Disposable = publishSubject .observeOn(taskThreading.ui()) .subscribeOn(taskThreading.computation()) ... } SearchViewDelegate w/ RxJava subject @raulhernandezl
  82. @ActivityScope class SearchViewDelegate constructor( private val presenter: SearchTweetPresenter ) {

    fun observeSubject(): Disposable = publishSubject .observeOn(taskThreading.ui()) .subscribeOn(taskThreading.computation()) .debounce(600, TimeUnit.MILLISECONDS) .distinctUntilChanged() .filter { query -> filterQuery(query) } .subscribe ( ... ) } SearchViewDelegate w/ RxJava Subject downstream @raulhernandezl
  83. @ActivityScope class SearchViewDelegate constructor( private val presenter: SearchTweetPresenter ) {

    fun observeSubject(): Disposable = publishSubject .observeOn(taskThreading.ui()) .subscribeOn(taskThreading.computation()) .debounce(600, TimeUnit.MILLISECONDS) .distinctUntilChanged() .filter { query -> filterQuery(query) } .subscribe ( presenter::searchTweet, presenter::showProblemHappened ) } SearchViewDelegate w/ RxJava starts Presenter @raulhernandezl
  84. @FlowPreview fun observeChannelAsFlow() { declareViewDelegate() .debounce(600) .distinctUntilChanged() .filter { query

    -> filterQuery(query) } .flowOn(taskThreading.computationDispatcher()) ... } SearchViewDelegate: w/ Flow flowOn upstream @raulhernandezl
  85. @FlowPreview fun observeChannelAsFlow() { declareViewDelegate() .debounce(600) .distinctUntilChanged() .filter { query

    -> filterQuery(query) } .flowOn(taskThreading.computationDispatcher()) .onEach { query -> presenter.searchTweet(query) } .catch { presenter::showProblemHappened } .launchIn(scope) } fun cancel() { scope.cancel() presenter.cancel() } SearchViewDelegate lifecycle handling and stream @raulhernandezl private val scope = CoroutineScope(taskThreading.uiDispatcher() + SupervisorJob())
  86. class TaskThreadingImpl @Inject constructor() : TaskThreading { private val computationScheduler

    = Schedulers.computation() private val ioScheduler = Schedulers.io() private val mainScheduler = AndroidSchedulers.mainThread() private val computationCoroutineContext: CoroutineDispatcher = Default private val ioCoroutineContext: CoroutineDispatcher = IO private val mainCoroutineContext: CoroutineDispatcher = Main Use native Coroutine Dispatchers @raulhernandezl
  87. class TaskThreadingImpl @Inject constructor() : TaskThreading { private val computationCoroutineContext:

    CoroutineDispatcher = Default private val ioCoroutineContext: CoroutineDispatcher = IO private val mainCoroutineContext: CoroutineDispatcher = Main override fun ui(): CoroutineContext = mainCoroutineContext override fun io(): CoroutineContext = ioCoroutineContext override fun computation(): CoroutineContext = computationCoroutineContext } Expose new dispatchers @raulhernandezl
  88. Presenter Use Case Repository View Network data source DB data

    source View Delegate starts executes requests View Listener Flows Channels as Flows injects @raulhernandezl Flows Flows results types StateFlow StateFlow STATEFLOW to get rid of CALLBACKS in USE CASE
  89. @RetainedScope class SearchTweetUseCase @Inject constructor( private val tweetsRepository: TweetsRepository, private

    val taskThreading: TaskThreading ) : UseCase<SearchCallback> { ) : UseCaseFlow { SearchTweetUseCase recap: constructor dependencies @raulhernandezl
  90. SearchTweetUseCase update: remove callback interface UseCaseStateFlow { fun execute(param: String,

    callbackInput: SearchCallback?) fun cancel() fun getStateFlow(): StateFlow<TweetsUIState?> } @raulhernandezl
  91. SearchTweetUseCase update: remove callback interface UseCaseStateFlow { fun execute(param: String,

    callbackInput: SearchCallback?) fun cancel() fun getStateFlow(): StateFlow<TweetsUIState?> } @raulhernandezl StateFlow can be passed across Java & Kotlin files
  92. @RetainedScope class SearchTweetUseCase @Inject constructor( private val tweetsRepository: TweetsRepository, private

    val taskThreading: TaskThreading ) : UseCaseFlow { private val tweetsUIStateFlow = MutableStateFlow<TweetsUIState?>(null) override fun getStateFlow(): StateFlow<TweetsUIState?> = tweetsUIStateFlow SearchTweetUseCase w/ MutableStateFlow @raulhernandezl
  93. @RetainedScope class SearchTweetUseCase @Inject constructor( private val tweetsRepository: TweetsRepository, private

    val taskThreading: TaskThreading ) : UseCaseFlow { private val tweetsUIStateFlow = MutableStateFlow<TweetsUIState?>(null) override fun getStateFlow(): StateFlow<TweetsUIState?> = tweetsUIStateFlow SearchTweetUseCase w/ MutableStateFlow distinctUntilChanged by default @raulhernandezl StateFlow uses distinctUntilChanged by default
  94. TweetsUIState w/ Results state @raulhernandezl /** * UI States defined

    for StateFlow in the workflow */ sealed class TweetsUIState { data class ListResultsUIState(val tweets: List<Tweet>): TweetsUIState() }
  95. TweetsUIState w/ Error state @raulhernandezl /** * UI States defined

    for StateFlow in the workflow */ sealed class TweetsUIState { data class ListResultsUIState(val tweets: List<Tweet>): TweetsUIState() data class ErrorUIState(val msg: String): TweetsUIState() }
  96. TweetsUIState w/ Empty state @raulhernandezl /** * UI States defined

    for StateFlow in the workflow */ sealed class TweetsUIState { data class ListResultsUIState(val tweets: List<Tweet>): TweetsUIState() data class ErrorUIState(val msg: String): TweetsUIState() object EmptyUIState: TweetsUIState() }
  97. TweetsUIState w/ Loading state @raulhernandezl /** * UI States defined

    for StateFlow in the workflow */ sealed class TweetsUIState { data class ListResultsUIState(val tweets: List<Tweet>): TweetsUIState() data class ErrorUIState(val msg: String): TweetsUIState() object EmptyUIState: TweetsUIState() object LoadingUIState: TweetsUIState() }
  98. override fun execute(query: String) { tweetsStateFlow.value = TweetsUIState.LoadingUIState repository.searchTweet(query) .onEach

    { tweets -> val stateFlow = if (tweets.isEmpty()) { TweetsUIState.EmptyUIState } tweetsStateFlow.value = stateFlow } .catch { e -> ... }.launchIn(scope) } SearchTweetUseCase: Empty state propagation @raulhernandezl
  99. override fun execute(query: String) { tweetsStateFlow.value = TweetsUIState.LoadingUIState repository.searchTweet(query) .onEach

    { tweets -> val stateFlow = if (tweets.isEmpty()) { TweetsUIState.EmptyUIState } else { TweetsUIState.ListResultsUIState(tweets) } tweetsStateFlow.value = stateFlow } .catch { e -> ... }.launchIn(scope) } SearchTweetUseCase: List results state propagation @raulhernandezl
  100. override fun execute(query: String) { tweetsStateFlow.value = TweetsUIState.LoadingUIState repository.searchTweet(query) .onEach

    { tweets -> val stateFlow = if (tweets.isEmpty()) { TweetsUIState.EmptyUIState } else { TweetsUIState.ListResultsUIState(tweets) } tweetsStateFlow.value = stateFlow } .catch { e -> tweetsStateFlow.value = TweetsUIState.ErrorUIState(e.msg)) }.launchIn(scope) } SearchTweetUseCase: Error state propagation @raulhernandezl
  101. Presenter Use Case Repository View Network data source DB data

    source View Delegate starts executes requests View Listener Flows Channels as Flows injects STATEHANDLER (KOTLIN) to COLLECT STATE @raulhernandezl Flows Flows StateFlow StateFlow Handler results types StateFlow initializes processes
  102. SearchTweetPresenter (Java): constructor @raulhernandezl @ActivityScope public class SearchTweetPresenter { @NotNull

    private final SearchTweetUseCase tweetSearchUseCase; @Inject public SearchTweetPresenter( @NotNull SearchTweetUseCase tweetSearchUseCase ) { this.tweetSearchUseCase = tweetSearchUseCase; } public void cancel() { tweetSearchUseCase.cancel(); }
  103. SearchTweetPresenter (Java) responsibilities @raulhernandezl @ActivityScope public class SearchTweetPresenter { ...

    public void searchTweets(@NotNull final String query) { if (callback == null && view != null) { callback = new SearchCallbackImpl(view); } tweetSearchUseCase.execute(query, callback); } @NotNull public StateFlow<TweetsUIState> getStateFlow() { return tweetSearchUseCase.getStateFlow(); } StateFlow can be passed across Java & Kotlin files
  104. SearchViewDelegate gets StateFlow from Presenter @raulhernandezl @ActivityScope class SearchViewDelegate @Inject

    constructor( private val presenter: SearchTweetPresenter, private val taskThreading: TaskThreading ) { private fun getStateFlow(): StateFlow<TweetsUIState> = presenter.stateFlow
  105. SearchViewDelegate initialises UI w/ StateFlow @raulhernandezl @ActivityScope class SearchViewDelegate @Inject

    constructor( private val presenter: SearchTweetPresenter, private val taskThreading: TaskThreading ) { private fun getStateFlow(): StateFlow<TweetsUIState> = presenter.stateFlow fun initialiseProcessingQuery( searchView: SearchView, uiResults: TweetsListUI? ) { uiResults?.apply { initStateFlow(getStateFlow()) } ... }
  106. TweetsListUI / View (Java) delegates to StateFlowHandler @raulhernandezl public class

    TweetsListUI extends ViewUI { @Inject SearchViewDelegate viewDelegate; @Inject StateFlowHandler stateFlowHandler; public void initStateFlow( @NotNull StateFlow<? extends TweetsUIState> stateFlow ) { stateFlowHandler.init(stateFlow, this); stateFlowHandler.processStateFlowCollection(); } StateFlow can be passed across Java & Kotlin files
  107. StateFlowHandler: constructor @raulhernandezl @ActivityScope class StateFlowHandler @Inject constructor( taskThreading: TaskThreading

    ) { private var tweetsListUI: TweetsListUI? = null private lateinit var stateFlow: StateFlow<TweetsUIState?> fun init( val stateFlow: StateFlow<TweetsUIState?>, val tweetsListUI: TweetsListUI ) { stateFlow = stateFlow tweetsListUI = tweetsListUI } ... }
  108. StateFlowHandler filters StateFlow prior collection private lateinit var stateFlow: StateFlow<TweetsUIState?>

    fun processStateCollection() { stateFlow .filter { stateFlow -> stateFlow != null } ... } @raulhernandezl
  109. StateFlowHandler collects StateFlow w/ onEach private lateinit var stateFlow: StateFlow<TweetsUIState?>

    fun processStateCollection() { stateFlow .filter { stateFlow -> stateFlow != null } .onEach { stateFlow -> tweetsListUI?.handleStates(stateFlow) }... } @raulhernandezl
  110. StateFlowHandler w/ StateFlow collection requires a scope private val scope

    = CoroutineScope(taskThreading.ui() + SupervisorJob()) fun processStateCollection() { stateFlow .filter { stateFlow -> stateFlow != null } .onEach { stateFlow -> tweetsListUI?.handleStates(stateFlow) }.launchIn(scope) } @raulhernandezl StateFlow needs a Scope to be collected
  111. TweetsListUI handles States @raulhernandezl fun TweetsListUI.handleStates(stateFlow: TweetsUIState?) { when (stateFlow)

    { is TweetsUIState.LoadingUIState -> { showLoading() } is TweetsUIState.ListResultsUIState -> { showResults(stateFlow) } is TweetsUIState.EmptyUIState -> { showEmptyState() } is TweetsUIState.ErrorUIState -> { showError(stateFlow) } else -> { showEmptyState() } } }
  112. StateFlowHandler cancels StateFlow collection @raulhernandezl @ActivityScope class StateFlowHandler @Inject constructor(

    taskThreading: TaskThreading ) { private val scope = CoroutineScope(taskThreading.ui() + SupervisorJob()) private var tweetsListUI: TweetsListUI? = null fun cancel() { scope.cancel() tweetsListUI = null }
  113. TweetsListUI / View lifecycle @raulhernandezl public class TweetsListUI extends ViewUI

    { @Inject SearchViewDelegate viewDelegate; @Inject StateFlowHandler stateFlowHandler; @Override public void onDestroy() { stateFlowHandler.cancel(); viewDelegate.cancel(); super.onDestroy(); } }
  114. Pros & Cons @raulhernandezl RxJava Declarative Coroutines + Flow Imperative

    + Declarative RxJava Manual lifecycle handling *by default Coroutines + Flow Structured Concurrency RxJava Java + Kotlin hybrid projects * Coroutines + Flow Perfect for Kotlin & friends
  115. References @raulhernandezl Fueled Reactive apps with Asynchronous Flow 1. Use

    case & Migration Strategy 2. Asynchronous communication: Streams & Basics 3. Data layer Implementation 4. Use Case layer Implementation 5. View Delegate Implementation 6. Lessons learned & Next steps 7. Synchronous communication with the UI using StateFlow