Adopting RxJava on Airbnb Andoid

Transcript

1. Adopting RxJava on Airbnb Android FELIPE LIMA / MARCH 14, 2016 / KOTLIN ∪ ANDROID MEETUP
2. Why RxJava?
3. We make shitty software
4. Mobile Engineering is hard
5. ReactiveX AN API FOR ASYNCHRONOUS PROGRAMMING WITH OBSERVABLE STREAMS
6. Streams
7. Reactive programming is hard
8. Too many concepts
9. • Observable & Observer Concepts
10. • Observable & Observer • Subscriber Concepts
11. • Observable & Observer • Subscriber • Subscription Concepts
12. • Observable & Observer • Subscriber • Subscription • Producer Concepts
13. • Observable & Observer • Subscriber • Subscription • Producer • Hot & Cold Observables Concepts
14. • Observable & Observer • Subscriber • Subscription • Producer • Hot & Cold Observables • Backpressure Concepts
15. • Observable & Observer • Subscriber • Subscription • Producer • Hot & Cold Observables • Backpressure • Scheduler Concepts
16. • Observable & Observer • Subscriber • Subscription • Producer • Hot & Cold Observables • Backpressure • Scheduler • Subject Concepts
17. • Observable & Observer • Subscriber • Subscription • Producer • Hot & Cold Observables • Backpressure • Scheduler • Subject • And more! Concepts
18. WAT?!
19. • Team Size: Getting everyone onboard and up to speed • Learning Curve: Steep, overwhelming • Terminology: Too many new concepts • Debugging: Unreadable stack traces Adoption Challenges
20. java.lang.IllegalStateException: Fatal Exception thrown on Scheduler.Worker thread. at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:62) at android.os.Handler.handleCallback(Handler.java:739) at android.os.Handler.dispatchMessage(Handler.java:95) at android.os.Looper.loop(Looper.java:145) at android.app.ActivityThread.main(ActivityThread.java:6141) at java.lang.reflect.Method.invoke(Method.java:-2) at java.lang.reflect.Method.invoke(Method.java:372) at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:1399) at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:1194) Caused by: rx.exceptions.OnErrorFailedException: Error occurred when trying to propagate error to Observer.onError at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:192) at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.checkTerminated(OperatorObserveOn.java:254) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.call(OperatorObserveOn.java:186) at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) at android.os.Handler.handleCallback(Handler.java:739) at android.os.Handler.dispatchMessage(Handler.java:95) at android.os.Looper.loop(Looper.java:145) at android.app.ActivityThread.main(ActivityThread.java:6141) at java.lang.reflect.Method.invoke(Method.java:-2) at java.lang.reflect.Method.invoke(Method.java:372) at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:1399) at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:1194) Caused by: rx.exceptions.CompositeException: 2 exceptions occurred. at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:192) at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.checkTerminated(OperatorObserveOn.java:254) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.call(OperatorObserveOn.java:186) at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) at android.os.Handler.handleCallback(Handler.java:739) at android.os.Handler.dispatchMessage(Handler.java:95) at android.os.Looper.loop(Looper.java:145) at android.app.ActivityThread.main(ActivityThread.java:6141) at java.lang.reflect.Method.invoke(Method.java:-2) at java.lang.reflect.Method.invoke(Method.java:372) at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:1399) at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:1194) Caused by: rx.exceptions.CompositeException$CompositeExceptionCausalChain: Chain of Causes for CompositeException In Order Received => at android.util.Log.getStackTraceString(Log.java:499) at com.android.internal.os.RuntimeInit.Clog_e(RuntimeInit.java:59) at com.android.internal.os.RuntimeInit.access$200(RuntimeInit.java:43) at com.android.internal.os.RuntimeInit$UncaughtHandler.uncaughtException(RuntimeInit.java:91) at com.bugsnag.android.ExceptionHandler.uncaughtException(ExceptionHandler.java:56) at java.lang.ThreadGroup.uncaughtException(ThreadGroup.java:693) at java.lang.ThreadGroup.uncaughtException(ThreadGroup.java:690) at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:66) at android.os.Handler.handleCallback(Handler.java:739) at android.os.Handler.dispatchMessage(Handler.java:95) at android.os.Looper.loop(Looper.java:145) at android.app.ActivityThread.main(ActivityThread.java:6141) at java.lang.reflect.Method.invoke(Method.java:-2) at java.lang.reflect.Method.invoke(Method.java:372) at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:1399) at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:1194) Caused by: com.airbnb.airrequest.NetworkException at com.airbnb.airrequest.ObservableFactory.lambda$responseMapper$1(ObservableFactory.java:59) at com.airbnb.airrequest.ObservableFactory.access$lambda$1(ObservableFactory.java:-1) at com.airbnb.airrequest.ObservableFactory$$Lambda$4.call(Unknown:-1) at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54) at rx.internal.operators.OperatorUnsubscribeOn$1.onNext(OperatorUnsubscribeOn.java:52) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.call(OperatorObserveOn.java:207) at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:422) at java.util.concurrent.FutureTask.run(FutureTask.java:237) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:152) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:265) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1112) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:587) at java.lang.Thread.run(Thread.java:818) Caused by: java.lang.ClassCastException: com.airbnb.android.responses.AirBatchResponse cannot be cast to com.airbnb.android.responses.ErrorResponse at com.airbnb.android.utils.NetworkUtil.error(NetworkUtil.java:357) at com.airbnb.android.utils.NetworkUtil.isExpiredOauthError(NetworkUtil.java:363) at com.airbnb.android.requests.base.ErrorLoggingAction.checkForExpiredToken(ErrorLoggingAction.java:96) at com.airbnb.android.requests.base.ErrorLoggingAction.call(ErrorLoggingAction.java:83) at com.airbnb.android.requests.base.ErrorLoggingAction.call(ErrorLoggingAction.java:33) at rx.Observable$10.onError(Observable.java:4561) at rx.internal.operators.OperatorDoOnEach$1.onError(OperatorDoOnEach.java:65) at com.airbnb.rxgroups.GroupSubscriptionTransformer$1.onError(GroupSubscriptionTransformer.java:45) at rx.Observable$30.onError(Observable.java:8280) at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:157) at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120) at rx.internal.operators.OperatorDoOnEach$1.onError(OperatorDoOnEach.java:71) at rx.subjects.SubjectSubscriptionManager$SubjectObserver.onError(SubjectSubscriptionManager.java:227) at rx.internal.operators.NotificationLite.accept(NotificationLite.java:147) at rx.subjects.ReplaySubject$UnboundedReplayState.accept(ReplaySubject.java:465) at rx.subjects.ReplaySubject$UnboundedReplayState.replayObserverFromIndex(ReplaySubject.java:514) at rx.subjects.ReplaySubject$UnboundedReplayState.replayObserver(ReplaySubject.java:502) at rx.subjects.ReplaySubject.caughtUp(ReplaySubject.java:427) at rx.subjects.ReplaySubject.onError(ReplaySubject.java:387) at rx.Observable$30.onError(Observable.java:8280) at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:157) at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.checkTerminated(OperatorObserveOn.java:254) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.call(OperatorObserveOn.java:186) at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) at android.os.Handler.handleCallback(Handler.java:739) at android.os.Handler.dispatchMessage(Handler.java:95) at android.os.Looper.loop(Looper.java:145) at android.app.ActivityThread.main(ActivityThread.java:6141) at java.lang.reflect.Method.invoke(Method.java:-2) at java.lang.reflect.Method.invoke(Method.java:372) at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:1399) at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:1194) Caused by: java.lang.ClassCastException: rx.exceptions.CompositeException cannot be cast to com.airbnb.airrequest.NetworkException at com.airbnb.android.requests.AirBatchRequestObserver.onError(AirBatchRequestObserver.java:60) at rx.Observable$30.onError(Observable.java:8280) at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:157) at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120) at rx.internal.operators.OperatorDoOnEach$1.onError(OperatorDoOnEach.java:68) at com.airbnb.rxgroups.GroupSubscriptionTransformer$1.onError(GroupSubscriptionTransformer.java:45) at rx.Observable$30.onError(Observable.java:8280) at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:157) at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120) at rx.internal.operators.OperatorDoOnEach$1.onError(OperatorDoOnEach.java:71) at rx.subjects.SubjectSubscriptionManager$SubjectObserver.onError(SubjectSubscriptionManager.java:227) at rx.internal.operators.NotificationLite.accept(NotificationLite.java:147) at rx.subjects.ReplaySubject$UnboundedReplayState.accept(ReplaySubject.java:465) at rx.subjects.ReplaySubject$UnboundedReplayState.replayObserverFromIndex(ReplaySubject.java:514) at rx.subjects.ReplaySubject$UnboundedReplayState.replayObserver(ReplaySubject.java:502) at rx.subjects.ReplaySubject.caughtUp(ReplaySubject.java:427) at rx.subjects.ReplaySubject.onError(ReplaySubject.java:387) at rx.Observable$30.onError(Observable.java:8280) at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:157) at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.checkTerminated(OperatorObserveOn.java:254) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.call(OperatorObserveOn.java:186) at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) at android.os.Handler.handleCallback(Handler.java:739) at android.os.Handler.dispatchMessage(Handler.java:95) at android.os.Looper.loop(Looper.java:145) at android.app.ActivityThread.main(ActivityThread.java:6141) at java.lang.reflect.Method.invoke(Method.java:-2) at java.lang.reflect.Method.invoke(Method.java:372) at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:1399) at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:1194)
21. Caused by: java.lang.ClassCastException: rx.exceptions.CompositeException cannot be cast to com.airbnb.airrequest.NetworkException at com.airbnb.android.requests.AirBatchRequestObserver.onError(AirBatchRequestObserver.java:60) at rx.Observable$30.onError(Observable.java:8280) at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:157) at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120) at rx.internal.operators.OperatorDoOnEach$1.onError(OperatorDoOnEach.java:68) at com.airbnb.rxgroups.GroupSubscriptionTransformer$1.onError(GroupSubscriptionTransformer.java:45) at rx.Observable$30.onError(Observable.java:8280) at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:157) at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120) at rx.internal.operators.OperatorDoOnEach$1.onError(OperatorDoOnEach.java:71) at rx.subjects.SubjectSubscriptionManager$SubjectObserver.onError(SubjectSubscriptionManager.java:227) at rx.internal.operators.NotificationLite.accept(NotificationLite.java:147) at rx.subjects.ReplaySubject$UnboundedReplayState.accept(ReplaySubject.java:465) at rx.subjects.ReplaySubject$UnboundedReplayState.replayObserverFromIndex(ReplaySubject.java:514) at rx.subjects.ReplaySubject$UnboundedReplayState.replayObserver(ReplaySubject.java:502) at rx.subjects.ReplaySubject.caughtUp(ReplaySubject.java:427) at rx.subjects.ReplaySubject.onError(ReplaySubject.java:387) at rx.Observable$30.onError(Observable.java:8280) at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:157) at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.checkTerminated(OperatorObserveOn.java:254) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.call(OperatorObserveOn.java:186) at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) at android.os.Handler.handleCallback(Handler.java:739) at android.os.Handler.dispatchMessage(Handler.java:95) at android.os.Looper.loop(Looper.java:145) at android.app.ActivityThread.main(ActivityThread.java:6141) at java.lang.reflect.Method.invoke(Method.java:-2) at java.lang.reflect.Method.invoke(Method.java:372) at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:1399) at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:1194)
22. Common Pitfalls
23. observeOn()
24. return observableFactory.<T>toObservable(this) .compose(this.<T>transform(observableRequest)) .observeOn(Schedulers.io()) .map(new ResponseMetadataOperator<>(this)) .flatMap(this::mapResponse) .observeOn(AndroidSchedulers.mainThread())
 .<AirResponse<T>>compose(group.transform(tag))
 .doOnError(new ErrorLoggingAction(request))
 .doOnError(NetworkUtil::checkForExpiredToken)
 .subscribe(request.observer());
25. subscribeOn()
26. return observableFactory.<T>toObservable(this) .compose(this.<T>transform(observableRequest)) .observeOn(Schedulers.io()) .map(new ResponseMetadataOperator<>(this)) .flatMap(this::mapResponse) .observeOn(AndroidSchedulers.mainThread())
 .<AirResponse<T>>compose(group.transform(tag))
 .doOnError(new ErrorLoggingAction(request))
 .doOnError(NetworkUtil::checkForExpiredToken) .subscribeOn(Schedulers.io())
 .subscribe(request.observer());
27. Error Handling
28. return observableFactory.<T>toObservable(this) .compose(this.<T>transform(observableRequest)) .observeOn(Schedulers.io()) .map(new ResponseMetadataOperator<>(this)) .flatMap(this::mapResponse) .observeOn(AndroidSchedulers.mainThread())
 .<AirResponse<T>>compose(group.transform(tag))
 .doOnError(new ErrorLoggingAction(request))
 .doOnError(NetworkUtil::checkForExpiredToken) .subscribeOn(Schedulers.io())
 .subscribe(request.observer());
29. return observableRequest .rawRequest() .<Observable<Response<T>>>newCall() .observeOn(Schedulers.io()) .unsubscribeOn(Schedulers.io()) .flatMap(responseMapper(airRequest)) .onErrorResumeNext(errorMapper(airRequest));
30. Unit Testing
31. @Test public void testErrorResponseNonJSON() { server.enqueue(new MockResponse() .setBody("something bad happened") .setResponseCode(500)); TestRequest request = new TestRequest.Builder<String>().build(); TestSubscriber<AirResponse<String>> subscriber = new TestSubscriber<>(); observableFactory.<String>toObservable(request).subscribe(subscriber); subscriber.awaitTerminalEvent(3L, TimeUnit.SECONDS); NetworkException exception = (NetworkException) subscriber.getOnErrorEvents().get(0); assertThat(exception.errorResponse(), equalTo(null)); assertThat(exception.bodyString(), equalTo("something bad happened")); }
32. @Test public void testUnicodeHeader() { server.enqueue(new MockResponse().setBody("\"Hello World\"")); TestRequest request = new TestRequest.Builder<String>() .header("Bogus", "Ӿ嶆櫮מ") .build(); observableFactory.toObservable(request) .toBlocking() .first(); RecordedRequest recordedRequest = server.takeRequest(); assertThat(recordedRequest.getHeader("Bogus"), equalTo("????")); }
33. Memory Leaks
34. private final CompositeSubscription pendingSubscriptions = new CompositeSubscription(); @Override public void onCreate() { pendingSubscriptions.add( observable.subscribe(observer)); } @Override public void onDestroy() { pendingSubscriptions.clear(); }
35. Additional Resources
36. Watch this: Reactive.community: Ben Christensen, Reactive Extensions at Netflix HTTPS://WWW.YOUTUBE.COM/WATCH?V=ET_SMMXKE5S
37. Questions?
38. Thanks!