Exploring RxJava 2 for Android (GOTOcph October 2016)

This talk will be an exploration on why reactive programming works so well for Android and how to use RxJava to apply it.

Video: https://www.youtube.com/watch?v=htIXKI5gOQU

1. Exploring RxJava Jake Wharton
2. Exploring RxJava 2 Jake Wharton
3. Why Reactive? Unless you can model your entire system synchronously...
4. Why Reactive? Unless you can model your entire system synchronously, a single asynchronous source breaks imperative programming.
5. Why Reactive? interface UserManager { User getUser();
 }A
6. Why Reactive? interface UserManager { User getUser(); void setName(String name); void setAge(int age);
 }A
7. Why Reactive? interface UserManager { User getUser(); void setName(String name); void setAge(int age);
 }A UserManager um = new UserManager();
8. Why Reactive? interface UserManager { User getUser(); void setName(String name); void setAge(int age);
 }A UserManager um = new UserManager(); System.out.println(um.getUser());
9. Why Reactive? interface UserManager { User getUser(); void setName(String name); void setAge(int age);
 }A UserManager um = new UserManager(); System.out.println(um.getUser()); um.setName("Jane Doe");
10. Why Reactive? interface UserManager { User getUser(); void setName(String name); void setAge(int age);
 }A UserManager um = new UserManager(); System.out.println(um.getUser()); um.setName("Jane Doe"); System.out.println(um.getUser());
11. Why Reactive? interface UserManager { User getUser(); void setName(String name); // <-- now async void setAge(int age); // <-- now async
 }A
12. Why Reactive? interface UserManager { User getUser(); void setName(String name); void setAge(int age);
 }A UserManager um = new UserManager(); System.out.println(um.getUser()); um.setName("Jane Doe"); System.out.println(um.getUser());
13. Why Reactive? interface UserManager { User getUser(); void setName(String name, Runnable callback); void setAge(int age, Runnable callback);
 }A
14. Why Reactive? interface UserManager { User getUser(); void setName(String name, Runnable callback);A void setAge(int age, Runnable callback);B
 }A UserManager um = new UserManager(); System.out.println(um.getUser()); um.setName("Jane Doe", new Runnable() { @Override public void run() { System.out.println(um.getUser()); }X });
15. Why Reactive? interface UserManager { User getUser(); void setName(String name, Listener listener);A void setAge(int age, Listener listener);B interface Listener { void success(User user); void failure(IOException e); }G
 }A
16. Why Reactive? UserManager um = new UserManager(); System.out.println(um.getUser()); um.setName("Jane Doe");
17. Why Reactive? UserManager um = new UserManager(); System.out.println(um.getUser()); um.setName("Jane Doe", new UserManager.Listener() { @Override public void success() { System.out.println(um.getUser()); }A @Override public void failure(IOException e) { // TODO show the error... }B });
18. Why Reactive? UserManager um = new UserManager(); System.out.println(um.getUser()); um.setName("Jane Doe", new UserManager.Listener() { @Override public void success() { System.out.println(um.getUser()); }A @Override public void failure(IOException e) { // TODO show the error... }B }); um.setAge(40, new UserManager.Listener() { @Override public void success() { System.out.println(um.getUser()); }C 2@Override public void failure(IOException e) {2 2// TODO show the error...2 }D });2
19. Why Reactive? UserManager um = new UserManager(); System.out.println(um.getUser()); um.setName("Jane Doe", new UserManager.Listener() { @Override public void success() { System.out.println(um.getUser()); um.setAge(40, new UserManager.Listener() { @Override public void success() { System.out.println(um.getUser()); }C 2@Override public void failure(IOException e) {2 2// TODO show the error...2 }D });2 }A @Override public void failure(IOException e) { // TODO show the error... }B });
20. Why Reactive? public final class UserActivity extends Activity { private final UserManager um = new UserManager(); @Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.user); TextView tv = (TextView) findViewById(R.id.user_name); tv.setText(um.getUser().toString()); um.setName("Jane Doe", new UserManager.Listener() { @Override public void success() { tv.setText(um.getUser().toString()); }A @Override public void failure(IOException e) { // TODO show the error... }B }); }Y }Z
21. Why Reactive? public final class UserActivity extends Activity { private final UserManager um = new UserManager(); @Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.user); TextView tv = (TextView) findViewById(R.id.user_name); tv.setText(um.getUser().toString()); um.setName("Jane Doe", new UserManager.Listener() { @Override public void success() { tv.setText(um.getUser().toString()); }A @Override public void failure(IOException e) { // TODO show the error... }B }); }Y }Z
22. Why Reactive? public final class UserActivity extends Activity { private final UserManager um = new UserManager(); @Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.user); TextView tv = (TextView) findViewById(R.id.user_name); tv.setText(um.getUser().toString()); um.setName("Jane Doe", new UserManager.Listener() { @Override public void success() { if (isDestroyed()) { tv.setText(um.getUser().toString()); } }A @Override public void failure(IOException e) { // TODO show the error... }B }); }Y }Z
23. Why Reactive? public final class UserActivity extends Activity { private final UserManager um = new UserManager(); @Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.user); TextView tv = (TextView) findViewById(R.id.user_name); tv.setText(um.getUser().toString()); um.setName("Jane Doe", new UserManager.Listener() { @Override public void success() { if (isDestroyed()) { tv.setText(um.getUser().toString()); } }A @Override public void failure(IOException e) { // TODO show the error... }B }); }Y }Z
24. Why Reactive? public final class UserActivity extends Activity { private final UserManager um = new UserManager(); @Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.user); TextView tv = (TextView) findViewById(R.id.user_name); tv.setText(um.getUser().toString()); um.setName("Jane Doe", new UserManager.Listener() { @Override public void success() { if (isDestroyed()) { tv.setText(um.getUser().toString()); }L }A @Override public void failure(IOException e) { // TODO show the error... }B }); }Y }Z
25. Why Reactive? public final class UserActivity extends Activity { private final UserManager um = new UserManager(); @Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.user); TextView tv = (TextView) findViewById(R.id.user_name); tv.setText(um.getUser().toString()); um.setName("Jane Doe", new UserManager.Listener() { @Override public void success() { runOnUiThread(new Runnable() { @Override public void run() { if (isDestroyed()) { tv.setText(um.getUser().toString()); }L }4 }); }A @Override public void failure(IOException e) { // TODO show the error... }B }); }Y }Z
26. Why Reactive? public final class UserActivity extends Activity { private final UserManager um = new UserManager(); @Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.user); TextView tv = (TextView) findViewById(R.id.user_name); tv.setText(um.getUser().toString()); um.setName("Jane Doe", new UserManager.Listener() { @Override public void success() { runOnUiThread(new Runnable() { @Override public void run() { if (isDestroyed()) { tv.setText(um.getUser().toString()); } }4 }); }A @Override public void failure(IOException e) { // TODO show the error... }B }); }Y }Z
27. Why Reactive? GET / 200 OK SELECT * Jane Doe setText onClick
28. Why Reactive? GET / 200 OK SELECT * Jane Doe setText onClick
29. Why Reactive? GET / 200 OK SELECT * Jane Doe setText onClick
30. Why Reactive? GET / 200 OK SELECT * Jane Doe setText onClick
31. Why Reactive? GET / 200 OK SELECT * Jane Doe setText onClick
32. Why Reactive?
33. Why Reactive?
34. Why Reactive? Unless you can model your entire system synchronously, a single asynchronous source breaks imperative programming.
35. Why Reactive?
36. Why Reactive?
37. Why Reactive?
38. Why Reactive?
39. Why Reactive?
40. Why Reactive?
41. RxJava
42. RxJava • A set of classes for representing sources of data.
43. RxJava • A set of classes for representing sources of data. • A set of classes for listening to data sources.
44. RxJava • A set of classes for representing sources of data. • A set of classes for listening to data sources. • A set of methods for modifying and composing the data.
45. RxJava • A set of classes for representing sources of data. • A set of classes for listening to data sources. • A set of methods for modifying and composing the data.
46. Sources
47. Sources • Usually do work when you start or stop listening.
48. Sources • Usually do work when you start or stop listening. • Synchronous or asynchronous.
49. Sources • Usually do work when you start or stop listening. • Synchronous or asynchronous. • Single item or many items.
50. Sources • Usually do work when you start or stop listening. • Synchronous or asynchronous. • Single item, many items, or empty.
51. Sources • Usually do work when you start or stop listening. • Synchronous or asynchronous. • Single item, many items, or empty. • Terminates with an error or succeeds to completion.
52. Sources • Usually do work when you start or stop listening. • Synchronous or asynchronous. • Single item, many items, or empty. • Terminates with an error or succeeds to completion. • May never terminate!
53. Sources • Usually do work when you start or stop listening. • Synchronous or asynchronous. • Single item, many items, or empty. • Terminates with an error or succeeds to completion. • May never terminate! • Just an implementation of the Observer pattern.
54. • Observable<T> • Flowable<T> void void
 Sources
55. • Observable<T> • Emits 0 to n items. • Terminates with complete or error. • Flowable<T> • Emits 0 to n items. • Terminates with complete or error. void void
 Sources
56. • Observable<T> • Emits 0 to n items. • Terminates with complete or error. • Does not have backpressure. • Flowable<T> • Emits 0 to n items. • Terminates with complete or error. • Has backpressure. void void
 Sources
57. Flowable vs. Observable
58. Flowable vs. Observable • Backpressure allows you to control how fast a source emits items.
59. Flowable vs. Observable • Backpressure allows you to control how fast a source emits items. • RxJava 1.x added backpressure late in the design process.
60. Flowable vs. Observable • Backpressure allows you to control how fast a source emits items. • RxJava 1.x added backpressure late in the design process. • All types exposed backpressure but not all sources respected it.
61. Flowable vs. Observable • Backpressure allows you to control how fast a source emits items. • RxJava 1.x added backpressure late in the design process. • All types exposed backpressure but not all sources respected it. • Backpressure, like inheritance, must be designed for.
62. Flowable vs. Observable • Backpressure, like inheritance, must be designed for.
63. Flowable vs. Observable • Backpressure, like inheritance, must be designed for. Observable<MotionEvent> events = RxView.touches(paintView);
64. Flowable vs. Observable • Backpressure, like inheritance, must be designed for. Observable<MotionEvent> events = RxView.touches(paintView); Observable<Row> rows = db.createQuery("SELECT * …");
65. Flowable vs. Observable • Backpressure, like inheritance, must be designed for. Observable<MotionEvent> events = RxView.touches(paintView); Observable<Row> rows = db.createQuery("SELECT * …");
66. Flowable vs. Observable • Backpressure, like inheritance, must be designed for. Observable<MotionEvent> events = RxView.touches(paintView); Observable<Row> rows = db.createQuery("SELECT * …");
67. Flowable vs. Observable • Backpressure, like inheritance, must be designed for. Observable<MotionEvent> events = RxView.touches(paintView); Observable<Row> rows = db.createQuery("SELECT * …");
68. Flowable vs. Observable • Backpressure, like inheritance, must be designed for. Observable<MotionEvent> events = RxView.touches(paintView); Observable<Row> rows = db.createQuery("SELECT * …"); MissingBackpressureException
69. Flowable vs. Observable • Backpressure, like inheritance, must be designed for. Observable<MotionEvent> events = RxView.touches(paintView); Flowable<Row> rows = db.createQuery("SELECT * …");
70. Flowable vs. Observable Observable<MotionEvent> Flowable<Row>
71. Flowable vs. Observable Observable<MotionEvent> Flowable<Row> interface Subscriber<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Subscription s);
 }B interface Observer<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Disposable d);
 }B
72. Flowable vs. Observable Observable<MotionEvent> Flowable<Row> interface Subscriber<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Subscription s);
 }B interface Observer<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Disposable d);
 }B
73. Flowable vs. Observable Observable<MotionEvent> Flowable<Row> interface Subscriber<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Subscription s);
 }B interface Observer<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Disposable d);
 }B
74. Flowable vs. Observable Observable<MotionEvent> Flowable<Row> interface Subscriber<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Subscription s);
 }B interface Observer<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Disposable d);
 }B
75. Flowable vs. Observable Observable<MotionEvent> Flowable<Row> interface Subscriber<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Subscription s);
 }B interface Observer<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Disposable d);
 }B
76. Flowable vs. Observable Observable<MotionEvent> Flowable<Row> interface Subscriber<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Subscription s);
 }B interface Observer<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Disposable d);
 }B interface Disposable {
 void dispose();
 }B
77. Flowable vs. Observable Observable<MotionEvent> Flowable<Row> interface Subscriber<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Subscription s);
 }B interface Observer<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Disposable d);
 }B interface Disposable {
 void dispose();
 }B interface Subscription {
 void cancel();
 void request(long r);
 }B
78. Backpressure No Backpressure 0…n items, complete|error Flowable Observable
79. Reactive Streams ...is an initiative to provide a standard for asynchronous stream processing with non-blocking back pressure.
80. Reactive Streams interface Publisher<T> {
 void subscribe(Subscriber<? super T> s);
 }A
81. Reactive Streams interface Publisher<T> {
 void subscribe(Subscriber<? super T> s);
 }A interface Subscriber<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Subscription s);
 }B
82. Reactive Streams interface Publisher<T> {
 void subscribe(Subscriber<? super T> s);
 }A interface Subscriber<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Subscription s);
 }B interface Subscription {
 void request(long n);
 void cancel();
 }C
83. Reactive Streams interface Publisher<T> {
 void subscribe(Subscriber<? super T> s);
 }A interface Subscriber<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Subscription s);
 }B interface Subscription {
 void request(long n);
 void cancel();
 }C interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
 }D

84. Backpressure No Backpressure 0…n items, complete|error Flowable Observable
85. Reactive Streams (Backpressure) No Backpressure 0…n items, complete|error Flowable Observable
86. interface UserManager { User getUser(); void setName(String name); void setAge(int age);
 }A Sources
87. Sources interface UserManager { Observable<User> getUser(); void setName(String name); void setAge(int age);
 }A
88. Source Specializations
89. Source Specializations • Encoding subsets of Observable into the type system.
90. Single • Either succeeds with an item or errors. • No backpressure support.
91. Single • Either succeeds with an item or errors. • No backpressure support. • Think "reactive scalar".
92. Completable • Either completes or errors. Has no items! • No backpressure support.
93. Completable • Either completes or errors. Has no items! • No backpressure support. • Think "reactive runnable".
94. Maybe • Either succeeds with an item, completes with no items, or errors. • No backpressure support.
95. Maybe • Either succeeds with an item, completes with no items, or errors. • No backpressure support. • Think "reactive optional".
96. Source Specializations • Encoding subsets of Observable into the type system. • Single – Item or error. Think "scalar". • Completable – Complete or error. Think "runnable". • Maybe – Item, complete, or error. Think "optional".
97. Reactive Streams (Backpressure) No Backpressure 0…n items, complete|error Flowable Observable item|complete|error Maybe item|error Single complete|error Completable
98. Sources interface UserManager { Observable<User> getUser(); void setName(String name); void setAge(int age);
 }A
99. Sources interface UserManager { Observable<User> getUser(); Completable setName(String name); Completable setAge(int age);
 }A void void

100. Creating Sources
101. Creating Sources Flowable.just("Hello");
 Flowable.just("Hello", "World"); 
 Observable.just("Hello");
 Observable.just("Hello", "World");
 Maybe.just("Hello");
 Single.just("Hello");
102. Creating Sources String[] array = { "Hello", "World" };
 List<String> list = Arrays.asList(array);
 Flowable.fromArray(array); Flowable.fromIterable(list); Observable.fromArray(array); Observable.fromIterable(list);
103. Creating Sources Observable.fromCallable(new Callable<String>() {
 @Override public String call() {Y
 return getName();
 }X
 });
104. Creating Sources Observable.fromCallable(new Callable<String>() {
 @Override public String call() throws Exception {Y
 return getName();Z
 }X
 });
105. Creating Sources OkHttpClient client = // … Request request = // … Observable.fromCallable(new Callable<String>() {
 @Override public String call() throws Exception {Y
 return client.newCall(request).execute();Z
 }X
 }); 
 
 getName()
106. Creating Sources Flowable.fromCallable(() -> "Hello");
 
 Observable.fromCallable(() -> "Hello");
 
 Maybe.fromCallable(() -> "Hello");
 
 Single.fromCallable(() -> "Hello");
 
 Completable.fromCallable(() -> "Ignored!");
107. Creating Sources Flowable.fromCallable(() -> "Hello");
 
 Observable.fromCallable(() -> "Hello");
 
 Maybe.fromCallable(() -> "Hello");
 Maybe.fromAction(() -> System.out.println("Hello"));
 Maybe.fromRunnable(() -> System.out.println("Hello"))
 
 Single.fromCallable(() -> "Hello");
 
 Completable.fromCallable(() -> "Ignored!");
 Completable.fromAction(() -> System.out.println("Hello"));
 Completable.fromRunnable(() -> System.out.println("Hello"));
108. Creating Sources Observable.create();
109. Creating Sources Observable.create(new ObservableOnSubscribe<String>() {
 @Override
 public void subscribe(ObservableEmitter<String> e) throws Exception {
 e.onNext("Hello");
 e.onComplete();
 }X
 });
110. Creating Sources Observable.create(new ObservableOnSubscribe<String>() {
 @Override
 public void subscribe(ObservableEmitter<String> e) throws Exception {
 e.onNext("Hello");
 e.onComplete();
 }X
 });
111. Creating Sources Observable.create(new ObservableOnSubscribe<String>() {
 @Override
 public void subscribe(ObservableEmitter<String> e) throws Exception {
 e.onNext("Hello");
 e.onComplete();
 }X
 });
112. Creating Sources Observable.create(new ObservableOnSubscribe<String>() {
 @Override
 public void subscribe(ObservableEmitter<String> e) throws Exception {
 e.onNext("Hello");
 e.onComplete();
 }X
 });
113. Creating Sources Observable.create(e -> {
 e.onNext("Hello");
 e.onComplete();
 });
114. Creating Sources Observable.create(e -> {
 e.onNext("Hello"); e.onNext("World");
 e.onComplete();
 });
115. Creating Sources OkHttpClient client = // … Request request = // … Observable.create(e -> { client.newCall(request).enqueue(new Callback() { @Override public void onResponse(Response r) throws IOException { e.onNext(r.body().string()); e.onComplete(); }A @Override public void onFailure(IOException e) { e.onError(e); }B });
 });
116. Creating Sources OkHttpClient client = // … Request request = // … Observable.create(e -> { Call call = client.newCall(request); call.enqueue(new Callback() { @Override public void onResponse(Response r) throws IOException { e.onNext(r.body().string()); e.onComplete(); }A @Override public void onFailure(IOException e) { e.onError(e); }B });
 });
117. Creating Sources OkHttpClient client = // … Request request = // … Observable.create(e -> { Call call = client.newCall(request); e.setCancelation(() -> call.cancel()); call.enqueue(new Callback() { @Override public void onResponse(Response r) throws IOException { e.onNext(r.body().string()); e.onComplete(); }A @Override public void onFailure(IOException e) { e.onError(e); }B });
 });
118. Creating Sources OkHttpClient client = // … Request request = // … Observable.create(e -> { Call call = client.newCall(request); e.setCancelation(() -> call.cancel()); call.enqueue(new Callback() { @Override public void onResponse(Response r) throws IOException { e.onNext(r.body().string()); e.onComplete(); }A @Override public void onFailure(IOException e) { e.onError(e); }B });
 });
119. Creating Sources View view = // … Observable.create(e -> { e.setCancelation(() -> view.setOnClickListener(null)); view.setOnClickListener(v -> e.onNext(v));
 });
120. Creating Sources Flowable.create(e -> { … }); Observable.create(e -> { … }); Maybe.create(e -> { … }); Single.create(e -> { … }); Completable.create(e -> { … });
121. Observing Sources
122. Observing Sources Observable<String> Flowable<String> interface Subscriber<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Subscription s);
 }B interface Observer<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Disposable d);
 }B
123. Observing Sources Observable<String> Flowable<String> interface Subscriber<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Subscription s);
 }B interface Observer<T> {
 void onNext(T t);A
 void onComplete();B
 void onError(Throwable t);C
 void onSubscribe(Disposable d);D
 }B interface Disposable {
 void dispose();
 }B interface Subscription {
 void cancel();
 void request(long r);
 }B
124. Observing Sources Observable<String> o = Observable.just("Hello"); o.subscribe(new Observer<String>() {
 @Override public void onNext(String s) { … }
 @Override public void onComplete() { … }
 @Override public void onError(Throwable t) { … } 
 @Override public void onSubscribe(Disposable d) { ??? }B
 }); Flowable<String> interface Subscriber<T> {
 void onNext(T t);
 void onComplete();
 void onError(Throwable t);
 void onSubscribe(Subscription s);
 }B interface Disposable {
 void dispose();
 }B interface Subscription {
 void cancel();
 void request(long r);
 }B interface T
 T t ;A
 ;B
 ;C
 ;D

125. Observing Sources Observable<String> o = Observable.just("Hello"); o.subscribe(new DisposableObserver<String>() {
 @Override public void onNext(String s) { … }
 @Override public void onComplete() { … }
 @Override public void onError(Throwable t) { … }
 }); 
 
 
 
 @Override public void onSubscribe(Disposable d) { ??? }B

126. Observing Sources Observable<String> o = Observable.just("Hello"); o.subscribe(new DisposableObserver<String>() {
 @Override public void onNext(String s) { … }
 @Override public void onComplete() { … }
 @Override public void onError(Throwable t) { … }
 });Z // TODO how do we dispose???
127. Observing Sources Observable<String> o = Observable.just("Hello"); DisposableObserver observer = new DisposableObserver<String>() {
 @Override public void onNext(String s) { … }
 @Override public void onComplete() { … }
 @Override public void onError(Throwable t) { … }
 } o.subscribe(observer);Z // TODO how do we dispose??????
128. Observing Sources Observable<String> o = Observable.just("Hello"); DisposableObserver observer = new DisposableObserver<String>() {
 @Override public void onNext(String s) { … }
 @Override public void onComplete() { … }
 @Override public void onError(Throwable t) { … }
 } o.subscribe(observer);Z observer.dispose();
129. Observing Sources Observable<String> o = Observable.just("Hello"); o.subscribe(new DisposableObserver<String>() {
 @Override public void onNext(String s) { … }
 @Override public void onComplete() { … }
 @Override public void onError(Throwable t) { … }
 });Z
130. Observing Sources Observable<String> o = Observable.just("Hello"); o.subscribeWith(new DisposableObserver<String>() {
 @Override public void onNext(String s) { … }
 @Override public void onComplete() { … }
 @Override public void onError(Throwable t) { … }
 });Z
131. Observing Sources Observable<String> o = Observable.just("Hello"); Disposable d = o.subscribeWith(new DisposableObserver<String>() {
 @Override public void onNext(String s) { … }
 @Override public void onComplete() { … }
 @Override public void onError(Throwable t) { … }
 });Z d.dispose();
132. Observing Sources Observable<String> o = Observable.just("Hello"); CompositeDisposable disposables = new CompositeDisposable(); disposables.add(o.subscribeWith(new DisposableObserver<String>() {
 @Override public void onNext(String s) { … }
 @Override public void onComplete() { … }
 @Override public void onError(Throwable t) { … }
 }));Z disposables.dispose(); 
 
 
 
 d.dispose(); Disposable d =
133. Observing Sources Observable<String> o = Observable.just("Hello"); o.subscribeWith(new DisposableObserver<String>() { … });Z Maybe<String> m = Maybe.just("Hello"); m.subscribeWith(new DisposableMaybeObserver<String>() { … });Z Single<String> s = Single.just("Hello"); s.subscribeWith(new DisposableSingleObserver<String>() { … });Z Completable c = Completable.completed(); c.subscribeWith(new DisposableCompletableObserver<String>() { … });Z disposables.add(
 @Override public void onNext(String s) { … }
 @Override public void onComplete() { … }
 @Override public void onError(Throwable t) { … }
 ) disposables.dispose();
134. Observing Sources Flowable<String> f = Flowable.just("Hello"); f.subscribeWith(new DisposableSubscriber<String>() { … }); Observable<String> o = Observable.just("Hello"); o.subscribeWith(new DisposableObserver<String>() { … }); Maybe<String> m = Maybe.just("Hello"); m.subscribeWith(new DisposableMaybeObserver<String>() { … }); Single<String> s = Single.just("Hello"); s.subscribeWith(new DisposableSingleObserver<String>() { … }); Completable c = Completable.completed(); c.subscribeWith(new DisposableCompletableObserver<String>() { … });
135. Observing Sources Flowable<String> f = Flowable.just("Hello"); f.subscribeWith(new DisposableSubscriber<String>() { … }); Observable<String> o = Observable.just("Hello"); o.subscribeWith(new DisposableObserver<String>() { … }); Maybe<String> m = Maybe.just("Hello"); m.subscribeWith(new DisposableMaybeObserver<String>() { … }); Single<String> s = Single.just("Hello"); s.subscribeWith(new DisposableSingleObserver<String>() { … }); Completable c = Completable.completed(); c.subscribeWith(new DisposableCompletableObserver<String>() { … });
136. Observing Sources Flowable<String> f = Flowable.just("Hello"); Disposable d1 = f.subscribeWith(new DisposableSubscriber<String>() { … }); Observable<String> o = Observable.just("Hello"); Disposable d2 = o.subscribeWith(new DisposableObserver<String>() { … }); Maybe<String> m = Maybe.just("Hello"); Disposable d3 = m.subscribeWith(new DisposableMaybeObserver<String>() { … }); Single<String> s = Single.just("Hello"); Disposable d4 = s.subscribeWith(new DisposableSingleObserver<String>() { … }); Completable c = Completable.completed(); Disposable d5 = c.subscribeWith(new DisposableCompletableObserver<String>() { … });
137. RxJava • A set of classes for representing sources of data. • A set of classes for listening to data sources. • A set of methods for modifying and composing data.
138. RxJava • A set of classes for representing sources of data. • A set of classes for listening to data sources. • A set of methods for modifying and composing data.
139. Operators
140. Operators • Manipulate or combine data in some way.
141. Operators • Manipulate or combine data in some way. • Manipulate threading in some way.
142. Operators • Manipulate or combine data in some way. • Manipulate threading in some way. • Manipulate emissions in some way.
143. Operators String greeting = "Hello";
144. Operators String greeting = "Hello";Z String yelling = greeting.toUppercase();
145. Operators Observable<String> greeting = Observable.just("Hello");Z String yelling = greeting.toUppercase();Y
146. Operators Observable<String> greeting = Observable.just("Hello"); Observable<String> yelling = greeting.map(s -> s.toUppercase());Y
147. Operators Observable<String> greeting = Observable.just("Hello"); Observable<String> yelling = greeting.map(s -> s.toUppercase());Y
148. Operators @Override public void success() { runOnUiThread(new Runnable() { @Override public void run() { tv.setText(um.getUser().toString()); }4 }); }A
149. Operators Observable<User> user = um.getUser();
150. Operators Observable<User> user = um.getUser(); Observable<User> mainThreadUser = user.observeOn(AndroidSchedulers.mainThread());
151. Operators Observable<User> user = um.getUser(); Observable<User> mainThreadUser = user.observeOn(AndroidSchedulers.mainThread());
152. Operators OkHttpClient client = // … Request request = // … Response response = client.newCall(request).execute();
153. Operators OkHttpClient client = // … Request request = // … Observable<Response> response = Observable.fromCallable(() -> { return client.newCall(request).execute(); });
154. Operators OkHttpClient client = // … Request request = // … Observable<Response> response = Observable.fromCallable(() -> { return client.newCall(request).execute(); }); Observable<Response> backgroundResponse = response.subscribeOn(Schedulers.io());
155. Operators OkHttpClient client = // … Request request = // … Observable<Response> response = Observable.fromCallable(() -> { return client.newCall(request).execute(); }); Observable<Response> backgroundResponse = response.subscribeOn(Schedulers.io());
156. Operators OkHttpClient client = // … Request request = // … Observable<Response> response = Observable.fromCallable(() -> { return client.newCall(request).execute(); }); Observable<Response> backgroundResponse = response.subscribeOn(Schedulers.io());
157. Operators OkHttpClient client = // … Request request = // … Observable<Response> response = Observable.fromCallable(() -> { return client.newCall(request).execute(); }) .subscribeOn(Schedulers.io());Y
158. Operators OkHttpClient client = // … Request request = // … Observable<Response> response = Observable.fromCallable(() -> { return client.newCall(request).execute(); }) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread());Y
159. Operators OkHttpClient client = // … Request request = // … Observable<Response> response = Observable.fromCallable(() -> { return client.newCall(request).execute(); }) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .map(response -> response.body().string());Y
160. Operators OkHttpClient client = // … Request request = // … Observable<Response> response = Observable.fromCallable(() -> { return client.newCall(request).execute(); }) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .map(response -> response.body().string());Y// NetworkOnMainThread!
161. Operators OkHttpClient client = // … Request request = // … Observable<Response> response = Observable.fromCallable(() -> { return client.newCall(request).execute(); }) .subscribeOn(Schedulers.io()) .map(response -> response.body().string()) // Ok! .observeOn(AndroidSchedulers.mainThread());Y// NetworkOnMainThread!
162. Operator Specialization
163. Operator Specialization Observable
164. Operator Specialization first() Observable
165. Operator Specialization first() Observable Observable
166. Operator Specialization 1 2 3 4 5 6 7 8 9 10 11 first() 1 Observable Observable
167. Operator Specialization 1 2 3 4 5 6 7 8 9 10 11 get(0) 1 List<Long> List<Long>
168. Operator Specialization first() Observable Observable
169. Operator Specialization first() Observable Single
170. Operator Specialization 1 2 3 4 5 6 7 8 9 10 11 first() 1 Observable Single
171. Operator Specialization first() Observable Single
172. Operator Specialization first() Observable Single NoSuchElementException
173. Operator Specialization first() Observable
174. Operator Specialization firstElement() Observable
175. Operator Specialization firstElement() Observable Maybe
176. Operator Specialization firstElement() Observable Maybe
177. Operator Specialization ignoreElements() Observable
178. Operator Specialization ignoreElements() Observable Completable
179. Operator Specialization 1 2 3 4 5 6 7 8 9 ignoreElements() Observable Completable
180. Operator Specialization ignoreElements() Flowable Completable
181. Operator Specialization firstElement() Flowable Maybe
182. Operator Specialization first() Flowable Single
183. Flowable Observable Maybe Single Completable Flowable toObservable() reduce() elementAt() firstElement() lastElement() singleElement() scan() elementAt() first()/firstOrError() last()/lastOrError() single/singleOrError() all()/any()/count() (and more) ignoreElements() Observable toFlowable() reduce() elementAt() firstElement() lastElement() singleElement() scan() elementAt() first()/firstOrError() last()/lastOrError() single/singleOrError() all()/any()/count() (and more) ignoreElements() Maybe toFlowable() toObservable() toSingle() sequenceEqual() toCompletable() Single toFlowable() toObservable() toMaybe() toCompletable() Completable toFlowable() toObservable() toMaybe() toSingle() toSingleDefault() From To
184. Flowable Observable Maybe Single Completable Flowable toObservable() reduce() elementAt() firstElement() lastElement() singleElement() scan() elementAt() first()/firstOrError() last()/lastOrError() single/singleOrError() all()/any()/count() (and more) ignoreElements() Observable toFlowable() reduce() elementAt() firstElement() lastElement() singleElement() scan() elementAt() first()/firstOrError() last()/lastOrError() single/singleOrError() all()/any()/count() (and more) ignoreElements() Maybe toFlowable() toObservable() toSingle() sequenceEqual() toCompletable() Single toFlowable() toObservable() toMaybe() toCompletable() Completable toFlowable() toObservable() toMaybe() toSingle() toSingleDefault() From To
185. Flowable Observable Maybe Single Completable Flowable toObservable() reduce() elementAt() firstElement() lastElement() singleElement() scan() elementAt() first()/firstOrError() last()/lastOrError() single/singleOrError() all()/any()/count() (and more) ignoreElements() Observable toFlowable() reduce() elementAt() firstElement() lastElement() singleElement() scan() elementAt() first()/firstOrError() last()/lastOrError() single/singleOrError() all()/any()/count() (and more) ignoreElements() Maybe toFlowable() toObservable() toSingle() sequenceEqual() toCompletable() Single toFlowable() toObservable() toMaybe() toCompletable() Completable toFlowable() toObservable() toMaybe() toSingle() toSingleDefault() From To
186. Being Reactive
187. Being Reactive um.getUser()
188. Being Reactive um.getUser() .observeOn(AndroidSchedulers.mainThread())
189. Being Reactive um.getUser() .observeOn(AndroidSchedulers.mainThread()) .subscribeWith(new DisposableObserver<User>() { @Override public void onNext(User user) { }1 @Override public void onComplete() { /* ignored */ } @Override public void onError(Throwable t) { /* crash or show */ } });
190. Being Reactive um.getUser() .observeOn(AndroidSchedulers.mainThread()) .subscribeWith(new DisposableObserver<User>() { @Override public void onNext(User user) { tv.setText(user.toString()); }1 @Override public void onComplete() { /* ignored */ } @Override public void onError(Throwable t) { /* crash or show */ } });2
191. Being Reactive disposables.add(um.getUser() .observeOn(AndroidSchedulers.mainThread()) .subscribeWith(new DisposableObserver<User>() { @Override public void onNext(User user) { tv.setText(user.toString()); }1 @Override public void onComplete() { /* ignored */ } @Override public void onError(Throwable t) { /* crash or show */ } }));2
192. Being Reactive // onCreate disposables.add(um.getUser() .observeOn(AndroidSchedulers.mainThread()) .subscribeWith(new DisposableObserver<User>() { @Override public void onNext(User user) { tv.setText(user.toString()); }1 @Override public void onComplete() { /* ignored */ } @Override public void onError(Throwable t) { /* crash or show */ } })); // onDestroy disposables.dispose();
193. Being Reactive um.setName("Jane Doe")
194. Being Reactive um.setName("Jane Doe") .subscribeOn(Schedulers.io())
195. Being Reactive um.setName("Jane Doe") .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread())
196. Being Reactive um.setName("Jane Doe") .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribeWith(new DisposableCompletableObserver() { @Override public void onComplete() { }1 @Override public void onError(Throwable t) { // retry or show }2 });
197. Being Reactive um.setName("Jane Doe") .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribeWith(new DisposableCompletableObserver() { @Override public void onComplete() { // success! re-enable editing }1 @Override public void onError(Throwable t) { // retry or show }2 });3
198. Being Reactive disposables.add(um.setName("Jane Doe") .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribeWith(new DisposableCompletableObserver() { @Override public void onComplete() { // success! re-enable editing }1 @Override public void onError(Throwable t) { // retry or show }2 }));3
199. Architecture
200. Architecture • Less allocation to create a stream. • Less overhead when subscribing to a stream.
201. RxJava 2
202. RxJava 2 • Build an app that reacts properly instead of copes desperately!
203. RxJava 2 • Build an app that reacts properly instead of copes desperately! • Final release scheduled for 2016-10-29.
204. Living with 1.x and 2.x class RxJavaInterop {
 static <T> Flowable<T> toV2Flowable(rx.Observable<T> o) { … }
 static <T> Observable<T> toV2Observable(rx.Observable<T> o) { … } static <T> Maybe<T> toV2Maybe(rx.Single<T> s) { … } static <T> Maybe<T> toV2Maybe(rx.Completable c) { … }
 static <T> Single<T> toV2Single(rx.Single<T> s) { … }
 static Completable toV2Completable(rx.Completable c) { … } 
 static <T> rx.Observable<T> toV1Observable(Publisher<T> p) { … } static <T> rx.Observable<T> toV1Observable(Observable<T> o, …) { … }
 static <T> rx.Single<T> toV1Single(Single<T> o) { … } static <T> rx.Single<T> toV1Single(Maybe<T> m) { … }
 static rx.Completable toV1Completable(Completable c) { … } static rx.Completable toV1Completable(Maybe<T> m) { … }
 }X
205. Living with 1.x and 2.x class RxJavaInterop {
 static <T> Flowable<T> toV2Flowable(rx.Observable<T> o) { … }
 static <T> Observable<T> toV2Observable(rx.Observable<T> o) { … } static <T> Maybe<T> toV2Maybe(rx.Single<T> s) { … } static <T> Maybe<T> toV2Maybe(rx.Completable c) { … }
 static <T> Single<T> toV2Single(rx.Single<T> s) { … }
 static Completable toV2Completable(rx.Completable c) { … } 
 static <T> rx.Observable<T> toV1Observable(Publisher<T> p) { … } static <T> rx.Observable<T> toV1Observable(Observable<T> o, …) { … }
 static <T> rx.Single<T> toV1Single(Single<T> o) { … } github.com/akarnokd/RxJava2Interop
206. dependencies {
 compile 'io.reactivex.rxjava2:rxjava:2.0.0-RC3'
 compile 'io.reactivex.rxjava2:rxandroid:2.0.0-RC1' // Optionally...
 compile 'com.github.akarnokd:rxjava2-interop:0.3.0' }
207. jakewharton jakewharton jakewharton twitter.com/ google.com/+ .com Exploring RxJava