Rx

Slides from Android Tech Talks #10 meetup at @techspacekrk

1. RX
2. chalup.github.io ! " + @jchalupski +Jerzy Chalupski
3. RX?
4. public interface GitHubService {
 @GET("/users")
 List<User> getUsers();
 
 @GET("/users/{user}/repos")
 List<Repo> listRepos(@Path("user") User user);
 }
 
 public class User {
 String getLogin();
 }
 
 public class Repo {
 String getDescription(); User getOwner();
 }
5. GitHubService service = /* retrofit magic */
 
 for (User user : service.getUsers()) {
 for (Repo repo : service.listRepos(user)) {
 if (repo.getDescription().contains("android")) {
 /* send invitation email */
 break;
 }
 }
 }
6. public interface GitHubService {
 @GET("/users")
 List<User> getUsers();
 
 @GET("/users/{user}")
 User getUserDetails(@Path("user") User user);
 
 @GET("/users/{user}/repos")
 List<Repo> listRepos(@Path("user") User user);
 }
 
 public class User {
 String getLogin();
 String getLocation();
 }
 
 public class Repo {
 String getName();
 String getDescription();
 String getLanguage();
 boolean isFork(); User getOwner();
 }
7. GitHubService service = /* retrofit magic */
 
 for (User user : service.getUsers()) {
 for (Repo repo : service.listRepos(user)) {
 boolean isAndroidRepo =
 repo.getDescription().contains("android") ||
 repo.getName().contains("android");
 boolean writtenInJava = repo.getLanguage().equals("Java");
 if (isAndroidRepo && writtenInJava && !repo.isFork()) {
 if (service.getDetails(user).getLocation().equals("Kraków")) { /* send invitation email */
 }
 break;
 }
 }
 }
8. public interface GitHubService {
 @GET("/users")
 List<User> getUsers();
 
 @GET("/users/{user}")
 User getUserDetails(@Path("user") User user);
 
 @GET("/users/{user}/repos")
 List<Repo> listRepos(@Path("user") User user);
 }
 
 public class User {
 String getLogin();
 String getLocation();
 }
 
 public class Repo {
 String getName();
 String getDescription();
 String getLanguage();
 boolean isFork(); User getOwner();
 }
9. public interface GitHubService {
 @GET("/users")
 void getUsers(Callback<List<User>> users);
 
 @GET("/users/{user}")
 void getUserDetails(@Path("user") User user, Callback<User> …);
 
 @GET("/users/{user}/repos")
 void listRepos(@Path("user") User user, Callback<List<Repo>> …); }
 
 public class User {
 String getLogin();
 String getLocation();
 }
 
 public class Repo {
 String getName();
 String getDescription();
 String getLanguage();
 boolean isFork(); User getOwner();
 }
10. GitHubService service = /* retrofit magic */
 
 service.getUsers(new Callback<List<User>>() {
 @Override
 public void success(List<User> users, Response response) {
 for (User user : users) {
 service.listRepos(user, new Callback<List<Repo>>() {
 @Override
 public void success(List<Repo> repos, Response response) {
 for (Repo repo : repos) {
 boolean isAndroidRepo = repo.getDescription().contains("android") || repo.getName().contains("android");
 boolean writtenInJava = repo.getLanguage().equals("Java");
 if (isAndroidRepo && writtenInJava && !repo.isFork()) {
 service.getUserDetails(user, new Callback<User>() {
 @Override
 public void success(User user, Response response) {
 if (user.getLocation().equals("Kraków")) {
 /* send invitation email */
 }
 }
 
 @Override
 public void failure(RetrofitError error) {
 }
 });
 break;
 }
 }
 }
 
 @Override
 public void failure(RetrofitError error) {
 }
 });
 }
 }
 
 @Override
 public void failure(RetrofitError error) {
 }
 });

11. GitHubService service = /* retrofit magic */
 
 service.getUsers(new Callback<List<User>>() {
 @Override
 public void success(List<User> users, Response response) {
 for (User user : users) {
 service.listRepos(user, new Callback<List<Repo>>() {
 @Override
 public void success(List<Repo> repos, Response response) {
 for (Repo repo : repos) {
 boolean isAndroidRepo = repo.getDescription().contains("android") || repo.getName().contains("android");
 boolean writtenInJava = repo.getLanguage().equals("Java");
 if (isAndroidRepo && writtenInJava && !repo.isFork()) {
 service.getUserDetails(user, new Callback<User>() {
 @Override
 public void success(User user, Response response) {
 if (user.getLocation().equals("Kraków")) {
 /* send invitation email */
 }
 }
 
 @Override
 public void failure(RetrofitError error) {
 }
 });
 break;
 }
 }
 }
 
 @Override
 public void failure(RetrofitError error) {
 }
 });
 }
 }
 
 @Override
 public void failure(RetrofitError error) {
 }
 });

12. FP 101
14. filter( )
15. filter( )
16. filter( ) map( )
17. filter( ) map( )
18. filter( ) map( ) flatMap( )
19. filter( ) map( ) flatMap( )
20. filter map flatMap Java 8 Stream API
21. filter transform transformAndConcat Guava FluentIterable
22. select map flat_map Ruby Enumerable
23. Where Select SelectMany C# LINQ
24. GitHubService service = /* retrofit magic */
 
 for (User user : service.getUsers()) {
 for (Repo repo : service.listRepos(user)) {
 boolean isAndroidRepo =
 repo.getDescription().contains("android") ||
 repo.getName().contains("android");
 boolean writtenInJava = repo.getLanguage().equals("Java");
 if (isAndroidRepo && writtenInJava && !repo.isFork()) {
 if (service.getDetails(user).getLocation().equals("Kraków")) { /* send invitation email */
 }
 break;
 }
 }
 }
25. GitHubService service = /* retrofit magic */
 
 service.getUsers().stream()
 .flatMap(user -> service.listRepos(user).stream())
 .filter(repo -> repo.getName().contains(“android") ||
 repo.getDescription().contains("android"))
 .filter(repo -> repo.getLanguage().equals("Java"))
 .filter(repo -> !repo.isFork())
 .map(Repo::getOwner)
 .distinct()
 .map(service::getUserDetails)
 .filter(user -> user.getLocation().equals("Kraków"))
 .forEach(user -> { /* send invitation email */ });
26. RX 101
27. TAKEAWAY #1 Observable<T> == Iterable<Event>
39. public interface GitHubService {
 @GET("/users")
 void getUsers(Callback<List<User>> users);
 }
 service.getUsers(new Callback<List<User>>() {
 @Override
 public void success(List<User> users, Response response) {
 }
 
 @Override
 public void failure(RetrofitError error) {
 }
 });

40. public interface GitHubService {
 @GET(“/users") Observable<List<User>> getUsers(); } 
 service.getUsers().subscribe(
 new Action1<List<User>>() {
 @Override
 public void call(List<User> users) { // onNext
 }
 },
 new Action1<Throwable>() {
 @Override
 public void call(Throwable throwable) { // onError
 }
 }
 );
41. filter map flatMap RxJava Operators
42. filter map flatMap RxJava Operators http://reactivex.io/documentation/operators.html
43. GitHubService service = /* retrofit magic */
 
 service.getUsers().stream()
 .flatMap(user -> service.listRepos(user).stream())
 .filter(repo -> repo.getName().contains(“android") ||
 repo.getDescription().contains("android"))
 .filter(repo -> repo.getLanguage().equals("Java"))
 .filter(repo -> !repo.isFork())
 .map(Repo::getOwner)
 .distinct()
 .map(service::getUserDetails)
 .filter(user -> user.getLocation().equals("Kraków"))
 .forEach(user -> { /* send invitation email */ });
44. public interface GitHubService {
 @GET("/users")
 List<User> getUsers();
 
 @GET("/users/{user}")
 User getUserDetails(@Path("user") User user);
 
 @GET("/users/{user}/repos")
 List<Repo> listRepos(@Path("user") User user);
 }
 
 public class User {
 String getLogin();
 String getLocation();
 }
 
 public class Repo {
 String getName();
 String getDescription();
 String getLanguage();
 boolean isFork(); User getOwner();
 }
45. public interface GitHubService {
 @GET("/users")
 Observable<List<User>> getUsers();
 
 @GET("/users/{user}")
 Observable<User> getUserDetails(@Path("user") User user);
 
 @GET("/users/{user}/repos")
 Observable<List<Repo>> listRepos(@Path("user") User user); }
 
 public class User {
 String getLogin();
 String getLocation();
 }
 
 public class Repo {
 String getName();
 String getDescription();
 String getLanguage();
 boolean isFork(); User getOwner();
 }
46. GitHubService service = /* retrofit magic */
 
 service.getUsers()
 .flatMap(Observable::from)
 .flatMap(service::listRepos)
 .flatMap(Observable::from)
 .filter(repo -> repo.getName().contains("android") ||
 repo.getDescription().contains("android"))
 .filter(repo -> repo.getLanguage().equals("Java"))
 .filter(repo -> !repo.isFork())
 .map(Repo::getOwner)
 .distinct()
 .flatMap(service::getUserDetails)
 .filter(user -> user.getLocation().equals("Kraków"))
 .subscribe(user -> { /* send invitation email */ });
47. GitHubService service = /* retrofit magic */
 
 service.getUsers().stream()
 .flatMap(user -> service.listRepos(user).stream()) 
 .filter(repo -> repo.getName().contains(“android") ||
 repo.getDescription().contains("android"))
 .filter(repo -> repo.getLanguage().equals("Java"))
 .filter(repo -> !repo.isFork())
 .map(Repo::getOwner)
 .distinct()
 .map(service::getUserDetails)
 .filter(user -> user.getLocation().equals("Kraków"))
 .forEach(user -> { /* send invitation email */ });
48. GitHubService service = /* retrofit magic */
 
 service.getUsers()
 .flatMap(Observable::from)
 .flatMap(service::listRepos)
 .flatMap(Observable::from)
 .filter(repo -> repo.getName().contains("android") ||
 repo.getDescription().contains("android"))
 .filter(repo -> repo.getLanguage().equals("Java"))
 .filter(repo -> !repo.isFork())
 .map(Repo::getOwner)
 .distinct()
 .flatMap(service::getUserDetails)
 .filter(user -> user.getLocation().equals("Kraków"))
 .subscribe(user -> { /* send invitation email */ });
49. GitHubService service = /* retrofit magic */
 
 service.getUsers()
 .flatMap(Observable::from)
 .flatMap(service::listRepos)
 .flatMap(Observable::from)
 .filter(repo -> repo.getName().contains("android") ||
 repo.getDescription().contains("android"))
 .filter(repo -> repo.getLanguage().equals("Java"))
 .filter(repo -> !repo.isFork())
 .map(Repo::getOwner)
 .distinct()
 .flatMap(service::getUserDetails)
 .filter(user -> user.getLocation().equals("Kraków"))
 .subscribe(user -> { /* send invitation email */ },
 error -> { /* report to analytics */ }
 );
50. GitHubService service = /* retrofit magic */
 
 service.getUsers(new Callback<List<User>>() {
 @Override
 public void success(List<User> users, Response response) {
 for (User user : users) {
 service.listRepos(user, new Callback<List<Repo>>() {
 @Override
 public void success(List<Repo> repos, Response response) {
 for (Repo repo : repos) {
 boolean isAndroidRepo = repo.getDescription().contains("android") || repo.getName().contains("android");
 boolean writtenInJava = repo.getLanguage().equals("Java");
 if (isAndroidRepo && writtenInJava && !repo.isFork()) {
 service.getUserDetails(user, new Callback<User>() {
 @Override
 public void success(User user, Response response) {
 if (user.getLocation().equals("Kraków")) {
 /* send invitation email */
 }
 }
 
 @Override
 public void failure(RetrofitError error) {
 }
 });
 break;
 }
 }
 }
 
 @Override
 public void failure(RetrofitError error) {
 }
 });
 }
 }
 
 @Override
 public void failure(RetrofitError error) {
 }
 });

51. RX!
52. RX! db access user input touch events validation web requests
54. GitHubService service = /* retrofit magic */
 
 service.getUsers()
 .flatMap(Observable::from)
 .flatMap(service::listRepos)
 .flatMap(Observable::from)
 .filter(repo -> repo.getName().contains("android") ||
 repo.getDescription().contains("android"))
 .filter(repo -> repo.getLanguage().equals("Java"))
 .filter(repo -> !repo.isFork())
 .map(Repo::getOwner)
 .distinct()
 .flatMap(service::getUserDetails)
 .filter(user -> user.getLocation().equals("Kraków"))
 .subscribe(user -> { /* send invitation email */ },
 error -> { /* report to analytics */ }
 );
55. GitHubService service = /* retrofit magic */
 
 service.getUsers()
 .flatMap(new Func1<List<User>, Observable<? extends User>>() {
 @Override
 public Observable<? extends User> call(List<User> users) {
 return Observable.from(users);
 }
 })
 .flatMap(new Func1<User, Observable<? extends List<Repo>>>() {
 @Override
 public Observable<? extends List<Repo>> call(User user) {
 return service.listRepos(user);
 }
 })
 .flatMap(new Func1<List<Repo>, Observable<? extends Repo>>() {
 @Override
 public Observable<? extends Repo> call(List<Repo> repos) {
 return Observable.from(repos);
 }
 })
 .filter(new Func1<Repo, Boolean>() {
 @Override
 public Boolean call(Repo repo) {
 return repo.getName().contains("android") ||
 repo.getDescription().contains("android");
 }
 })
 .filter(new Func1<Repo, Boolean>() {
 @Override
 public Boolean call(Repo repo) {
 return repo.getLanguage().equals("Java");
 }
 })
 .filter(new Func1<Repo, Boolean>() {
 @Override
 public Boolean call(Repo repo) {
 return !repo.isFork();
 }
 })
 .map(new Func1<Repo, User>() {
 @Override
 public User call(Repo repo) {
 return repo.getOwner();
 }
 })
 .distinct()
 .flatMap(new Func1<User, Observable<? extends User>>() {
 @Override
 public Observable<? extends User> call(User user) {
 return service.getUserDetails(user);
 }
 })
 .filter(new Func1<User, Boolean>() {
 @Override
 public Boolean call(User user) {
 return user.getLocation().equals("Kraków");
 }
 })
 .subscribe(
 new Action1<User>() {
 @Override
 public void call(User user) { /* send invitation email */ }
 },
 new Action1<Throwable>() {
 @Override
 public void call(Throwable error) { /* report to analytics */ }
 }
 );

56. ! orfjackal/retrolambda
57. ! evant/gradle-retrolambda
60. RX @ $
61. ! ReactiveX/RxAndroid
62. GitHubService service = /* retrofit magic */
 
 service.getUsers()
 .map(it -> { /* time consuming operation */ } )
 .subscribe( /* show users on UI */ );
63. GitHubService service = /* retrofit magic */
 
 service.getUsers()
 .subscribeOn(Schedulers.io())
 .observeOn(Schedulers.computation())
 .map(it -> { /* time consuming operation */ } )
 .observeOn(AndroidSchedulers.mainThread())
 .subscribe( /* show users on UI */ );
64. GitHubService service = /* retrofit magic */
 
 service.getUsers()
 .subscribeOn(Schedulers.io())
 .observeOn(Schedulers.computation())
 .map(it -> { /* time consuming operation */ } )
 .observeOn(AndroidSchedulers.mainThread())
 .subscribe( /* show users on UI */ ); subscribeOn(): “Create the Observable on this thread”
65. observeOn(): “Perform all other operations on this thread” GitHubService service = /* retrofit magic */
 
 service.getUsers()
 .subscribeOn(Schedulers.io())
 .observeOn(Schedulers.computation())
 .map(it -> { /* time consuming operation */ } )
 .observeOn(AndroidSchedulers.mainThread())
 .subscribe( /* show users on UI */ );
66. ! dlew/android-subscription-leaks
67. AppObservable.bindFoo() Simple & composable May leak
68. RxActivity / RxFragment Never leaks Requires base class
69. Never leaks Requires base class RoboMapsListViewSupportObservableActiv RxActivity / RxFragment
70. ! ReactiveX/RxAndroid
71. WidgetObservable ViewObservable ContentObservable
72. WidgetObservable
 .text(textView)
 .filter(t -> t.text().length() > 2)
 .debounce(300, TimeUnit.MILLISECONDS)
 .observeOn(Schedulers.io())
 .map( /* fetch suggestions */ )
 .observeOn(AndroidSchedulers.mainThread())
 .subscribe( /* show suggestions in autocomplete */ );
73. WidgetObservable
 .text(textView)
 .filter(t -> t.text().length() > 2)
 .observeOn(Schedulers.io())
 .map( /* fetch suggestions */ )
 .observeOn(AndroidSchedulers.mainThread())
 .subscribe( /* show suggestions in autocomplete */ );
74. Preconditions.checkState(isUserAMonkey()); WidgetObservable
 .text(textView)
 .filter(t -> t.text().length() > 2)
 .observeOn(Schedulers.io())
 .map( /* fetch suggestions */ )
 .observeOn(AndroidSchedulers.mainThread())
 .subscribe( /* show suggestions in autocomplete */ );
75. https://github.com/ReactiveX/RxJava/wiki/Backpressure Preconditions.checkState(isUserAMonkey()); WidgetObservable
 .text(textView)
 .filter(t -> t.text().length() > 2)
 .observeOn(Schedulers.io())
 .map( /* fetch suggestions */ )
 .observeOn(AndroidSchedulers.mainThread())
 .subscribe( /* show suggestions in autocomplete */ );
76. RX @ !
77. I want to… • load data in background, • process data in background, • cache processed data, • subscribe on changes and reload automatically… • …unless user left the UI which needs this data, in that case just mark the data as dirty and reload as needed, • have a basic backpressure handling.
78. ! futuresimple/android-db-commons
79. public class RxLoaderManager {
 public RxLoaderManager(Fragment fragment);
 
 public static abstract class LoaderBuilder<T> {
 public LoaderBuilder(int loaderId);
 protected abstract Loader<T> onCreateLoader(Context context);
 }
 
 public <T> Observable<T> load(LoaderBuilder<T> loaderBuilder); }
80. GitHubService service = /* retrofit magic */
 
 service.getUsers()
 .flatMap(Observable::from)
 .flatMap(service::listRepos)
 .flatMap(Observable::from)
 .filter(repo -> repo.getName().contains("android") ||
 repo.getDescription().contains("android"))
 .filter(repo -> repo.getLanguage().equals("Java"))
 .filter(repo -> !repo.isFork())
 .map(Repo::getOwner)
 .distinct()
 .flatMap(service::getUserDetails)
 .filter(user -> user.getLocation().equals("Kraków"))
 .subscribe(user -> { /* send invitation email */ });
81. GitHubService service = /* retrofit magic */
 
 service.getUsers()
 .flatMap()
 .flatMap()
 .flatMap()
 .filter( )
 .filter()
 .filter()
 .map()
 .distinct()
 .flatMap()
 .filter()
 .subscribe();
82. GitHubService service = /* retrofit magic */
 
 Observable<T> t = service.getUsers()
 .flatMap()
 .flatMap()
 .flatMap(); Observable<U> u = t .filter() .map() .flatMap() .distinct(); Observable<V> v = t .filter() .map() .filter(); Observable.combineLatest(u, v) .distinct()
 .flatMap()
 .filter()
 .subscribe();
83. GitHubService service = /* retrofit magic */
 
 Observable<T> t = service.getUsers()
 .flatMap()
 .flatMap()
 .flatMap(); Observable<U> u = t .filter() .map() .flatMap() .distinct(); Observable<V> v = t .filter() .map() .filter(); Observable.combineLatest(u, v) .distinct()
 .flatMap()
 .filter()
 .subscribe();
84. /*
 RxFlowGraph
 deal ~> b1 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~> assoc_contact_ids.in1
 b3 ~> assoc_contact_ids.in2
 assoc_contact_ids.out ~> assoc_contacts_list ~> associated_contact_data.in1
 b1 ~> primary_contact ~~~> b3 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~> associated_contact_data.in2
 b1 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~> associated_contact_data.in3
 b1 ~> owner_id ~> b2 ~> users_with_contacts_permission ~> permissions_map.in1
 b2 ~> users_with_accounts_permission ~> permissions_map.in2
 permissions_map.out ~> associated_contact_data.in4
 associated_contact_data.out ~>
 
 .subscribe(
 onNext ->
 );
 */
 • Inspired by Akka Streams (http://goo.gl/gjG4FZ) • Java API & implementation ~ready, but a bit ugly • Big pain: RxFlowGraph elements != Observables • To be revisited if/when Kotlin is officially supported?
85. RECAP
86. Observable<T> == Iterable<Async>
87. GOOD Escape from callback hell
88. GOOD Single point of error handling
89. GOOD Highlights spaghetti code
90. GOOD Highlights sloppy state usage
91. BAD Lifecycle handling
92. UGLY Anonymous classes boilerplate
93. HOT or NOT?
94. HOT or NOT? HOT in WidgetObservables
95. HOT or NOT? HOT in Retrofit
96. HOT or NOT? HOT in complex db access
97. HOT or NOT? NOT hot everywhere
98. “Well, Mr. Frankel, who started this program, began to suffer from the computer disease that anybody who works with computers now knows about. It's a very serious disease and it interferes completely with the work. The trouble with computers is you *play* with them. They are so wonderful. You have these switches - if it's an even number you do this, if it's an odd number you do that - and pretty soon you can do more and more elaborate things if you are clever enough, on one machine. After a while the whole system broke down. Frankel wasn't paying any attention; he wasn't supervising anybody. The system was going very, very slowly - while he was sitting in a room figuring out how to make one tabulator automatically print arc- tangent X, and then it would start and it would print columns and then bitsi, bitsi, bitsi, and calculate the arc-tangent automatically by integrating as it went along and make a whole table in one operation. Absolutely useless. We *had* tables of arc-tangents. But if you've ever worked with computers, you understand the disease - the *delight* in being able to see how much you can do. But he got the disease for the first time, the poor fellow who invented the thing.” ― Richard P. Feynman, Surely You're Joking, Mr. Feynman!
99. ?
100. Thanks.