Where the Reactive Rubber Meets the Road

Square's simple idiom for binding RxJava to the UI without boilerplate or side effects.

https://www.youtube.com/watch?v=va1d4MqLUGY

1. Where the Reactive Rubber Meets the Road
2. What could possibly go wrong? Hard Learned Lesson // We want to ensure that the animation end 
 // event happens after the transition has
 // actually finished. 
 newChild.post(new Runnable() {
 @Override public void run() {
 animationEnded(newChild);
 }
 });
3. What could possibly go wrong? Hard Learned Lesson // We want to ensure that the animation end 
 // event happens after the transition has
 // actually finished. 
 newChild.installLandMine(new Runnable() {
 @Override public void run() {
 animationEnded(newChild);
 }
 });
4. Every Post is a Race java.lang.NullPointerException: at com.squareup.ui.root.HomeDrawerPresenter.closeDrawer(HomeDrawerPresenter.java:303) at com.squareup.ui.root.HomeScreen$Presenter.onFlowAnimationEnd(HomeScreen.java:226) at com.squareup.ui.root.HomeView.onFlowAnimationEnd(HomeView.java:81) at com.squareup.flow.RegisterFlowContainerSupport.flowAnimationEnded(RegisterFlowContainerSupport.java:258) at com.squareup.flow.RegisterFlowContainerSupport.access$000(RegisterFlowContainerSupport.java:45) at com.squareup.flow.RegisterFlowContainerSupport$1.run(RegisterFlowContainerSupport.java:120) at android.os.Handler.handleCallback(Handler.java:733) at android.os.Handler.dispatchMessage(Handler.java:95) at android.os.Looper.loop(Looper.java:136) at android.app.ActivityThread.main(ActivityThread.java:5103) at java.lang.reflect.Method.invokeNative(Method.java:-2) at java.lang.reflect.Method.invoke(Method.java:515) at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:790) at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:606) at dalvik.system.NativeStart.main(NativeStart.java:-2)
5. ‣ Millions of downloads ‣ Every code path, every corner case will be hit every day ‣ Each post() adds another combinatorial dimension Every Race Will be Run
6. Keep your main thread synchronous
7. Embrace the net’s asynchrony Keep your main thread synchronous
8. Belly Burgers Unlike any Burger You’ve Ever Had. $1.99 fee • 40 - 60 min • $$ Taqueria Cazado… Ole! $1.99 fee • 40 - 60 min • $
9. Belly Burgers Unlike any Burger You’ve Ever Had. $1.99 fee • 40 - 60 min • $$ Taqueria Cazado… Ole! $1.99 fee • 40 - 60 min • $ Picasso
10. Belly Burgers Unlike any Burger You’ve Ever Had. $1.99 fee • 40 - 60 min • $$ Taqueria Cazado… Ole! $1.99 fee • 40 - 60 min • $ Model Picasso
11. RestaurantListView.java Ain’t Rx Grand? restaurants.restaurant(restaurantId)
 .observeOn(mainThread()) .subscribe(new Action1<Restaurant>() {
 @Override public void call(Restaurant restaurant) {
 nameView.setText(restaurant.name); // etc.
 }
 }));
12. RestaurantListView.java Command-minus Ain’t Rx Grand? restaurants.restaurant(restaurantId)
 .observeOn(mainThread()) .subscribe((Action1) (restaurant) -> {
 nameView.setText(restaurant.name); // etc.
 }));
13. RestaurantListView.java Command-minus Ain’t Rx Grand? restaurants.restaurant(restaurantId)
 .observeOn(mainThread()) .subscribe((Action1) (restaurant) -> {
 nameView.setText(restaurant.name); // etc.
 }));
14. rx.android.schedulers. HandlerScheduler Uh Oh public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) {
 if(this.compositeSubscription.isUnsubscribed()) {
 return Subscriptions.unsubscribed();
 } else {
 action = RxAndroidPlugins.getInstance().getSchedulersHook().onSchedule(action);
 final ScheduledAction scheduledAction = new ScheduledAction(action);
 scheduledAction.addParent(this.compositeSubscription);
 this.compositeSubscription.add(scheduledAction);
 this.handler.postDelayed(scheduledAction, unit.toMillis(delayTime));
 scheduledAction.add(Subscriptions.create(new Action0() {
 public void call() {
 HandlerWorker.this.handler.removeCallbacks(scheduledAction);
 }
 }));
 return scheduledAction;
 }
 }

15. rx.android.schedulers. HandlerScheduler Uh Oh public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) {
 if(this.compositeSubscription.isUnsubscribed()) {
 return Subscriptions.unsubscribed();
 } else {
 action = RxAndroidPlugins.getInstance().getSchedulersHook().onSchedule(action);
 final ScheduledAction scheduledAction = new ScheduledAction(action);
 scheduledAction.addParent(this.compositeSubscription);
 this.compositeSubscription.add(scheduledAction);
 this.handler.postDelayed(scheduledAction, unit.toMillis(delayTime));
 scheduledAction.add(Subscriptions.create(new Action0() {
 public void call() {
 HandlerWorker.this.handler.removeCallbacks(scheduledAction);
 }
 }));
 return scheduledAction;
 }
 }
 Just make it block if we’re already on the handler’s thread.
16. rx.android.schedulers. HandlerScheduler Uh Oh public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) {
 if(this.compositeSubscription.isUnsubscribed()) {
 return Subscriptions.unsubscribed();
 } else {
 action = RxAndroidPlugins.getInstance().getSchedulersHook().onSchedule(action);
 final ScheduledAction scheduledAction = new ScheduledAction(action);
 scheduledAction.addParent(this.compositeSubscription);
 this.compositeSubscription.add(scheduledAction);
 this.handler.postDelayed(scheduledAction, unit.toMillis(delayTime));
 scheduledAction.add(Subscriptions.create(new Action0() {
 public void call() {
 HandlerWorker.this.handler.removeCallbacks(scheduledAction);
 }
 }));
 return scheduledAction;
 }
 }
 If I could then I would… Just make it block if we’re already on the handler’s thread.
17. Belly Burgers Unlike any Burger You’ve Ever Had. $1.99 fee • 40 - 60 min • $$ Taqueria Cazado… Ole! $1.99 fee • 40 - 60 min • $ Model
18. Belly Burgers Unlike any Burger You’ve Ever Had. $1.99 fee • 40 - 60 min • $$ Taqueria Cazado… Ole! $1.99 fee • 40 - 60 min • $ Model
19. Rx is single-threaded by default
20. What starts on the main thread, stays on the main thread. restaurants.restaurant(restaurantId)
 .observeOn(mainThread()) .subscribe((Action1) (restaurant) -> {
 nameView.setText(restaurant.name); // etc.
 }));
21. RestaurantList.java Observables all the way down public Observable<Restaurant> restaurant(final String id) {
 return restaurantsSubject
 .map((Func1) (restaurants) -> {
 for (Restaurant r : restaurants) {
 if (r.id.equals(id)) return r;
 }
 return null;
 }) 
 .filter(notNull());
 }
22. RestaurantList.java Promise Keeper /** May be called from any thread. */ public void update( List<Restaurants> restaurants) {
 Observable.just(restaurants)
 .observeOn(mainThread())
 .subscribe((Action1) (list) {
 restaurantsSubject.onNext(list);
 });
 }

23. RestaurantList.java Better: Promise Enforcer /** Must be called from the main thread. */ public void update( List<Restaurants> restaurants) {
 checkMainThread();
 restaurantsSubject.onNext(restaurants);
 }
 void checkMainThread() {
 if (myLooper() != getMainLooper()) { throw new AssertionError( “Must be called from main thread.”); }
 }

24. Keep your main thread synchronous Embrace the net’s asynchrony Pick two
25. Bonus Hack Round: Cleaner stack traces service.doThing(thing)
 .subscribe((Action1) (response) {
 onResponse(response);
 },
 (e) {
 if (e instanceof RetrofitError) {
 onNetworkError(e);
 } else {
 throw new OnErrorNotImplementedException(e);
 }
 });
 If you must handle errors
26. @rjrjr https://squareup.com/careers