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

Fueled Reactive apps with Asynchronous Flow

Fueled Reactive apps with Asynchronous Flow

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?
These slides have been used during #VirtualAndroidMakers @AndroidMakersFR

Raul Hernandez Lopez

April 20, 2020
Tweet

More Decks by Raul Hernandez Lopez

Other Decks in Programming

Transcript

  1. 1. Use case 2. Migration strategy 3. Asynchronous communication: Streams

    4. Basics 5. Implementation 6. Demo time 7. Lessons learnt 8. Next steps @raulhernandezl AGENDA
  2. Presenter Use Case Repository View / Callbacks Network data source

    DB data source Presentation Layer @raulhernandezl
  3. Presenter Use Case Repository View / Callbacks Network data source

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

    DB data source requests View Delegate View Listener JAVA + RXJAVA EVERYWHERE @raulhernandezl executes starts injects
  5. 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
  6. Presenter Use Case Repository Network data source DB data source

    requests executes View / Callbacks View Delegate View Listener MIGRATION of DATASOURCES: KOTLIN + RXJAVA + FLOW @raulhernandezl starts injects
  7. 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
  8. 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
  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 + FLOW / SUSPEND @raulhernandezl starts injects
  10. 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
  11. Presenter Use Case Repository Network data source DB data source

    requests executes Flows Flows Flows View / Callbacks View Delegate View Listener @raulhernandezl WHERE USING FLOWS? starts injects
  12. Presenter Use Case Repository Network data source DB data source

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

    DB data source View Delegate starts executes requests View Listener Flows Channels as Flows injects VIEWDELEGATE: CHANNELS @raulhernandezl Flows Flows
  14. ONE SHOT OPERATIONS @raulhernandezl SINGLE Single<T> SUSPEND FUNCTION OBJECT suspend

    () -> T MAYBE Maybe<T> SUSPEND FUNCTION NULLABLE suspend () -> T?
  15. 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
  16. Presenter Use Case Repository Network data source DB data source

    requests executes View / Callbacks View Delegate View Listener FLOWS to OBSERVABLES @raulhernandezl
  17. @Singleton class NetworkDataSourceImpl @Inject constructor( private val twitterApi: TwitterApi, private

    val connectionHandler: ConnectionHandler, private val requestsIOHandler: RequestsIOHandler, private val taskThreading: TaskThreading ) : NetworkDataSource NetworkDataSourceImpl constructor dependencies @raulhernandezl
  18. class TaskThreadingImpl @Inject constructor() : TaskThreading { private val computationScheduler

    = Schedulers.computation() private val ioScheduler = Schedulers.io() private val mainScheduler = AndroidSchedulers.mainThread() } TaskThreading and threads @raulhernandezl
  19. 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 and Schedulers @raulhernandezl
  20. public Single<List<TweetApiModel>> search(String token, String query) { return twitterApi.search( requestsIOHandler.getTokenFormatted(access.getToken()),

    query) .subscribeOn(taskThreading.io()) .observeOn(taskThreading.computation()) .filter(requestsIOHandler::searchHasNoErrorResponse) .map(requestsIOHandler::getSearchContent) .map(requestsIOHandler::getTweets) .flatMapSingle(Single::just)); } Case 1) NetworkDataSource w/ Single @raulhernandezl
  21. public Single<List<TweetApiModel>> search(String token, String query) { return twitterApi.search( requestsIOHandler.getTokenFormatted(access.getToken()),

    query) .subscribeOn(taskThreading.io()) .observeOn(taskThreading.computation()) .filter(requestsIOHandler::searchHasNoErrorResponse) .map(requestsIOHandler::getSearchContent) .map(requestsIOHandler::getTweets) .flatMapSingle(Single::just)); Case 1) NetworkDataSource w/ Single @raulhernandezl
  22. 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())) } } Case 1) NetworkDataSource with suspend @raulhernandezl
  23. @ExperimentalCoroutinesApi override fun search(token: String, query: String) : Flow<Either<Throwable, List<TweetApiModel>>>

    = flow<Either<Throwable, List<TweetApiModel>>> { ... emit(myListOfTweets) } Case 2) NetworkDataSource emit with the flow builder @raulhernandezl
  24. @ExperimentalCoroutinesApi override fun search(token: String, query: String) : Flow<Either<Throwable, List<TweetApiModel>>>

    = flow<Either<Throwable, List<TweetApiModel>>> { ... }.flowOn(taskThreading.ioDispatcher()) flowOn operator @raulhernandezl
  25. 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 Dispatchers @raulhernandezl
  26. @ExperimentalCoroutinesApi override fun search(token: String, query: String) : Flow<Either<Throwable, List<TweetApiModel>>>

    = flow<Either<Throwable, List<TweetApiModel>>> { val response = twitterApi.search( requestsIOHandler.getTokenFormatted(token), query) if (requestsIOHandler.searchIsSuccessful(response)) { val tweets = requestsIOHandler.getTweets(response) emit(Either.right(tweets)) } else { emit(Either.left(EmptyResponseException(response.msg()))) } }.flowOn(taskThreading.ioDispatcher()) Case 2) NetworkDataSource emit with Flow @raulhernandezl
  27. @ExperimentalCoroutinesApi override fun search(token: String, query: String) : Observable<Either<Throwable, List<TweetApiModel>>>

    = flow<Either<Throwable, List<TweetApiModel>>> { val response = twitterApi.search( requestsIOHandler.getTokenFormatted(token), query) if (requestsIOHandler.searchIsSuccessful(response)) { val tweets = requestsIOHandler.getTweets(response) emit(Either.right(tweets)) } else { emit(Either.left(EmptyResponseException())) } }.flowOn(taskThreading.ioDispatcher()) .asObservable() Case 2) NetworkDataSource with asObservable kotlinx-coroutines-rx2 @raulhernandezl
  28. @Query("SELECT * FROM ${Tweet.TABLE_NAME} WHERE tweet_id IN(:tweetIds) ORDER BY created_at

    DESC") fun retrieveAllTweetsForTweetsIdsRx(tweetIds: List<String>): Observable<List<Tweet>> Database datasource (DAO) @raulhernandezl
  29. 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
  30. @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
  31. 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()) } } } Case 1) NetworkDataSource with... rxSingle @raulhernandezl kotlinx-coroutines-rx2
  32. val eitherResult = networkDataSource.search(token, query).await() val tweets = mapperTweets.map(eitherResult) Case

    1) TweetsRepository waiting for Single @raulhernandezl kotlinx-coroutines-rx2
  33. val eitherResult = networkDataSource.search(token, query).await() val tweets = mapperTweets.map(eitherResult) Case

    1) TweetsRepository waiting for Single I can wait for that single value without blocking anything @raulhernandezl kotlinx-coroutines-rx2
  34. fun getSearchTweets(token: String, query: String): Observable<List<Tweet>> { return networkDataSource.search(token, query)

    .map { either -> mapperTweets.map(either) } ... } Case 2) TweetsRepository: Kotlin + RxJava
  35. fun getSearchTweets(token: String, query: String): Observable<List<Tweet>> { return networkDataSource.search(token, query)

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

    .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) } } } Case 2) TweetsRepository: Kotlin + RxJava
  37. ... fun getSearchTweets(token: String, query: String): Flow<List<Tweet>> { return networkDataSource

    .search(token, query) .map { either -> mapperTweets.map(either) } ... } Case 2) TweetsRepository: Flow @raulhernandezl
  38. ... 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)) } ... } Case 2) TweetsRepository: onEach @raulhernandezl
  39. ... 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) } } Case 2) TweetsRepository: flatMapConcat @raulhernandezl
  40. ... 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()) } Case 2) TweetsRepository: flowOn @raulhernandezl
  41. @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) with Flow @raulhernandezl
  42. ... 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() } Case 2) TweetsRepository: Flow to Observable @raulhernandezl kotlinx-coroutines-rx2
  43. 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
  44. 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
  45. 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
  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) tweetQueryJoinDataSource.deleteTweets(query) tweetsDataSource.insertTweets(tweets) insertQuery(query) tweetQueryJoinDataSource.insertQueryAndTweetsId( tweetQueryJoinDataSource.convert(tweets, query)) ... } 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)) val tweetIds = tweetQueryJoinDataSource.retrieveAllTweetsIdAQuery(query) return tweetsDataSource.retrieveAllTweetsForTweetsIdsFlow(tweetIds) } @raulhernandezl Case 1) TweetsRepository: Flow returned into suspend function?
  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) } Case 1) TweetsRepository: Flow returned into suspend function? @raulhernandezl NO!
  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.retrieveAllTweetsForTweetsIds(tweetIds) } Case 1) TweetsRepository: a suspend function is needed @raulhernandezl YES!
  50. @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
  51. ... fun getSearchTweets(query: String) = flow<List<Tweet>> { ... val tweetIds

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

    = tweetQueryJoinDataSource.retrieveAllTweetsIdAQuery(query) emit(tweetsDataSource.retrieveAllTweetsForTweetsIds(tweetIds)) }.flowOn(taskThreading.ioDispatcher()) Case 1.. 2) TweetsRepository: emit a suspend List @raulhernandezl YES!
  53. ... fun getSearchTweets(query: String) = flow<List<Tweet>> { ... // 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 1.. 2) TweetsRepository: emit more than once @raulhernandezl YES!
  54. 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
  55. @ActivityScope class SearchTweetUseCase @Inject constructor( private val tweetsRepository: TweetsRepository, private

    val taskThreading: TaskThreading ) : UseCase<SearchCallback> { SearchTweetUseCase constructor dependencies @raulhernandezl
  56. @Override void execute(final String query, @Nullable SearchCallback callback) { this.callback

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

    = callback; if(callback != null) { callback.onShowLoader(); } repository.searchTweet(query) .subscribe( this::onSuccess,// action for each stream this::onError); // error handling } SearchTweetUseCase: RxJava + Java @raulhernandezl
  58. @Override void execute(final 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 .subscribe( this::onSuccess,// action for each stream this::onError); // error handling } SearchTweetUseCase: Lifecycle & threading @raulhernandezl
  59. @ExperimentalCoroutinesApi override fun execute(query: String, callbackInput: SearchCallback?) { callback =

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

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

    callbackInput callback?.onShowLoader() scope.launch { repository.searchTweet(query) .catch { // error handling } .collect { // UI actions for each stream } } } SearchTweetUseCase: Flow + Kotlin with scope.launch @raulhernandezl private val scope = CoroutineScope(taskThreading.uiDispatcher() + SupervisorJob())
  62. @ExperimentalCoroutinesApi override fun execute(query: String, callbackInput: SearchCallback?) { callback =

    callbackInput callback?.onShowLoader() repository.searchTweet(query) .onEach { //iteration for each stream } .catch { // error handling } .launchIn(scope) } SearchTweetUseCase: Flow + Kotlin launchIn @raulhernandezl private val scope = CoroutineScope(taskThreading.uiDispatcher() + SupervisorJob())
  63. SearchTweetUseCase: onEach @ExperimentalCoroutinesApi override fun execute(query: String, callbackInput: SearchCallback?) {

    ... repository.searchTweet(query) .onEach { onSuccess(it) } .catch { e -> onError(e) }.launchIn(scope) } @raulhernandezl
  64. SearchTweetUseCase: catch @ExperimentalCoroutinesApi override fun execute(query: String, callbackInput: SearchCallback?) {

    ... repository.searchTweet(query) .onEach { onSuccess(it) } .catch { e -> onError(e) }.launchIn(scope) } @raulhernandezl
  65. SearchTweetUseCase: scope @ExperimentalCoroutinesApi override fun execute(query: String, callbackInput: SearchCallback?) {

    ... repository.searchTweet(query) .onEach { onSuccess(it) } .catch { e -> onError(e) }.launchIn(scope) } @raulhernandezl
  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 dispose() { if (disposable != null &&

    !disposable.isDisposed()) { disposable.dispose(); disposable = null; } callback = null; } SearchTweetUseCase: observables disposal .subscribeOn(taskThreading.computation()) .observeOn(taskThreading.ui()) @raulhernandezl
  69. private val computationScheduler = Schedulers.computation() private val ioScheduler = Schedulers.io()

    private val mainScheduler = AndroidSchedulers.mainThread() Recap: Schedulers for RxJava @raulhernandezl
  70. class TaskThreadingImpl @Inject constructor() : TaskThreading { private val computationScheduler

    = Schedulers.computation() private val ioScheduler = Schedulers.io() private val mainScheduler = AndroidSchedulers.mainThread() ... } Can we use the same RxJava schedulers / dispatchers? @raulhernandezl
  71. 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
  72. class TaskThreadingImpl @Inject constructor() : TaskThreading { private val computationScheduler

    = Schedulers.computation() private val ioScheduler = Schedulers.io() private val mainScheduler = AndroidSchedulers.mainThread() private val computationCoroutineContext = computationScheduler.asCoroutineDispatcher() private val ioCoroutineContext = ioScheduler.asCoroutineDispatcher() private val mainCoroutineContext = mainScheduler.asCoroutineDispatcher() TaskThreading: same Schedulers transformed to CoroutineDispatchers kotlinx-coroutines-rx2 @raulhernandezl
  73. ... fun getSearchTweets(query: String): Flow<List<Tweet>> = flow<List<Tweet>> { ... }.flowOn(taskThreading.ioDispatcher()).asObserv..

    Reminder: TweetsRepository doesn’t need to return Observable anymore, just Flow @raulhernandezl
  74. 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
  75. class ViewDelegate constructor( private val presenter: SearchTweetPresenter ) { private

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

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

    onDispose() { view.setOnQueryTextListener(null) disposable.dispose() } fun observeSubject(): Disposable { ... } } ViewDelegate: RxJava disposal @raulhernandezl
  78. class ViewDelegate constructor( private val presenter: SearchTweetPresenter ) { fun

    prepareViewDelegateListener(val view: SearchView): Flow<String>= ... } ViewDelegate: Flows with channels as flows @raulhernandezl
  79. @ExperimentalCoroutinesApi @FlowPreview class ViewDelegate constructor( private val presenter: SearchTweetPresenter )

    { fun prepareViewDelegateListener(val view: SearchView): Flow<String> = (channelFlow<String> { // define listener ... }) } ViewDelegate: Flows with channelFlow builder @raulhernandezl
  80. @ExperimentalCoroutinesApi @FlowPreview class ViewDelegate constructor( private val presenter: SearchTweetPresenter )

    { fun prepareViewDelegateListener(val view: SearchView): Flow<String> = (channelFlow<String> { // define listener awaitClose { // close listener } }) } ViewDelegate: Flows self-closing themselves, awaitClose @raulhernandezl
  81. @ExperimentalCoroutinesApi @FlowPreview class ViewDelegate constructor( private val presenter: SearchTweetPresenter )

    { fun prepareViewDelegateListener(val view: SearchView): Flow<String> = (channelFlow<String> { // define listener awaitClose { // close listener } }).flowOn(taskThreading.ioDispatcher()) } ViewDelegate: Flows with flowOn @raulhernandezl
  82. @ExperimentalCoroutinesApi @FlowPreview class ViewDelegate constructor( private val presenter: SearchTweetPresenter )

    { fun prepareViewDelegateListener(val view: SearchView): Flow<String> = (channelFlow<String> { val listener = ViewListener(channel) ... awaitClose { ... } }).flowOn(taskThreading.ioDispatcher()) } ViewDelegate: Flows with channelFlow builder channel @raulhernandezl
  83. @ExperimentalCoroutinesApi @FlowPreview class ViewDelegate constructor( private val presenter: SearchTweetPresenter )

    { fun prepareViewDelegateListener(val view: SearchView): Flow<String> = (channelFlow<String> { val listener = ViewListener(channel) ... awaitClose { ... } }).flowOn(taskThreading.ioDispatcher()) } ViewDelegate: Flows with channelFlow builder channel @raulhernandezl
  84. @ExperimentalCoroutinesApi @FlowPreview class ViewDelegate constructor( private val presenter: SearchTweetPresenter )

    { fun prepareViewDelegateListener(val view: SearchView): Flow<String> = (channelFlow<String> { val listener = ViewListener(channel) view.setOnQueryTextListener(listener) awaitClose { view.setOnQueryTextListener(null) } }).flowOn(taskThreading.ioDispatcher()) } ViewDelegate: Flows setting the query listener @raulhernandezl
  85. class ViewDelegate constructor( private val presenter: SearchTweetPresenter ) { private

    val publishSubject: PublishSubject<String> = PublishSubject.create() fun prepareViewDelegateListener(val view: SearchView) { val listener = ViewListener(publishSubject) view.setOnQueryTextListener(listener) } fun onDispose() { view.setOnQueryTextListener(null) disposable.dispose() } } ViewDelegate: RxJava needs to manually dispose the view and disposable @raulhernandezl
  86. class ViewListener constructor( private val publishSubject: PublishSubject<String> ): SearchView.OnQueryTextListener {

    ´ override fun onQueryTextChange(query: String): Boolean { ... return true } override fun onQueryTextSubmit(query: String) = false } ViewListener: injected PublishSubject @raulhernandezl
  87. class ViewListener 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 } ViewListener: RxJava with publishSubject @raulhernandezl
  88. class ViewListener( @ExperimentalCoroutinesApi constructor( private val queryChannel: SendChannel<String> ): SearchView.OnQueryTextListener

    { ´ override fun onQueryTextChange(query: String): Boolean { ... return true } override fun onQueryTextSubmit(query: String) = false } ViewListener: Channels with SendChannel @raulhernandezl
  89. class ViewListener( @ExperimentalCoroutinesApi 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 } ViewListener: Channels with SendChannel @raulhernandezl
  90. class ViewDelegate constructor( private val presenter: SearchTweetPresenter ) { fun

    observeSubject(): Disposable = publishSubject .observeOn(taskThreading.ui()) .subscribeOn(taskThreading.computation()) ... } ViewDelegate: RxJava observes @raulhernandezl
  91. class ViewDelegate constructor( private val presenter: SearchTweetPresenter ) { fun

    observeSubject(): Disposable = publishSubject .observeOn(taskThreading.ui()) .subscribeOn(taskThreading.computation()) .debounce(600, TimeUnit.MILLISECONDS) .distinctUntilChanged() .filter { filterQuery(it) } .subscribe ( ... ) } ViewDelegate: RxJava operators @raulhernandezl
  92. class ViewDelegate constructor( private val presenter: SearchTweetPresenter ) { fun

    observeSubject(): Disposable = publishSubject .observeOn(taskThreading.ui()) .subscribeOn(taskThreading.computation()) .debounce(600, TimeUnit.MILLISECONDS) .distinctUntilChanged() .filter { filterQuery(it) } .subscribe ( presenter.searchTweet(it), presenter::showProblemHappened ) } ViewDelegate: RxJava uses Presenter @raulhernandezl
  93. @ExperimentalCoroutinesApi @FlowPreview fun observeChannelAsFlow() { scope.launch { declareViewDelegate() .debounce(600) .distinctUntilChanged()

    .filter { filterQuery(it) } .flowOn(taskThreading.computationDispatcher()) ... } } ViewDelegate: with Flow @raulhernandezl
  94. @ExperimentalCoroutinesApi @FlowPreview fun observeChannelAsFlow() { scope.launch { declareViewDelegate() .debounce(600) .distinctUntilChanged()

    .filter { filterQuery(it) } .flowOn(taskThreading.computationDispatcher()) .catch { captureError(e) } .collect { try { presenter.searchTweet(it) } catch (e: Exception) { presenter::showProblemHappened } } } } ViewDelegate: with Flow @raulhernandezl
  95. @ExperimentalCoroutinesApi @FlowPreview fun observeChannelAsFlow() { declareViewDelegate() .debounce(600) .distinctUntilChanged() .filter {

    filterQuery(it) } .flowOn(taskThreading.computationDispatcher()) .onEach { presenter.searchTweet(it) } .catch { presenter::showProblemHappened } .launchIn(scope) } ViewDelegate: with Flow @raulhernandezl
  96. class TaskThreadingImpl @Inject constructor() : TaskThreading { private val computationScheduler

    = Schedulers.computation() private val ioScheduler = Schedulers.io() private val mainScheduler = AndroidSchedulers.mainThread() private val computationCoroutineContext = computationScheduler.asCoroutineDispatcher() private val ioCoroutineContext = ioScheduler.asCoroutineDispatcher() private val mainCoroutineContext = mainScheduler.asCoroutineDispatcher() Reuse of schedulers as dispatchers kotlinx-coroutines-rx2 @raulhernandezl
  97. class TaskThreadingImpl @Inject constructor() : TaskThreading { private val computationScheduler

    = Schedulers.computation() private val ioScheduler = Schedulers.io() private val mainScheduler = AndroidSchedulers.mainThread() private val computationCoroutineContext = // ? private val ioCoroutineContext = // ? private val mainCoroutineContext = // ? Remove reuse schedulers as dispatchers @raulhernandezl
  98. 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
  99. 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