Rx Java

Intro to Rx Java and example how it's used in the QuizUp on Android

Transcript

1. RxJava Siggi Jónsson - siggi@q.is
2. • RxJava is a Java VM implementation of Reactive Extensions: a library for composing asynchronous and event-based programs by using observable sequences. What is RxJava?
3. • Java: RxJava • JavaScript: RxJS • C#: Rx.NET • C#(Unity): UniRx • Scala: RxScala • Clojure: RxClojure • C++: RxCpp Platforms • Ruby: Rx.rb • Python: RxPY • Groovy: RxGroovy • JRuby: RxJRuby • Kotlin: RxKotlin • Swift: RxSwift • ObjC: ReactiveCocoa*
4. • Java: RxJava • JavaScript: RxJS • C#: Rx.NET • C#(Unity): UniRx • Scala: RxScala • Clojure: RxClojure • C++: RxCpp Platforms • Ruby: Rx.rb • Python: RxPY • Groovy: RxGroovy • JRuby: RxJRuby • Kotlin: RxKotlin • Swift: RxSwift • ObjC: ReactiveCocoa*
5. • An awesome tool to work with asynchronous streams of data What is RxJava?
6. • Button clicks • Text edits • Network requests • Push notifications • …. What are streams?
7. What is not a stream?
8. The Past Life before RxJava
9. Asynchronous Operations • AsyncTasks • Callbacks • Handlers • Threads • Runnables • Executors • EventBus
10. AsyncTask public class SomeTask extends AsyncTask<String, Void, SomeResult> {
 private WeakReference<Callback> callback; 
 public SomeTask(Callback callback) {
 callback = new WeakReference<Callback>(callback);
 }
 
 @Override
 protected SomeResult doInBackground(String... params) { try { //do stuff } catch (Exception e) { //oh god, what now? return null?
 }
 return null;
 }
 
 @Override
 protected void onPostExecute(SomeResult result) { if (callback.get() != null) { callback.onResult(result); }
 }
 }
11. Callback Hell button.setOnClickListener(new View.OnClickListener() {
 public void onClick(View v) {
 loadPlayer(playerId, new Callback() {
 public void playerLoaded(Player p1) {
 loadPlayer(p1.bestFriendId, new Callback() {
 public void playerLoaded(Player p2) {
 loadImage(p2.profileUrl, new ImageCallback() {
 public void imageLoaded(Bitmap bitmap) {
 imageView.setImageBitmap(bitmap);
 }
 });
 }
 });
 }
 });
 }
 });
12. • Did the player reach the Top 10 in a topic in the first month after he joined or the first month after the topic was created (which ever is first)? Complexity grows fast
13. RxJava to the rescue
14. • Observable -> produces events • Observer/Subscriber -> receives events • Operators • Schedulers -> Multithreading RxJava Concepts
15. Observable Observable.create(new Observable.OnSubscribe<String>() {
 @Override
 public void call(Subscriber<? super String> subscriber) {
 try {
 subscriber.onNext("Hello, World!");
 subscriber.onCompleted();
 }
 catch (Exception e) {
 subscriber.onError(e);
 }
 }
 });
16. Observable Observable.create(new Observable.OnSubscribe<String>() {
 @Override
 public void call(Subscriber<? super String> subscriber) {
 try {
 subscriber.onNext("Hello, World!");
 subscriber.onCompleted();
 }
 catch (Exception e) {
 subscriber.onError(e);
 }
 }
 });
17. Observable Observable.just("Hello, World!");
18. Observer interface Observer<T> { void onNext(T t); 
 void onCompleted();
 void onError(Throwable e);
 }
19. Putting it together Observable.just("Hello, World").subscribe(new Observer<String>() {
 @Override
 public void onCompleted() {
 System.out.println("Done!");
 }
 
 @Override
 public void onError(Throwable e) {
 System.err.println(e.getMessage());
 }
 
 @Override
 public void onNext(String s) {
 System.out.println(s);
 }
 });
20. Putting it together Observable.just("Hello, World").subscribe(new Observer<String>() {
 @Override
 public void onCompleted() {
 System.out.println("Done!");
 }
 
 @Override
 public void onError(Throwable e) {
 System.err.println(e.getMessage());
 }
 
 @Override
 public void onNext(String s) {
 System.out.println(s);
 }
 });
21. Putting it together Observable.just("Hello, World").subscribe(new Observer<String>() {
 @Override
 public void onCompleted() {
 System.out.println("Done!");
 }
 
 @Override
 public void onError(Throwable e) {
 System.err.println(e.getMessage());
 }
 
 @Override
 public void onNext(String s) {
 System.out.println(s);
 }
 });
22. Putting it together Observable.just("Hello, World").subscribe(new Observer<String>() {
 @Override
 public void onCompleted() {
 System.out.println("Done!");
 }
 
 @Override
 public void onError(Throwable e) {
 System.err.println(e.getMessage());
 }
 
 @Override
 public void onNext(String s) {
 System.out.println(s);
 }
 });
23. Putting it together Observable.just("Hello, World").subscribe(new Observer<String>() {
 @Override
 public void onCompleted() {
 System.out.println("Done!");
 }
 
 @Override
 public void onError(Throwable e) {
 System.err.println(e.getMessage());
 }
 
 @Override
 public void onNext(String s) {
 System.out.println(s);
 }
 });
24. Putting it together Observable.just("Hello, World").subscribe(new Observer<String>() {
 @Override
 public void onCompleted() {
 System.out.println("Done!");
 }
 
 @Override
 public void onError(Throwable e) { System.err.println(e.getMessage());
 }
 
 @Override
 public void onNext(String s) {
 System.out.println(s);
 }
 });
25. Putting it together api.getHello().subscribe(new Observer<String>() {
 @Override
 public void onCompleted() {
 System.out.println("Done!");
 }
 
 @Override
 public void onError(Throwable e) { System.err.println(e.getMessage());
 }
 
 @Override
 public void onNext(String s) {
 System.out.println(s);
 }
 });
26. Putting it together api.getHello().subscribe(new Observer<String>() {
 @Override
 public void onCompleted() {
 System.out.println("Done!");
 }
 
 @Override
 public void onError(Throwable e) { System.err.println(e.getMessage());
 }
 
 @Override
 public void onNext(String s) {
 System.out.println(s);
 }
 });
27. Operators The Toolbox
35. • Did the player reach the Top 10 in a topic in the first month after he joined or the first month after the topic was created (which ever is first)? Complexity grows fast
36. Easy Observable.zip(
 playerService.getPlayer(“<player id>”),
 topicsService.getTopic("<topic id>”), player, topic -> {
 if (player.created.before(topic.created))
 return player.created;
 else return topic.created;
 }
 ).flatMap(date -> {
 return topicsService.getLeaderboard("<topic id>”, date);
 }).flatMap(leaderboard -> 
 return Observable.from(leaderboard.top);
 }).exists(player -> {
 return player.id.equals("<player_id>"); 
 });
37. Easy
38. Schedulers RxAndroid
39. Schedulers
40. Schedulers • Skipping Frames. The application may be doing too much work on its main thread. • CalledFromWrongThreadException: Only the original thread that created a view hierarchy can touch its views. • NetworkOnMainThreadException • :(
41. Schedulers playerService.getFollowers() .subscribe(response -> {
 adapter.addAll(response.players);
 });
42. Schedulers playerService.getFollowers() .observeOn(AndroidSchedulers.mainThread()) .subscribe(response -> {
 adapter.addAll(response.players);
 });
43. Schedulers playerService.getFollowers() .observeOn(AndroidSchedulers.mainThread()) .subscribe(response -> {
 adapter.addAll(response.players);
 });
44. QuizUp Gameplay
45. QuizUp Round Observable<MatchEvent> events = Observable.concat(
 roundStarted(round),
 showQuestion(round, question).delay(ROUND_START_DELAY),
 showAnswers(round, question).delay(showAnswersDelay),
 answerPeriodStart(round).delay(ANSWER_PERIOD_START_DELAY),
 Observable.merge(
 playerAnswer.takeUntil(timer),
 opponentAnswer.takeUntil(timer),
 timer.takeUntil(bothPlayersHaveAnswered)
 ),
 answerPeriodEnd(round),
 playerDidNotAnswer(round, question, playerDidAnswer),
 verifyOpponentIsStillHere(opponentDidAnswer),
 showCorrectAnswer(round, question).delay(SHOW_CORRECT_ANSWER_DELAY),
 Observable.empty().delaySubscription(FINISH_UP_ROUND_DELAY)
 );
46. QuizUp Round Observable<MatchEvent> events = Observable.concat(
 roundStarted(round),
 showQuestion(round, question).delay(ROUND_START_DELAY),
 showAnswers(round, question).delay(showAnswersDelay),
 answerPeriodStart(round).delay(ANSWER_PERIOD_START_DELAY),
 Observable.merge(
 playerAnswer.takeUntil(timer),
 opponentAnswer.takeUntil(timer),
 timer.takeUntil(bothPlayersHaveAnswered)
 ),
 answerPeriodEnd(round),
 playerDidNotAnswer(round, question, playerDidAnswer),
 verifyOpponentIsStillHere(opponentDidAnswer),
 showCorrectAnswer(round, question).delay(SHOW_CORRECT_ANSWER_DELAY),
 Observable.empty().delaySubscription(FINISH_UP_ROUND_DELAY)
 );
47. QuizUp Round Observable<MatchEvent> events = Observable.concat(
 roundStarted(round),
 showQuestion(round, question).delay(ROUND_START_DELAY),
 showAnswers(round, question).delay(showAnswersDelay),
 answerPeriodStart(round).delay(ANSWER_PERIOD_START_DELAY),
 Observable.merge(
 playerAnswer.takeUntil(timer),
 opponentAnswer.takeUntil(timer),
 timer.takeUntil(bothPlayersHaveAnswered)
 ),
 answerPeriodEnd(round),
 playerDidNotAnswer(round, question, playerDidAnswer),
 verifyOpponentIsStillHere(opponentDidAnswer),
 showCorrectAnswer(round, question).delay(SHOW_CORRECT_ANSWER_DELAY),
 Observable.empty().delaySubscription(FINISH_UP_ROUND_DELAY)
 );
48. QuizUp Round Observable<MatchEvent> events = Observable.concat(
 roundStarted(round),
 showQuestion(round, question).delay(ROUND_START_DELAY),
 showAnswers(round, question).delay(showAnswersDelay),
 answerPeriodStart(round).delay(ANSWER_PERIOD_START_DELAY),
 Observable.merge(
 playerAnswer.takeUntil(timer),
 opponentAnswer.takeUntil(timer),
 timer.takeUntil(bothPlayersHaveAnswered)
 ),
 answerPeriodEnd(round),
 playerDidNotAnswer(round, question, playerDidAnswer),
 verifyOpponentIsStillHere(opponentDidAnswer),
 showCorrectAnswer(round, question).delay(SHOW_CORRECT_ANSWER_DELAY),
 Observable.empty().delaySubscription(FINISH_UP_ROUND_DELAY)
 );
49. QuizUp Round Observable<MatchEvent> events = Observable.concat(
 roundStarted(round),
 showQuestion(round, question).delay(ROUND_START_DELAY),
 showAnswers(round, question).delay(showAnswersDelay),
 answerPeriodStart(round).delay(ANSWER_PERIOD_START_DELAY),
 Observable.merge(
 playerAnswer.takeUntil(timer),
 opponentAnswer.takeUntil(timer),
 timer.takeUntil(bothPlayersHaveAnswered)
 ),
 answerPeriodEnd(round),
 playerDidNotAnswer(round, question, playerDidAnswer),
 verifyOpponentIsStillHere(opponentDidAnswer),
 showCorrectAnswer(round, question).delay(SHOW_CORRECT_ANSWER_DELAY),
 Observable.empty().delaySubscription(FINISH_UP_ROUND_DELAY)
 );
50. QuizUp Round Observable<MatchEvent> events = Observable.concat(
 roundStarted(round),
 showQuestion(round, question).delay(ROUND_START_DELAY),
 showAnswers(round, question).delay(showAnswersDelay),
 answerPeriodStart(round).delay(ANSWER_PERIOD_START_DELAY),
 Observable.merge(
 playerAnswer.takeUntil(timer),
 opponentAnswer.takeUntil(timer),
 timer.takeUntil(bothPlayersHaveAnswered)
 ),
 answerPeriodEnd(round),
 playerDidNotAnswer(round, question, playerDidAnswer),
 verifyOpponentIsStillHere(opponentDidAnswer),
 showCorrectAnswer(round, question).delay(SHOW_CORRECT_ANSWER_DELAY),
 Observable.empty().delaySubscription(FINISH_UP_ROUND_DELAY)
 );
51. QuizUp Round Observable<MatchEvent> events = Observable.concat(
 roundStarted(round),
 showQuestion(round, question).delay(ROUND_START_DELAY),
 showAnswers(round, question).delay(showAnswersDelay),
 answerPeriodStart(round).delay(ANSWER_PERIOD_START_DELAY),
 Observable.merge(
 playerAnswer.takeUntil(timer),
 opponentAnswer.takeUntil(timer),
 timer.takeUntil(bothPlayersHaveAnswered)
 ),
 answerPeriodEnd(round),
 playerDidNotAnswer(round, question, playerDidAnswer),
 verifyOpponentIsStillHere(opponentDidAnswer),
 showCorrectAnswer(round, question).delay(SHOW_CORRECT_ANSWER_DELAY),
 Observable.empty().delaySubscription(FINISH_UP_ROUND_DELAY)
 );
52. QuizUp Round Observable<MatchEvent> events = Observable.concat(
 roundStarted(round),
 showQuestion(round, question).delay(ROUND_START_DELAY),
 showAnswers(round, question).delay(showAnswersDelay),
 answerPeriodStart(round).delay(ANSWER_PERIOD_START_DELAY),
 Observable.merge(
 playerAnswer.takeUntil(timer),
 opponentAnswer.takeUntil(timer),
 timer.takeUntil(bothPlayersHaveAnswered)
 ),
 answerPeriodEnd(round),
 playerDidNotAnswer(round, question, playerDidAnswer),
 verifyOpponentIsStillHere(opponentDidAnswer),
 showCorrectAnswer(round, question).delay(SHOW_CORRECT_ANSWER_DELAY),
 Observable.empty().delaySubscription(FINISH_UP_ROUND_DELAY)
 );
53. QuizUp Round Observable<MatchEvent> events = Observable.concat(
 roundStarted(round),
 showQuestion(round, question).delay(ROUND_START_DELAY),
 showAnswers(round, question).delay(showAnswersDelay),
 answerPeriodStart(round).delay(ANSWER_PERIOD_START_DELAY),
 Observable.merge(
 playerAnswer.takeUntil(timer),
 opponentAnswer.takeUntil(timer),
 timer.takeUntil(bothPlayersHaveAnswered)
 ),
 answerPeriodEnd(round),
 playerDidNotAnswer(round, question, playerDidAnswer),
 verifyOpponentIsStillHere(opponentDidAnswer),
 showCorrectAnswer(round, question).delay(SHOW_CORRECT_ANSWER_DELAY),
 Observable.empty().delaySubscription(FINISH_UP_ROUND_DELAY)
 );
54. QuizUp Round Observable<MatchEvent> events = Observable.concat(
 roundStarted(round),
 showQuestion(round, question).delay(ROUND_START_DELAY),
 showAnswers(round, question).delay(showAnswersDelay),
 answerPeriodStart(round).delay(ANSWER_PERIOD_START_DELAY),
 Observable.merge(
 playerAnswer.takeUntil(timer),
 opponentAnswer.takeUntil(timer),
 timer.takeUntil(bothPlayersHaveAnswered)
 ),
 answerPeriodEnd(round),
 playerDidNotAnswer(round, question, playerDidAnswer),
 verifyOpponentIsStillHere(opponentDidAnswer),
 showCorrectAnswer(round, question).delay(SHOW_CORRECT_ANSWER_DELAY),
 Observable.empty().delaySubscription(FINISH_UP_ROUND_DELAY)
 );
55. QuizUp Round Observable<MatchEvent> events = Observable.concat(
 roundStarted(round),
 showQuestion(round, question).delay(ROUND_START_DELAY),
 showAnswers(round, question).delay(showAnswersDelay),
 answerPeriodStart(round).delay(ANSWER_PERIOD_START_DELAY),
 Observable.merge(
 playerAnswer.takeUntil(timer),
 opponentAnswer.takeUntil(timer),
 timer.takeUntil(bothPlayersHaveAnswered)
 ),
 answerPeriodEnd(round),
 playerDidNotAnswer(round, question, playerDidAnswer),
 verifyOpponentIsStillHere(opponentDidAnswer),
 showCorrectAnswer(round, question).delay(SHOW_CORRECT_ANSWER_DELAY),
 Observable.empty().delaySubscription(FINISH_UP_ROUND_DELAY)
 );
56. QuizUp Round
57. QuizUp Round Observable<MatchEvent> events = Observable.concat(
 roundStarted(round),
 showQuestion(round, question).delay(ROUND_START_DELAY),
 showAnswers(round, question).delay(showAnswersDelay),
 answerPeriodStart(round).delay(ANSWER_PERIOD_START_DELAY),
 Observable.merge(
 playerAnswer.takeUntil(timer),
 opponentAnswer.takeUntil(timer),
 timer.takeUntil(bothPlayersHaveAnswered)
 ),
 answerPeriodEnd(round),
 playerDidNotAnswer(round, question, playerDidAnswer),
 verifyOpponentIsStillHere(opponentDidAnswer),
 showCorrectAnswer(round, question).delay(SHOW_CORRECT_ANSWER_DELAY),
 Observable.empty().delaySubscription(FINISH_UP_ROUND_DELAY)
 );
58. All Rounds 
 Observable<MatchEvent> match = Observable.range(1, 7).concatMap(round -> {
 return round(round);
 });

59. All Rounds 
 Observable<MatchEvent> match = Observable.range(1, 7).concatMap(round -> {
 return round(round);
 });

60. All Rounds 
 Observable<MatchEvent> match = Observable.range(1, 7).concatMap(round -> {
 return round(round);
 });

61. All Rounds
62. Thank You q.is/jobs