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
290
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
110
The 65K Dex Limit
berglind
0
79
The Android QuizUp Saga
berglind
1
180
Other Decks in Technology
See All in Technology
Platform Engineering for Software Developers and Architects
syntasso
1
520
テストコード品質を高めるためにMutation Testingライブラリ・Strykerを実戦導入してみた話
ysknsid25
7
2.7k
障害対応指揮の意思決定と情報共有における価値観 / Waroom Meetup #2
arthur1
5
490
ノーコードデータ分析ツールで体験する時系列データ分析超入門
negi111111
0
420
DynamoDB でスロットリングが発生したとき_大盛りver/when_throttling_occurs_in_dynamodb_long
emiki
1
430
DynamoDB でスロットリングが発生したとき/when_throttling_occurs_in_dynamodb_short
emiki
0
260
FlutterアプリにおけるSLI/SLOを用いたユーザー体験の可視化と計測基盤構築
ostk0069
0
100
The Rise of LLMOps
asei
8
1.7k
Zennのパフォーマンスモニタリングでやっていること
ryosukeigarashi
0
160
10XにおけるData Contractの導入について: Data Contract事例共有会
10xinc
6
660
OS 標準のデザインシステムを超えて - より柔軟な Flutter テーマ管理 | FlutterKaigi 2024
ronnnnn
0
240
いざ、BSC討伐の旅
nikinusu
2
780
Featured
See All Featured
Why Our Code Smells
bkeepers
PRO
334
57k
Bootstrapping a Software Product
garrettdimon
PRO
305
110k
Why You Should Never Use an ORM
jnunemaker
PRO
54
9.1k
Responsive Adventures: Dirty Tricks From The Dark Corners of Front-End
smashingmag
250
21k
Music & Morning Musume
bryan
46
6.2k
Ruby is Unlike a Banana
tanoku
97
11k
Templates, Plugins, & Blocks: Oh My! Creating the theme that thinks of everything
marktimemedia
26
2.1k
Art, The Web, and Tiny UX
lynnandtonic
297
20k
ピンチをチャンスに:未来をつくるプロダクトロードマップ #pmconf2020
aki_iinuma
109
49k
Build The Right Thing And Hit Your Dates
maggiecrowley
33
2.4k
I Don’t Have Time: Getting Over the Fear to Launch Your Podcast
jcasabona
28
2k
JavaScript: Past, Present, and Future - NDC Porto 2020
reverentgeek
47
5k
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