Common RxJava Mistakes

Talk given at Droidcon SF 2016.

Recording here: https://www.youtube.com/watch?v=QdmkXL7XikQ

1. Common RxJava Mistakes Dan Lew
2. Lambda Notation new Func1<String, Integer>() {
 @Override public Integer call(String s) {
 return s.length();
 }
 }; s -> s.length();
3. Lambda Notation new Func1<String, Integer>() {
 @Override public Integer call(String s) {
 return s.length();
 }
 }; s -> {
 return s.length();
 } s -> s.length()
4. Lambda Notation new Func1<String, Integer>() {
 @Override public Integer call(String s) {
 return s.length();
 }
 }; (String s) -> {
 return s.length();
 } s -> s.length() …is the same as…
5. Lambda Notation new Func1<String, Integer>() {
 @Override public Integer call(String s) {
 return s.length();
 }
 }; (String s) -> {
 return s.length();
 } s -> s.length() …is the same as… …is the same as…
6. The Stream
7. Observable<String> obs = Observable.just("Hello!");
 obs.map(s -> s.length());
 obs.subscribe(i -> System.out.println(i)); Observable.just("Hello!")
 .map(s -> s.length())
 .subscribe(i -> System.out.println(i)); Versus DOES NOT WORK
8. Observable<String> step1 = Observable.just("Hello!"); 
 Observable<Integer> step2 = step1.map(s -> s.length());
 step2.subscribe(i -> System.out.println(i));
9. Observable<String> step1 = Observable.just("Hello!"); 
 Observable<Integer> step2 = step1.map(s -> s.length());
 step2.subscribe(i -> System.out.println(i));
10. Observable<String> step1 = Observable.just("Hello!"); 
 Observable<Integer> step2 = step1.map(s -> s.length());
 step2.subscribe(i -> System.out.println(i));
11. Upstream Downstream
12. Creating Observables 101 • just(object) • from(iterable) • Whatever Retrofit is doing • create() NO
13. Difficulties • Observable contract • onNext* -> onCompleted | onError • Unsubscription • Backpressure
14. class MyClass {
 String value;
 
 Observable<String> value() {
 return Observable.create(subscriber -> {
 subscriber.onNext(value);
 subscriber.onCompleted();
 });
 }
 } ???
15. class MyClass {
 String value;
 
 Observable<String> value() {
 return Observable.create(subscriber -> {
 subscriber.onNext(value);
 subscriber.onCompleted();
 });
 }
 } BAD
16. class MyClass {
 String value;
 
 Observable<String> value() {
 return Observable.create(subscriber -> {
 subscriber.onNext(value);
 subscriber.onCompleted();
 });
 }
 } return Observable.fromCallable(() -> value);
17. class MyClass {
 String value;
 
 Observable<String> value() {
 return Observable.create(subscriber -> {
 subscriber.onNext(value);
 subscriber.onCompleted();
 });
 }
 } return Observable.defer(() -> Observable.just(value));
18. class MyClass {
 BehaviorSubject<String> value;
 
 Observable<String> value() {
 return value.asObservable();
 }
 
 void setValue(String s) {
 value.onNext(s);
 }
 }
19. RxRelay • Safer Subjects • https://github.com/JakeWharton/RxRelay
20. Lessons • Learn your creation methods! • Resort to create() when appropriate • (which is almost never)
21. Operator Misunderstandings
22. map vs. flatMap
27. Types • map: T -> R • flatMap: T -> Observable<R>
28. Exercise • Find an API with search • Query search results via Observable • Request extra data on results via Observable • Use only one subscribe()
29. flatMap vs. concatMap • Same, except… • #ordermatters • flatMap -> Unordered merge • concatMap -> Ordered merge
31. toList() • Item(s) -> List<Item> • “Why does nothing happen?!” • Missing onCompleted
32. Versus Key!
33. List of Lists Observable<List<String>> obs = /* ...etc... */;
 
 obs.flatMap(list -> Observable.from(list))
 .map(/* ...something... */)
 .toList();
 
 obs.flatMap(
 list -> Observable.from(list)
 .map(/* ...something... */)
 .toList()
 );
34. List of Lists Observable<List<String>> obs = /* ...etc... */;
 
 obs.flatMap(list -> Observable.from(list))
 .map(/* ...something... */)
 .toList();
 
 obs.flatMap(
 list -> Observable.from(list)
 .map(/* ...something... */)
 .toList()
 ); result: 1 list
35. List of Lists Observable<List<String>> obs = /* ...etc... */;
 
 obs.flatMap(list -> Observable.from(list))
 .map(/* ...something... */)
 .toList();
 
 obs.flatMap(
 list -> Observable.from(list)
 .map(/* ...something... */)
 .toList()
 ); result: 1 list result: N lists
36. take(1) vs. first() • take(1) - emits once, or not at all Observable.empty().take(1) • first() - emits once, or crashes Observable.empty().first() crash! emits nothing
37. cache() vs replay() • Don’t use cache() • Use replay() • cache() == replay().autoConnect()
38. Custom Operators
39. Operators I Have Written • OperatorNullToDefault • OperatorSortLists • OperatorSubscribeUntil • ContinueWithIfEmptyOperator
40. Operators I Should Have Written
41. Why not? • Each operator —> different flow • Framework handles common flows • Resort to custom operators for unique flows • Operators are hard!
42. • OperatorNullToDefault • OperatorSortLists • OperatorSubscribeUntil • ContinueWithIfEmptyOperator map() switchIfEmpty() takeUntil()
43. Threading
44. Observable.range(0, 5)
 .subscribe(); Observable.interval(1, TimeUnit.SECONDS)
 .subscribe(); current thread computation()
45. Observable.just("Hello")
 .subscribe(); Observable.just("Hello")
 .delay(1, TimeUnit.SECONDS)
 .subscribe(); current thread computation()
46. Observable.just("Hello")
 .subscribeOn(Schedulers.io())
 .map()
 .subscribeOn(Schedulers.computation())
 .subscribe(); io()
47. Observable.just("Hello")
 .observeOn(Schedulers.io())
 .map()
 .observeOn(Schedulers.computation())
 .subscribe(); io() computation() current thread
49. Lessons • Know your scheduler-based operators • Use subscribeOn once • Use observeOn liberally
50. Subscriptions
51. Observer<Integer> observer = /* some Observer */
 o1.subscribe(observer);
 o2.subscribe(observer); Subscriber<Integer> subscriber = /* some Subscriber */ o1.subscribe(subscriber);
 o2.subscribe(subscriber); Versus
52. • Observer is an interface • Subscriber is a class • Observer is stateless • Subscriber is stateful
53. Observer<Integer> observer = /* some Observer */
 o1.subscribe(observer);
 o2.subscribe(observer); Subscriber<Integer> subscriber = /* some Subscriber */ o1.subscribe(subscriber);
 o2.subscribe(subscriber); Versus Can blow up in your face
54. Memory Leaks • Subscriptions can leak! • Detect w/ LeakCanary • Solution: Unsubscribe
55. Unsubscribe Subscription subscription = Observable.just(1).subscribe();
 subscription.unsubscribe();
56. CompositeSubscription CompositeSubscription cs = new CompositeSubscription(); 
 cs.add(Observable.interval(1, TimeUnit.SECONDS).subscribe());
 cs.add(Observable.never().subscribe()); 
 cs.clear(); use clear()
57. RxLifecycle • Automatic sequence completion Observable.interval(1, TimeUnit.SECONDS)
 .compose(bindToLifecycle())
 .subscribe();
58. Error Handling
59. onError Uses • Unexpected exceptions • Unrecoverable streams • Flow control Blame Retrofit 1
60. Terminal events suck
61. onError ≈ try-catch
62. Exception workarounds • Error state • onErrorReturn • onErrorResumeNext
63. Debugging onError Observable.empty()
 .first()
 .subscribeOn(Schedulers.io())
 .subscribe(); Crash!
64. Debugging onError Exception in thread "RxCachedThreadScheduler-1" java.lang.IllegalStateException: Exception thrown on Scheduler.Worker thread. Add `onError` handling.
 at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:60)
 at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
 at java.util.concurrent.FutureTask.run(FutureTask.java:266)
 at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
 at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
 at java.lang.Thread.run(Thread.java:745)
 Caused by: rx.exceptions.OnErrorNotImplementedException: Sequence contains no elements
 at rx.Observable$26.onError(Observable.java:7881)
 at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:159)
 at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120)
 at rx.internal.operators.OperatorSubscribeOn$1$1$1.onError(OperatorSubscribeOn.java:71)
 at rx.internal.operators.OperatorSingle$ParentSubscriber.onCompleted(OperatorSingle.java:131)
 at rx.internal.operators.OperatorTake$1.onCompleted(OperatorTake.java:53)
 at rx.Observable$EmptyHolder$1.call(Observable.java:1073)
 at rx.Observable$EmptyHolder$1.call(Observable.java:1070)
 at rx.Observable$2.call(Observable.java:162)
 at rx.Observable$2.call(Observable.java:154)
 at rx.Observable$2.call(Observable.java:162)
 at rx.Observable$2.call(Observable.java:154)
 at rx.Observable.unsafeSubscribe(Observable.java:8098)
 at rx.internal.operators.OperatorSubscribeOn$1$1.call(OperatorSubscribeOn.java:62)
 at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55)
 ... 7 more
 Caused by: java.util.NoSuchElementException: Sequence contains no elements
 ... 18 more No references to my code!
65. Implement onError Observable.empty()
 .first()
 .subscribeOn(Schedulers.io())
 .subscribe(o -> System.out.println(o),
 err -> {
 throw new OnErrorNotImplementedException("Source!", err);
 }); Caused by: rx.exceptions.OnErrorNotImplementedException: Source!
 at net.danlew.experiments.Tester.lambda$main$1(Tester.java:52)
66. Implement onError Observable.empty()
 .first()
 .subscribeOn(Schedulers.io())
 .subscribe(o -> System.out.println(o),
 err -> {
 throw new OnErrorNotImplementedException("Source!", err);
 }); Caused by: rx.exceptions.OnErrorNotImplementedException: Source!
 at net.danlew.experiments.Tester.lambda$main$1(Tester.java:52)
67. Implement onError Observable.empty()
 .first()
 .subscribeOn(Schedulers.io())
 .subscribe(o -> System.out.println(o),
 err -> {
 throw new OnErrorNotImplementedException("Source!", err);
 }); Caused by: rx.exceptions.OnErrorNotImplementedException: Source!
 at net.danlew.experiments.Tester.lambda$main$1(Tester.java:52)
68. Sharing • Wanted: Multiple subscribers retrofitRequest.subscribe(/* do a thing */);
 retrofitRequest.subscribe(/* do another thing */); • Bad: Duplicate expensive operations
69. Hot vs. Cold
70. Sharing ConnectableObservable<Long> obs =
 Observable.fromCallable(() -> System.nanoTime())
 .publish();
 
 obs.subscribe(l -> System.out.println("1: " + l));
 obs.subscribe(l -> System.out.println("2: " + l));
 
 obs.connect();
71. Sharing ConnectableObservable<Long> obs =
 Observable.fromCallable(() -> System.nanoTime())
 .publish();
 
 obs.subscribe(l -> System.out.println("1: " + l));
 obs.subscribe(l -> System.out.println("2: " + l));
 
 obs.connect();
72. Sharing ConnectableObservable<Long> obs =
 Observable.fromCallable(() -> System.nanoTime())
 .publish();
 
 obs.subscribe(l -> System.out.println("1: " + l));
 obs.subscribe(l -> System.out.println("2: " + l));
 
 obs.connect();
73. Sharing ConnectableObservable<Long> obs =
 Observable.fromCallable(() -> System.nanoTime())
 .publish();
 
 obs.subscribe(l -> System.out.println("1: " + l));
 obs.subscribe(l -> System.out.println("2: " + l));
 
 obs.connect();
74. Sharing ConnectableObservable<Long> obs =
 Observable.fromCallable(() -> System.nanoTime())
 .publish();
 
 obs.subscribe(l -> System.out.println("1: " + l));
 obs.subscribe(l -> System.out.println("2: " + l));
 
 obs.connect();
75. refCount Observable<Long> obs =
 Observable.fromCallable(() -> System.nanoTime())
 .publish() .refCount();
 
 obs.subscribe(l -> System.out.println("1: " + l));
 obs.subscribe(l -> System.out.println("2: " + l)); // Output 1: 395093905388878 2: 395093906717841
76. refCount Observable<Long> obs =
 Observable.fromCallable(() -> System.nanoTime())
 .publish() .refCount();
 
 obs.subscribe(l -> System.out.println("1: " + l));
 obs.subscribe(l -> System.out.println("2: " + l)); // Output 1: 395093905388878 2: 395093906717841
77. refCount Observable<Long> obs =
 Observable.fromCallable(() -> System.nanoTime())
 .publish() .refCount();
 
 obs.subscribe(l -> System.out.println("1: " + l));
 obs.subscribe(l -> System.out.println("2: " + l)); // Output 1: 395093905388878 2: 395093906717841 huh?
78. Lessons • Share cold observables • Explicitly call connect() • Use autoConnect()
79. it’s over! @danlew42