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

Going Reactive - An Architectural Journey

Going Reactive - An Architectural Journey

Initially held at GOTO Copenhagen 2015
http://gotocon.com/cph-2015/presentation/Going%20Reactive%20-%20An%20Android%20architectural%20journey

Ever since its inception in 2008, the Android platform has remained challenging to work with when building large, event driven applications in a consistent and reliable way. We've been among the earliest companies to embrace FRP as a vehicle to tackle some of the complexity arising as a direct result of that. In this talk I will give an overview of what our Android application architecture has evolved into, and how the use of RxJava for dealing with asynchronous streams of events has impacted its design, from data sources, to business logic and presentation

Matthias Käppler

October 06, 2015
Tweet

More Decks by Matthias Käppler

Other Decks in Programming

Transcript

  1. commit 24c61b35754ff5ca153ce37c5886279153f0d16f Author: Matthias Kaeppler <[email protected]> Date: Wed Mar 13

    16:09:04 2013 +0100 Throw RxJava into the mix! diff --git a/app/pom.xml b/app/pom.xml index 86ba988..1bf5109 100644 --- a/app/pom.xml +++ b/app/pom.xml @@ -178,6 +178,11 @@ + <dependency> + <groupId>com.netflix.rxjava</groupId> + <artifactId>rxjava-core</artifactId> + <version>0.5.4</version> + </dependency> </dependencies>
  2. Layer Objects Rx Screen Presenter
 Life-cycle dispatch,
 view binding Feature

    Operations
 Business logic, data 
 wiring, scheduling Storage & API
 Network, database,
 flat files, syncer Rx[Android]
  3. Presenters class SoundStreamPresenter extends RecyclerViewPresenter<StreamItem> { ...
 @Override protected CollectionBinding<StreamItem>

    onBuildBinding(Bundle args) { return CollectionBinding.from(
 streamOperations.initialStreamItems()) .withAdapter(adapter) .withPager(streamOperations.pagingFunction()) .build(); } }
  4. Paging p1 p2 p3 *current *next p1 onNext(p1) subscribe()
 /

    next() PublishSubject switchOnNext PagingFunction Pager
  5. Use Cases class SoundStreamOperations { Observable<List<StreamItem>> initialStreamItems() { return loadFirstPageOfStream()

    .zipWith(
 facebookInvites.loadWithPictures(),
 prependFacebookInvites()) .subscribeOn(scheduler); } ... }
  6. Feature Data class SoundStreamStorage { Observable<PropertySet> streamItems(int limit) { Query

    query = Query.from(“SoundStreamTable”).limit(limit); return propellerRx.query(query).map(new StreamItemMapper()); 
 } ... }
  7. Screen-to-Screen Updates Observable<PropertySet> toggleLike(Urn urn, 
 boolean addLike) { return

    storeLikeCommand.toObservable(urn, addLike) .map(toChangeSet(targetUrn, addLike)) .doOnNext(publishChangeSet); }
  8. Screen-to-Screen Updates @Override public void call(PropertySet changeSet) { eventBus.publish(
 EventQueue.ENTITY_STATE_CHANGED,

    
 EntityStateChangedEvent.fromLike(changeSet)
 ); } publishChangeSet: Action1<PropertySet> RxSubject in disguise!
  9. Life-Cycle Subscriptions private CompositeSubscription viewLifeCycle;
 
 protected void onViewCreated(...) {


    viewLifeCycle = new CompositeSubscription(); viewLifeCycle.add(...); ...
 }
 
 protected void onDestroyView() {
 viewLifeCycle.unsubscribe();
 }
  10. Observable Transformers Observable<Model> scheduledModel() { return Observable.create(...).compose(schedulingStrategy) } class HighPrioUiTask<T>

    extends Transformer<T, T> { public Observable<T> call(Observable<T> source) { return source
 .subscribeOn(Schedulers.HIGH_PRIO)
 .observeOn(AndroidSchedulers.mainThread()) } }
  11. Observable<Integer> intSequence() { return Observable.create((subscriber) -> { List<Integer> ints =

    computeListOfInts();
 for (int n : ints) {
 subscriber.onNext(n);
 subscriber.onCompleted(); } } Deferred Execution Observable<Integer> intSequence() { return Observable.defer(() -> { return Observable.from(computeListOfInts()); } } expensive!
  12. ObserveOn: Backpressure public void onStart() { request(RxRingBuffer.SIZE); } public void

    onNext(final T t) { ... if (!queue.offer(on.next(t))) { onError(new MissingBackpressureException()); return; } schedule(); } 16!
  13. ObserveOn: Backpressure ★ Take load off of target thread˝ ★

    Use buffering operators (buffer, toList, …)˝ ★ Use onBackpressure* operators˝ ★ System.setProperty(“rx.ring-buffer.size”)
  14. Gandalf ★ Annotation based byte code injection
 ★ Based on

    Hugo 
 https://github.com/JakeWharton/hugo
 ★ AspectJ + Gradle plugin
 ★ @RxLogObservable, @RxLogSubscriber
  15. Gandalf @RxLogObservable Observable<String> createObservable() {
 return Observable.just(1, 2, 3) .map((n)

    -> {return Integer.toString(n);} .observeOn(mainThread()); } @RxLogSubscriber class StringSubscriber extends Subscriber<String> {}
  16. Gandalf [@Observable :: @InClass -> MainActivity :: @Method 
 ->

    createObservable()] [@Observable#createObservable -> onSubscribe() :: 
 @SubscribeOn -> main] [@Observable#createObservable -> onNext() -> 1] [@Observable#createObservable -> onNext() -> 2] [@Observable#createObservable -> onNext() -> 3] [@Observable#createObservable -> onCompleted()] [@Observable#createObservable -> onTerminate() :: 
 @Emitted -> 3 elements :: @Time -> 4 ms] [@Observable#createObservable -> onUnsubscribe()]