Upgrade to Pro
— share decks privately, control downloads, hide ads and more …
Speaker Deck
Features
Speaker Deck
PRO
Sign in
Sign up for free
Search
Search
Rx Java
Search
berglind
October 11, 2015
Technology
1
300
Rx Java
Intro to Rx Java and example how it's used in the QuizUp on Android
berglind
October 11, 2015
Tweet
Share
More Decks by berglind
See All by berglind
Cleaner Code
berglind
0
120
The 65K Dex Limit
berglind
0
87
The Android QuizUp Saga
berglind
1
210
Other Decks in Technology
See All in Technology
サラリーマンの小遣いで作るtoCサービス - Cloudflare Workersでスケールする開発戦略
shinaps
2
470
エンジニアが主導できる組織づくり ー 製品と事業を進化させる体制へのシフト
ueokande
1
110
slog.Handlerのよくある実装ミス
sakiengineer
4
480
なぜテストマネージャの視点が 必要なのか? 〜 一歩先へ進むために 〜
moritamasami
0
240
LLMを搭載したプロダクトの品質保証の模索と学び
qa
1
1.1k
Autonomous Database - Dedicated 技術詳細 / adb-d_technical_detail_jp
oracle4engineer
PRO
4
10k
プラットフォーム転換期におけるGitHub Copilot活用〜Coding agentがそれを加速するか〜 / Leveraging GitHub Copilot During Platform Transition Periods
aeonpeople
1
240
はじめてのOSS開発からみえたGo言語の強み
shibukazu
4
1k
テストを軸にした生き残り術
kworkdev
PRO
0
220
Terraformで構築する セルフサービス型データプラットフォーム / terraform-self-service-data-platform
pei0804
1
200
Aurora DSQLはサーバーレスアーキテクチャの常識を変えるのか
iwatatomoya
1
1.2k
「全員プロダクトマネージャー」を実現する、Cursorによる仕様検討の自動運転
applism118
22
12k
Featured
See All Featured
Distributed Sagas: A Protocol for Coordinating Microservices
caitiem20
333
22k
Practical Tips for Bootstrapping Information Extraction Pipelines
honnibal
PRO
23
1.4k
GraphQLとの向き合い方2022年版
quramy
49
14k
XXLCSS - How to scale CSS and keep your sanity
sugarenia
248
1.3M
Faster Mobile Websites
deanohume
309
31k
The Web Performance Landscape in 2024 [PerfNow 2024]
tammyeverts
9
810
Reflections from 52 weeks, 52 projects
jeffersonlam
352
21k
Building a Modern Day E-commerce SEO Strategy
aleyda
43
7.6k
Context Engineering - Making Every Token Count
addyosmani
3
62
No one is an island. Learnings from fostering a developers community.
thoeni
21
3.4k
Into the Great Unknown - MozCon
thekraken
40
2k
Measuring & Analyzing Core Web Vitals
bluesmoon
9
580
Transcript
RxJava Siggi Jónsson -
[email protected]
• 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?
• 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*
• 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*
• An awesome tool to work with asynchronous streams of
data What is RxJava?
• Button clicks • Text edits • Network requests •
Push notifications • …. What are streams?
What is not a stream?
The Past Life before RxJava
Asynchronous Operations • AsyncTasks • Callbacks • Handlers • Threads
• Runnables • Executors • EventBus
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); } } }
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); } }); } }); } }); } });
• 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
RxJava to the rescue
• Observable -> produces events • Observer/Subscriber -> receives events
• Operators • Schedulers -> Multithreading RxJava Concepts
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); } } });
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); } } });
Observable Observable.just("Hello, World!");
Observer interface Observer<T> { void onNext(T t); void onCompleted();
void onError(Throwable e); }
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); } });
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); } });
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); } });
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); } });
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); } });
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); } });
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); } });
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); } });
Operators The Toolbox
None
None
None
None
None
None
None
• 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
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>"); });
Easy
Schedulers RxAndroid
Schedulers
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 • :(
Schedulers playerService.getFollowers() .subscribe(response -> { adapter.addAll(response.players); });
Schedulers playerService.getFollowers() .observeOn(AndroidSchedulers.mainThread()) .subscribe(response -> { adapter.addAll(response.players); });
Schedulers playerService.getFollowers() .observeOn(AndroidSchedulers.mainThread()) .subscribe(response -> { adapter.addAll(response.players); });
QuizUp Gameplay
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) );
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) );
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) );
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) );
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) );
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) );
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) );
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) );
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) );
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) );
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) );
QuizUp Round
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) );
All Rounds Observable<MatchEvent> match = Observable.range(1, 7).concatMap(round -> {
return round(round); });
All Rounds Observable<MatchEvent> match = Observable.range(1, 7).concatMap(round -> {
return round(round); });
All Rounds Observable<MatchEvent> match = Observable.range(1, 7).concatMap(round -> {
return round(round); });
All Rounds
Thank You q.is/jobs