Trip into the async world

This deck is the base of the presentation about migrating from RxJava to Coroutines I gave at Conference for Kotliners, in Budapest

1. Trip into the async world
2. Rob
3. Why RxJava?
4. Why RxJava? Retrofit
5. Why RxJava? Retrofit Async ops
6. Why RxJava? Retrofit Async ops Basic operators
7. Why RxJava? Retrofit Async ops Basic operators Advanced operators
8. Why RxJava? Retrofit Async ops Basic operators Advanced operators
9. Why RxJava? Retrofit Async ops Basic operators Advanced operators
10. implementation 'com.squareup.retrofit2:retrofit:2.6.0’
11. @GET("/path/to/resource") fun networkRequest(): Single<Data> @GET("/path/to/resource") suspend fun networkRequest(): Data
12. Retrofit.Builder() ... .addCallAdapterFactory(CoroutineCallAdapterFactory()) .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) .build()
13. CoroutineCallAdapterFactory() RxJava2CallAdapterFactory.create()
14. @Test fun test() { runBlocking { val response = api.networkRequest() assertThat(response)... } api.networkRequest() .test() .assertValue { } }
16. class DataViewModel()) : ViewModel() { private val subscriptions = CompositeDisposable() val output = BehaviorSubject<LCE<Data>>.create() fun loadCurrentStatus() { disposable += store.get(BarCode.empty()) .startWith(LceLoading()) .onErrorReturn({e -> LceError(e)}) .map({data -> mapEmptyData(data)}) .observeOn(AndroidSchedulers.mainThread()) .subscribeOn(Schedulers.io()) .subscribe(output) } override fun onCleared() { disposable.clear() } }
17. class DataViewModel()) : ViewModel() { private val subscriptions = CompositeDisposable() val output = BehaviorSubject<LCE<Data>>.create() fun loadCurrentStatus() { disposable += store.get(BarCode.empty()) .startWith(LceLoading()) .onErrorReturn({e -> LceError(e)}) .map({data -> mapEmptyData(data)}) .observeOn(AndroidSchedulers.mainThread()) .subscribeOn(Schedulers.io()) .subscribe(output) } override fun onCleared() { disposable.clear() } } Hidden knowledge
18. te val subscriptions = CompositeDisposable() utput = BehaviorSubject<LCE<Data>>.create() oadCurrentStatus() { isposable += store.get(BarCode.empty()) .startWith(LceLoading()) .onErrorReturn({e -> LceError(e)}) .map({data -> mapEmptyData(data)}) .observeOn(AndroidSchedulers.mainThread()) .subscribeOn(Schedulers.io()) .subscribe(output) ide fun onCleared() { isposable.clear() Learning curve
19. class DataViewModel()) : ViewModel() { val output = MutableLiveData<Data>() fun loadCurrentStatus() { output.postValue(LceLoading()) viewModelScope.launch { try { val data = store.get(BarCode.empty()).await() if (data.isEmpty()) { output.postValue(LceError(EmptyResultSetException())) } else { output.postValue(LceSuccess(data)) } } catch (e: Exception) { output.postValue(LceError(e)) } } } }
20. class DataViewModel()) : ViewModel() { val output = MutableLiveData<Data>() fun loadCurrentStatus() { output.postValue(LceLoading()) viewModelScope.launch { try { val data = store.get(BarCode.empty()).await() if (data.isEmpty()) { output.postValue(LceError(EmptyResultSetException())) } else { output.postValue(LceSuccess(data)) } } catch (e: Exception) { output.postValue(LceError(e)) } } } }
21. adCurrentStatus() { output.postValue(LceLoading()) viewModelScope.launch { try { val data = store.get(BarCode.empty()).await() if (data.isEmpty()) { output.postValue(LceError(EmptyResultSetException())) } else { output.postValue(LceSuccess(data)) } } catch (e: Exception) { output.postValue(LceError(e)) } }
22. map flatmap filter first last
23. map flatmap filter first last forEach orEmptyT mapNotNull
24. @NotNull public static <T> Observable<Response<T>> from(@NotNull final ApolloCall<T> call) { checkNotNull(call, "call == null"); return Observable.create(new ObservableOnSubscribe<Response<T>>() { @Override public void subscribe(final ObservableEmitter<Response<T>> emitter) throws Exception { cancelOnObservableDisposed(emitter, call); call.enqueue(new ApolloCall.Callback<T>() { @Override public void onResponse(@NotNull Response<T> response) { if (!emitter.isDisposed()) { emitter.onNext(response); } } @Override public void onFailure(@NotNull ApolloException e) { Exceptions.throwIfFatal(e); if (!emitter.isDisposed()) { emitter.onError(e); } } @Override public void onStatusEvent(@NotNull ApolloCall.StatusEvent event) { if (event == ApolloCall.StatusEvent.COMPLETED && !emitter.isDisposed()) { emitter.onComplete(); } } }); } }); }
25. @NotNull public static <T> Observable<Response<T>> from(@NotNull final ApolloCall<T> call) { checkNotNull(call, "call == null"); return Observable.create(new ObservableOnSubscribe<Response<T>>() { @Override public void subscribe(final ObservableEmitter<Response<T>> emitter) throws Exception { cancelOnObservableDisposed(emitter, call); call.enqueue(new ApolloCall.Callback<T>() { @Override public void onResponse(@NotNull Response<T> response) { if (!emitter.isDisposed()) { emitter.onNext(response); } } @Override public void onFailure(@NotNull ApolloException e) { Exceptions.throwIfFatal(e); if (!emitter.isDisposed()) { emitter.onError(e); } } @Override public void onStatusEvent(@NotNull ApolloCall.StatusEvent event) { if (event == ApolloCall.StatusEvent.COMPLETED && !emitter.isDisposed()) { emitter.onComplete(); } } }); } }); } suspend fun <T> ApolloCall<T>.await(): Response<T> = suspendCancellableCoroutine { cont -> enqueue(object : ApolloCall.Callback<T>() { override fun onFailure(e: ApolloException) { cont.resumeWithException(e) } override fun onResponse(response: Response<T>) { cont.resume(response) } }) cont.invokeOnCancellation { cancel() } }
26. suspend fun <T> ApolloCall<T>.await(): Response<T> = suspendCancellableCoroutine { cont -> enqueue(object : ApolloCall.Callback<T>() { override fun onFailure(e: ApolloException) { cont.resumeWithException(e) } override fun onResponse(response: Response<T>) { cont.resume(response) } }) cont.invokeOnCancellation { cancel() } }
27. val job = Job() val scope = CoroutineScope(Dispatchers.IO + job) fun doThings() { scope.launch { ... } } fun inTheEnd() { job.cancel() }
28. val job = Job() val scope = CoroutineScope(Dispatchers.IO + job) fun doThings() { scope.launch { ... } } fun inTheEnd() { job.cancelChildren() scope.coroutineContext.cancelChildren() }
29. job.cancelChildren() scope.coroutineContext.cancelChildren() public fun Job.cancelChildren(cause: CancellationException? = null) { children.forEach { it.cancel(cause) } } public fun CoroutineContext.cancelChildren(cause: CancellationException? = null) { this[Job]?.children?.forEach { it.cancel(cause) } }
31. open.nytimes.com @nytdev github.com/NYTimes developers.nytimes.com
32. QA &