RxJava Easy Wins - Droidcon NYC 2014

Everyone keeps talking about Reactive Programming and RxJava, but how is it actually used in Android programming? This talk will discuss the building blocks of reactive programming, Observables, and how to understand the core parts of push-not-pull programming. Learn how to easily do work on a background thread in one line while still updating your UI on the main thread, easily implement a Publish-Subscribe pattern with Observables and Subjects, and take advantage of RxJava to make form-based UIs simple. Finally, we'll look at how using Observables can improve testing and ensure that your code will not block the UI thread if operations are blocked.

Transcript

1. Ron Shapiro · @rdshapiro DroidconNYC 2014 Easy wins to help get you in the groove RxJava + Android
2. Quick Survey Who has played around with RxJava? Who uses RxJava in production? compile 'io.reactivex:rxjava:1.+' formerly: compile 'com.netflix.rxjava:rxjava-core:+'
3. What is Rx? Let's take a step back. Imperative programming:
4. What is Rx? int a = 1; int b = a + 2; a = 5; assertTrue(3, b); Let's take a step back. Imperative programming:
5. A "real-world" example of Reactive Programming
6. A "real-world" example of Reactive Programming Excel spreadsheet:
7. A "real-world" example of Reactive Programming Excel spreadsheet: A 1 2 2 4 3 =A1 + A2
8. A "real-world" example of Reactive Programming Excel spreadsheet: A 1 2 2 4 3 =A1 + A2 A 1 2 2 4 3 6
9. A "real-world" example of Reactive Programming Excel spreadsheet: A 1 2 2 4 3 =A1 + A2 A 1 2 2 4 3 6 A 1 2 2 5 3 7
10. RxJava Contract void onNext(T t); void onError(Throwable throwable); void onCompleted();
11. RxJava Contract void onNext(T t); void onError(Throwable throwable); void onCompleted(); T get(); Iterable<T> get(); vs. Imperative Programming
12. RxJava Contract void onNext(T t); void onError(Throwable throwable); void onCompleted(); T get() throws Throwable; Iterable<T> get() throws Throwable; vs. Imperative Programming
13. In Action Observable.just(1, 2, 3, 4) .subscribe(new Subscriber<Integer>() { @Override public void onNext(Integer integer) { System.out.println(integer); } @Override public void onCompleted() { System.out.println("Done!"); } ! @Override public void onError(Throwable e) { System.err.println("An error occurred: " + e.getMessage()); } }); ! > 1 > 2 > 3 > 4 > Done!
14. Java 8 to the Rescue!
15. Java 8 to the Rescue! Java 8 to the Rescue!
16. Java 8 to the Rescue! Java 8 to the Rescue! Yes, on Android!
17. Retrolambda ! https://github.com/evant/gradle-retrolambda Observable.just(1, 2, 3, 4).subscribe( integer -> System.out.println(integer), error -> System.err.println("An error occurred: " + error.getMessage()), () -> System.out.println(“Done!"));
18. Observable.just(1, 2, 3, 4).subscribe( + integer -> System.out.println(integer), - System.out::println, error -> System.err.println("An error occurred: " + error.getMessage()), () -> System.out.println("Done!")); Retrolambda ! https://github.com/evant/gradle-retrolambda
19. Observable.just(1, 2, 3, 4).subscribe( + integer -> System.out.println(integer), - System.out::println, error -> System.err.println("An error occurred: " + error.getMessage()), () -> System.out.println("Done!")); Retrolambda ! https://github.com/evant/gradle-retrolambda Not only is this more concise, Java 8 lambdas don't have an implicit reference to their enclosing block's class like anonymous classes do (unless you explicitly refer to MyClass.this)
20. So why would I actually use this? Ok.
21. Long running tasks
22. (Potentially) long running tasks
23. (Potentially) long running tasks db.loadSomeData() .subscribe(data -> textView.setText(data.toString));
24. (Potentially) long running tasks db.loadSomeData() + .subscribeOn(Schedulers.io()) + .observeOn(AndroidSchedulers.mainThread()) .subscribe(data -> textView.setText(data.toString));
25. (Potentially) long running tasks Observable<Data> loadSomeData() { // loading is un-opinionated about // what thread it is running on } db.loadSomeData() + .subscribeOn(Schedulers.io()) + .observeOn(AndroidSchedulers.mainThread()) .subscribe(data -> textView.setText(data.toString));
26. (Potentially) long running tasks Observable<Data> loadSomeData() { // loading is un-opinionated about // what thread it is running on } db.loadSomeData() .subscribeOn(Schedulers.newThread()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(data -> textView.setText(data.toString));
27. Observables + SQL
28. Observables + SQL shameless plug for github.com/venmo/cursor-utils
29. Observables + SQL shameless plug for github.com/venmo/cursor-utils IterableCursor<T>: turn a Cursor row into a T IterableCursorAdapter<T> adapter = new IterableCursorAdapter<T>(context); db.loadSomeData() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(adapter::changeCursor);
30. Observables + SQL Because IterableCursor<T> extends Iterable<T>, you can easily transform your queries into Observable<T>s:
31. Observables + SQL db.loadSomeData() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .flatMap(cursor -> Observable.from(cursor)) .onNext(pojo -> /* ... */); Because IterableCursor<T> extends Iterable<T>, you can easily transform your queries into Observable<T>s:
32. Observables + SQL Because IterableCursor<T> extends Iterable<T>, you can easily transform your queries into Observable<T>s: db.loadSomeData() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .flatMap(cursor -> Observable.from(cursor)) - .onNext(pojo -> /* ... */); + .toList() + .onNext(pojoList -> /* ... */);
33. Observables have a semi-declaritive syntax
34. Observables have a semi-declaritive syntax // detect when the user pauses typing, // then search the api ViewObservable.text(myTextView) .map(TextView::getText) .delay(400, TimeUnit.MILLISECONDS) .filter(delayedText -> TextUtils.equals( delayedText, myTextView.getText())) .subscribe(api::search);
35. Observables have a semi-declaritive syntax // detect when the user pauses typing, // then search the api ViewObservable.text(myTextView) .map(TextView::getText) + .map(CharSequence::toString()) .delay(400, TimeUnit.MILLISECONDS) .filter(delayedText -> TextUtils.equals( delayedText, myTextView.getText())) .subscribe(api::search); // make sure you're always dealing with immutable objects!
36. Form Validation
37. Form Validation Observable<Boolean> validName = ViewObservable .text(nameTextView) .map(view -> TextUtils.isEmpty(view.getText())); Observable<Boolean> validTermsOfService = ViewObservable.input(tosCheckBox, true); Observable<Boolean> validBirthdate = // ...
38. Form Validation
39. Form Validation Func3<Boolean, Boolean, Boolean, Boolean> combinator = (name, tos, date) -> name && tos && date); Observable.combineLatest(validName, validTermsOfService, validBirthdate, combinator) .subscribe(allValid -> { if (allValid) { submitButton.setVisibility(View.VISIBLE); } else { submitButton.setVisibility(View.GONE); } });
40. Form Validation Func3<Boolean, Boolean, Boolean, Boolean> combinator = (name, tos, date) -> name && tos && date); Observable.combineLatest(validName, validTermsOfService, validBirthdate, combinator) .subscribe(allValid -> { if (allValid) { submitButton.setVisibility(View.VISIBLE); } else { submitButton.setVisibility(View.GONE); } }); combineLatest and FuncX are type-safe! Adding a new input source in one area but not the other will break at compile-time instead of a silent runtime mistake
41. Composability
42. sp? Composability
43. API Client
44. API Client Observable<Metadata> getMetaData() {/* ... */} Observable<Permalink> postPicture(String url, Bitmap b) {/* ... */} Observable<Void> linkPostToPermalink(Strind id, Permalink p) {/* ... */}
45. API Client Observable<Metadata> getMetaData() {/* ... */} Observable<Permalink> postPicture(String url, Bitmap b) {/* ... */} Observable<Void> linkPostToPermalink(Strind id, Permalink p) {/* ... */} ! // if any of these fail, i.e. call `onError` instead of `onCompleted`, // the subsequent calls are not made getMetaData() .flatMap(metadata -> postPicture(metadata.getUrl(), bitmap) .flatMap(permalink -> linkPostToPermalink(postId, permalink) .subscribe(/* ... */);
46. API Client Observable<Metadata> getMetaData() {/* ... */} Observable<Permalink> postPicture(String url, Bitmap b) {/* ... */} Observable<Void> linkPostToPermalink(Strind id, Permalink p) {/* ... */} ! // if any of these fail, i.e. call `onError` instead of `onCompleted`, // the subsequent calls are not made getMetaData() .flatMap(metadata -> postPicture(metadata.getUrl(), bitmap) .flatMap(permalink -> linkPostToPermalink(postId, permalink) + .subscribeOn(Schedulers.io()) + .observeOn(AndroidSchedulers.mainThread()) .subscribe(/* ... */);
47. API Client
48. API Client Observable<T1> first = // ... Observable<T2> second = // ... Observable<T3> third = // ... Func3<T1, T2, T3, R> combineFunc = // ...
49. API Client // wait for all 3 to complete Observable.combineLatest(first, second, third, combineFunc) .subscribe((R r) -> doSomethingAwesome()); Observable<T1> first = // ... Observable<T2> second = // ... Observable<T3> third = // ... Func3<T1, T2, T3, R> combineFunc = // ...
50. API Client // wait for all 3 to complete Observable.combineLatest(first, second, third, combineFunc) .subscribe((R r) -> doSomethingAwesome()); Observable<T1> first = // ... Observable<T2> second = // ... Observable<T3> third = // ... Func3<T1, T2, T3, R> combineFunc = // ... Make sure each of first, second, and third handles off-loading/concurrency concerns. !
51. RxEventBus
52. RxEventBus Subject<T, R> // observe a T, emit an R (can be the same) AsyncSubject: last emitted item BehaviorSubject: most recent + all future items PublishSubject: all future items ReplaySubject: caches all* items, emits each on subscribing
53. RxEventBus class RxEventBus { private Subject<Object, Object> subject = PublishSubject.create(); <T> Subscription subscribe(Action1<? super T> action, Class<T> clazz) { return subject.ofType(clazz).subscribe(action); } <T> void post(T t) { subject.onNext(t); } <T> void post(Observable<T> observable) { observable.subscribe(subject); } }
54. RxEventBus class RxEventBus { // ... /** Forwarding terminal events will terminate your Subject! */ <T> void post(Observable<T> observable) { observable.materialize().map((Notification<T> notification) -> { if (notification.isOnNext()) { return notification.getValue(); } else if (notification.isOnError()) { return new OnErrorEvent(notification.getThrowable()); } return new OnCompletedEvent(); }).subscribe(subject); } }
55. RxEventBus class RxEventBus { // ... /** Forwarding terminal events will terminate your Subject! */ <T> void post(Observable<T> observable) { observable.materialize().map((Notification<T> notification) -> { if (notification.isOnNext()) { return notification.getValue(); } else if (notification.isOnError()) { return new OnErrorEvent(notification.getThrowable()); } return new OnCompletedEvent(); }).subscribe(subject); } }
56. Simplified Testing
57. Simplified Testing <T> Observable<T> sleepObservable(long timeout, TimeUnit unit) { return Observable.create((Subscriber<? super T> subscriber) -> { SystemClock.sleep(unit.toMillis(timeout)); subscriber.onCompleted(); }); }
58. Simplified Testing <T> Observable<T> sleepObservable(long timeout, TimeUnit unit) { return Observable.create((Subscriber<? super T> subscriber) -> { SystemClock.sleep(unit.toMillis(timeout)); subscriber.onCompleted(); }); } Database db = Mockito.spy(db); doAnswer(mockedMethod -> { return sleepObservable(2, SECONDS) .concatWith(mockedMethod.callRealMethod()); }).when(db).loadSomeData()
59. Simplified Testing AtomicBoolean wasClicked = new AtomicBoolean(false); new Thread(() -> { SystemClock.sleep(1000); assertTrue(wasClicked.get()); }).start(); onView(withId(R.id.my_view) .perform(setOnClickListener((v) -> wasClicked.set(true)) .perform(click(); // if your data is being loaded on the main-thread, // the view will not be clicked
60. Look familiar? UNEXPECTED TOP-LEVEL EXCEPTION: java.lang.IllegalArgumentException: method ID not in [0, 0xffff]: 65536 at com.android.dx.merge.DexMerger$6.updateIndex(DexMerger.java:501) at com.android.dx.merge.DexMerger$IdMerger.mergeSorted(DexMerger.java:276) at com.android.dx.merge.DexMerger.mergeMethodIds(DexMerger.java:490) at com.android.dx.merge.DexMerger.mergeDexes(DexMerger.java:167) at com.android.dx.merge.DexMerger.merge(DexMerger.java:188) at com.android.dx.command.dexer.Main.mergeLibraryDexBuffers(Main.java:439) at com.android.dx.command.dexer.Main.runMonoDex(Main.java:287) at com.android.dx.command.dexer.Main.run(Main.java:230) at com.android.dx.command.dexer.Main.main(Main.java:199) at com.android.dx.command.Main.main(Main.java:103)
61. Look familiar? UNEXPECTED TOP-LEVEL EXCEPTION: java.lang.IllegalArgumentException: method ID not in [0, 0xffff]: 65536 at com.android.dx.merge.DexMerger$6.updateIndex(DexMerger.java:501) at com.android.dx.merge.DexMerger$IdMerger.mergeSorted(DexMerger.java:276) at com.android.dx.merge.DexMerger.mergeMethodIds(DexMerger.java:490) at com.android.dx.merge.DexMerger.mergeDexes(DexMerger.java:167) at com.android.dx.merge.DexMerger.merge(DexMerger.java:188) at com.android.dx.command.dexer.Main.mergeLibraryDexBuffers(Main.java:439) at com.android.dx.command.dexer.Main.runMonoDex(Main.java:287) at com.android.dx.command.dexer.Main.run(Main.java:230) at com.android.dx.command.dexer.Main.main(Main.java:199) at com.android.dx.command.Main.main(Main.java:103) RxJava is pretty big - you may/may not need ProGuard
62. Ron Shapiro shapiro.rd@gmail.com @rdshapiro ! venmo.com/ronshapiro Questions?