Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Rx Java

berglind
October 11, 2015

Rx Java

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

berglind

October 11, 2015
Tweet

More Decks by berglind

Other Decks in Technology

Transcript

  1. • 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?
  2. • 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*
  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. • Button clicks • Text edits • Network requests •

    Push notifications • …. What are streams?
  5. 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); }
 }
 }
  6. 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);
 }
 });
 }
 });
 }
 });
 }
 });
  7. • 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
  8. • Observable -> produces events • Observer/Subscriber -> receives events

    • Operators • Schedulers -> Multithreading RxJava Concepts
  9. 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);
 }
 }
 });
  10. 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);
 }
 }
 });
  11. 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);
 }
 });
  12. 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);
 }
 });
  13. 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);
 }
 });
  14. 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);
 }
 });
  15. 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);
 }
 });
  16. 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);
 }
 });
  17. 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);
 }
 });
  18. 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);
 }
 });
  19. • 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
  20. 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>"); 
 });
  21. 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 • :(
  22. 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)
 );
  23. 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)
 );
  24. 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)
 );
  25. 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)
 );
  26. 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)
 );
  27. 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)
 );
  28. 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)
 );
  29. 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)
 );
  30. 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)
 );
  31. 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)
 );
  32. 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)
 );
  33. 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)
 );