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

The Mayans Lost Guide to RxJava on Android

The Mayans Lost Guide to RxJava on Android

In this little journey, we are gonna see what RxJava offers to us in terms of android development. We will also learn how we can apply it to real life mobile projects by showing different examples and use cases.

Fernando Cejas

April 25, 2015
Tweet

More Decks by Fernando Cejas

Other Decks in Programming

Transcript

  1. #Reactive 1- of, relating to, or marked by reaction or

    reactance. 2- readily responsive to a stimulus.
  2. RxJava is a Java VM implementation of Reactive X (Reactive

    Extensions): ˝ a library for composing asynchronous and event-based programs by using observable sequences. What is RxJava?
  3. Multithreading is always complex˝ Concurrency˝ Java Futures are Expensive to

    Compose˝ Java Futures˝ Callbacks Have Their Own Problems˝ Callbacks˝ @benjchristensen from @netflix
  4. public class FuturesA { public static void run() throws Exception

    { ExecutorService executor = new ThreadPoolExecutor(4, 4, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>()); Future<String> f1 = executor.submit(new CallToRemoteServiceA()); Future<String> f2 = executor.submit(new CallToRemoteServiceB()); System.out.println(f1.get() + " - " + f2.get()); } } https://gist.github.com/benjchristensen/4670979 #Java Futures
  5. #Java Futures https://gist.github.com/benjchristensen/4671081 public static void run() throws Exception {

    ExecutorService executor = new ThreadPoolExecutor(4, 4, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>()); try { // get f3 with dependent result from f1 Future<String> f1 = executor.submit(new CallToRemoteServiceA()); Future<String> f3 = executor.submit(new CallToRemoteServiceC(f1.get())); /* The work below can not proceed until f1.get() completes even though there is no dependency */ // also get f4/f5 after dependency f2 completes Future<Integer> f2 = executor.submit(new CallToRemoteServiceB()); Future<Integer> f4 = executor.submit(new CallToRemoteServiceD(f2.get())); Future<Integer> f5 = executor.submit(new CallToRemoteServiceE(f2.get())); System.out.println(f3.get() + " => " + (f4.get() * f5.get())); } finally { executor.shutdownNow(); } }
  6. #Java Futures https://gist.github.com/benjchristensen/4671081 public static void run4() throws Exception {

    ExecutorService executor = new ThreadPoolExecutor(4, 4, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>()); try { List<Future<?>> futures = new ArrayList<Future<?>>(); // kick off several async tasks futures.add(executor.submit(new CallToRemoteServiceA())); futures.add(executor.submit(new CallToRemoteServiceB())); futures.add(executor.submit(new CallToRemoteServiceC("A"))); futures.add(executor.submit(new CallToRemoteServiceC("B"))); futures.add(executor.submit(new CallToRemoteServiceD(1))); futures.add(executor.submit(new CallToRemoteServiceE(2))); futures.add(executor.submit(new CallToRemoteServiceE(3))); // as each completes do further work for (Future<?> f : futures) { /* this blocks so even if other futures in the list complete earlier they will wait until this one is done */ doMoreWork(f.get()); } } finally { executor.shutdownNow(); } }
  7. Multithreading is always complex˝ Concurrency˝ Java Futures are Expensive to

    Compose˝ Java Futures˝ Callbacks Have Their Own Problems˝ Callbacks˝ @benjchristensen from @netflix
  8. #Callbacks https://gist.github.com/benjchristensen/4677544 ... // get f3 with dependent result from

    f1 executor.execute(new CallToRemoteServiceA(new Callback<String>() { @Override public void call(String f1) { executor.execute(new CallToRemoteServiceC(new Callback<String>() { @Override public void call(String f3) { // we have f1 and f3 now need to compose with others System.out.println("intermediate callback: " + f3 + " => " + ("f4 * f5")); // set to thread-safe variable accessible by external scope f3Value.set(f3); latch.countDown(); } }, f1)); } })); ...
  9. Observables The Observable object is who does the job. Represents

    an object that sends notifications (Provider) to a Subscriptor (Observer).
  10. Observables Add 2 missing semantics to the Observer pattern: #1:

    Emits a signal to the consumer when there is no more data available. #2: Emits a signal to the consumer when an error has occurred.
  11. Subscribers Subscribers provides a mechanism for receiving push- based notifications

    from Observables, and permits manual unsubscribing from these Observables.
  12. Subscribers Not an observer pattern: Observables often don't start emitting

    items until someone explicitly subscribes to them.
  13. Subscribers public class DefaultSubscriber<T> extends rx.Subscriber<T> { @Override public void

    onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(T t) { } }
  14. Subscriptions Subscriptions represents the link between your Observable and your

    Subscriber. #1: Subscriptions #2: CompositeSubscriptions
  15. #1: Schedulers.io() #2: Schedulers.computation() #3: Schedulers.from() Schedulers If you want

    to introduce multithreading into your cascade of Observable operators, you can do so by instructing those operators (or particular Observables) to operate on particular Schedulers.
  16. Operators Operators can be used in between the source Observable

    and the ultimate Subscriber to manipulate emitted items. You can even write your own custom operators.
  17. flatMap() Transforms the items emitted by an Observable into Observables,

    then flatten the emissions from those into a single Observable (no order)
  18. concatMap() Transforms the items emitted by an Observable into Observables,

    then flatten the emissions from those into a single Observable (keeps order)
  19. filter() Emits the same item it received, but only if

    it passes the boolean check (predicate).
  20. onError() is called if an exception is thrown at any

    time. Error handling The operators do not have to handle the exception.
  21. onErrorResumeNext() Instructs an Observable to emit a sequence of items

    if it encounters an error. onErrorReturn() Instructs an Observable to emit a particular item when it encounters an error. onExceptionResumeNext() Instructs an Observable to continue emitting items after it encounters an exception. retry() If a source Observable emits an error, resubscribe to it in the hopes that it will complete without error. retryWhen() If a source Observable emits an error, pass that error to another Observable to determine whether to resubscribe to the source. Error handling Operators
  22. #1: Observable and Subscriber can do anything #2: The Observable

    and Subscriber are independent of the transformational steps in between them. #3: Operators let you do anything to the stream of data. Key ideas behind RxJava http://blog.danlew.net/2014/09/15/grokking-rxjava-part-1/
  23. #1: Learning curve #2: Too many anonymous classes generated (OutOfMemory?)

    #3: Verbosity (retrolambda to the rescue?) But there are some pitfalls…
  24. @PerActivity public class UserListPresenter extends DefaultSubscriber<List<User>> implements Presenter { private

    UserListView viewListView; private final UseCase getUserListUseCase; @Inject public UserListPresenter(@Named("userList") UseCase getUserListUserCase, UserModelDataMapper userModelDataMapper) { this.getUserListUseCase = getUserListUserCase; } @Override public void destroy() { this.getUserListUseCase.unsubscribe(); } private void getUserList() { this.getUserListUseCase.execute(this); } ... } #Example: Reactive Presenter
  25. #Example: Reactive Presenter @PerActivity public class UserListPresenter extends DefaultSubscriber<List<User>> implements

    Presenter { ... @Override public void onCompleted() { this.hideViewLoading(); } @Override public void onError(Throwable e) { this.hideViewLoading(); this.showErrorMessage(new DefaultErrorBundle((Exception) e)); this.showViewRetry(); } @Override public void onNext(List<User> users) { this.showUsersCollectionInView(users); } }
  26. #Example: UseCase public abstract class UseCase { private final ThreadExecutor

    threadExecutor; private final PostExecutionThread postExecutionThread; private Subscription subscription = Subscriptions.empty(); protected UseCase(ThreadExecutor threadExecutor, PostExecutionThread postExecutionThread) { this.threadExecutor = threadExecutor; this.postExecutionThread = postExecutionThread; } protected abstract Observable buildUseCaseObservable(); public void execute(Subscriber UseCaseSubscriber) { this.subscription = this.buildUseCaseObservable() .subscribeOn(Schedulers.from(threadExecutor)) .observeOn(postExecutionThread.getScheduler()) .subscribe(UseCaseSubscriber); } public void unsubscribe() { if (!subscription.isUnsubscribed()) subscription.unsubscribe(); } }
  27. #Example: Execution Thread /** * MainThread (UI Thread) implementation based

    on a * {@link rx.Scheduler} which will execute actions on * the Android UI thread */ @Singleton public class UIThread implements PostExecutionThread { @Inject public UIThread() {} @Override public Scheduler getScheduler() { return AndroidSchedulers.mainThread(); } }
  28. #Example: UseCase /** * This class is an implementation of

    {@link UseCase} that represents a * use case for retrieving a collection of all {@link User}. */ public class GetUserListUseCase extends UseCase { private final UserRepository userRepository; @Inject public GetUserListUseCase(UserRepository userRepository, ThreadExecutor threadExecutor, PostExecutionThread postExecutionThread) { super(threadExecutor, postExecutionThread); this.userRepository = userRepository; } @Override public Observable buildUseCaseObservable() { return this.userRepository.getUsers(); } }
  29. #Example: Operator and Action public class CloudUserDataStore implements UserDataStore {

    private final RestApi restApi; private final UserCache userCache; private final Action1<UserEntity> saveToCacheAction = new Action1<UserEntity>() { @Override public void call(UserEntity userEntity) { if (userEntity != null) { CloudUserDataStore.this.userCache.put(userEntity); } } }; public CloudUserDataStore(RestApi restApi, UserCache userCache) { this.restApi = restApi; this.userCache = userCache; } @Override public Observable<List<UserEntity>> getUserEntityList() { return this.restApi.getUserEntityList(); } @Override public Observable<UserEntity> getUserEntityDetails(final int userId) { return this.restApi.getUserEntityById(userId).doOnNext(saveToCacheAction); } }
  30. #Example: Data transformation @Singleton public class UserDataRepository implements UserRepository {

    private final UserDataStoreFactory userDataStoreFactory; private final UserEntityDataMapper userEntityDataMapper; private final Func1<List<UserEntity>, List<User>> userListEntityMapper = new Func1<List<UserEntity>, List<User>>() { @Override public List<User> call(List<UserEntity> userEntities) { return UserDataRepository.this.userEntityDataMapper.transform(userEntities); } }; private final Func1<UserEntity, User> userDetailsEntityMapper = new Func1<UserEntity, User>() { @Override public User call(UserEntity userEntity) { return UserDataRepository.this.userEntityDataMapper.transform(userEntity); } }; @Override public Observable<List<User>> getUsers() { final UserDataStore userDataStore = this.userDataStoreFactory.createCloudDataStore(); return userDataStore.getUserEntityList().map(userListEntityMapper); } }
  31. #Example: Observable creation @Override public Observable<List<UserEntity>> getUserEntityList() { return Observable.create(new

    Observable.OnSubscribe<List<UserEntity>>() { @Override public void call(Subscriber<? super List<UserEntity>> subscriber) { if (isThereInternetConnection()) { try { String responseUserEntities = getUserEntitiesFromApi(); subscriber.onNext(userEntityJsonMapper.transformUserEntityCollection( responseUserEntities)); subscriber.onCompleted(); } catch (Exception e) { subscriber.onError(new NetworkConnectionException(e.getCause())); } } else { subscriber.onError(new NetworkConnectionException()); } } }); }
  32. #1: Good starting point to switch to Rx Observables. #2:

    No need to deal with threading an synchronization. #3: Very simple to wrap an http connection in an Observable How do I start with RxJava? Rx at data level
  33. #1: We can convert our events into Rx Observables How

    do I start with RxJava? Rx at view level Observable input = Observable.FromEventPattern(textView, "TextChanged") .Select(_ => textbox.Text) .Throttle(TimeSpan.FromSeconds(0.5)) .DistinctUntilChanged();
  34. #1: You will return Rx Observables in domain layer. #2:

    Be careful with side effects (Rx Schedulers other than UI Thread) How do I start with RxJava? Rx at domain level
  35. #1: By default, RxJava is synchronous. #2: onSubscribe() is executed

    separately for every new subscriber. #3: Subscriptions leak memory. Tips and Tricks #4: Read the official documentation
  36. References Reactive Programming on Android With RxJava https://mttkay.github.io/blog/2013/08/25/functional-reactive-programming-on-android-with-rxjava/ Grokking RxJava

    http://blog.danlew.net/2014/09/15/grokking-rxjava-part-1/ Reactive Programming in the Netflix API with RxJava http://techblog.netflix.com/2013/02/rxjava-netflix-api.html Rx for .NET and RxJava for Android http://futurice.com/blog/tech-pick-of-the-week-rx-for-net-and-rxjava-for-android https://github.com/android10/Android-CleanArchitecture Official Documentation https://github.com/ReactiveX/RxJava/wiki https://github.com/android10/Android-ReactiveProgramming