Learning Reactive-Extensions (RxJava & RxJS)

[JimmyLv]

• ReactiveX/rxjs: A reactive programming library for JavaScript
• RxJS - The basics tutorial
• Use Higher Order Observables in RxJS Effectively - Course by @andrestaltz @eggheadio

项目中的经验非常有用！从问题描述开始，聚合性 portfolio 所 deps 的各种 APIs：

Sam Newman - Backends For Frontends 文中的
And Multiple Downstream Services (Microservices!) 章节：

Multiple services hold the pieces of information we want. The Wishlist service stores information about the list, and IDs of each item. The Catalog service stores the name and price of each item, and the Stock levels are stored in our inventory service. So in our BFF we'd expose a method for retrieving the full playlist, which would consist of at least 3 calls:

Making multiple downstream calls to construct a view of a wishlist

From an efficiency point of view, it would be much smarter to run as many calls in parallel as possible. Once the initial call to the Wishlist service completes, ideally we'd like to then run the calls to the other services at the same time to reduce the overall call time. This need to mix calls that we want to run in parallel vs those that run in sequence can quickly become painful to manage, especially for more complex scenarios.

This is one area where a reactive style of programming can help (such as that provided by RxJava or Finagle's futures system) as the composition of multiple calls becomes easier to manage.

References

http://reactivex.io/documentation/operators.html official documents
https://www.javadoc.io/doc/io.reactivex/rxjava/1.1.1 rxjava 1.1.1 api docs
http://reactivex.io/RxJava/javadoc/ rxjava 2 api docs
http://gank.io/post/560e15be2dca930e00da1083 useful tutorial for beginners

[JimmyLv]

对比 MobX #75，放弃的原因：Mobx vs Reactive Stream Libraries (RxJS, Bacon, etc) · mobxjs/mobx Wiki

[JimmyLv]

RxJS 并不只是 JavaScript，而是整个 Reactive-Extensions 体系，还包括现在项目所用的 RxJava，其思想和操作符更为重要和值得学习，而反而是 MobX 只局限于 React 世界里，并且还不一定拼得过 Redux，以及未来的 Relay 和 GraphQL…

[JimmyLv]

@Yanghai 组内分享 RxJava 的 Session 很不错，AGENDA 如下：

• Introduction
• Core Conceptions
• Event Transformation
• More Operations

Introduction

RxJava is a Java implementation of ReactiveX:
a library for composing asynchronous and event-based programs by using observable sequences.

Observable.from(folders)
    .flatMap((Func1) (folder) -> { Observable.from(file.listFiles()) })
    .filter((Func1) (file) -> { file.getName().endsWith(".png") })
    .map((Func1) (file) -> { getBitmapFromFile(file) })
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe((Action1) (bitmap)->  { imageCollectorView.addImage(bitmap) });

Core Conceptions

RxJava contains the following core concepts:

Observable: event source
Observer/Subscriber: event consumer
Scheduler: thread controller
Subscribe: the operation to trigger event sequence

Obervable.onSubscrible() triggers events

Observable

Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {
        subscriber.onNext("Hello");
        subscriber.onNext("Hi");
        subscriber.onNext("Aloha");
        subscriber.onCompleted();
    }
});

Observable observable = Observable.just("Hello", "Hi", “Aloha");

It’s equivalent to:

String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);

Subscriber

一般不用关心，Subscriber 属于 RxJava 实现上的内部概念。

Subscriber<String> subscriber = new Subscriber<String>() {
    @Override
    public void onNext(String s) {
        Log.d(tag, "Item: " + s);
    }

    @Override
    public void onCompleted() {
        Log.d(tag, "Completed!");
    }

    @Override
    public void onError(Throwable e) {
        Log.d(tag, "Error!");
    }
};

Observer

Observer will be converted into Subscriber.

Observer<String> observer = new Observer<String>() {
    @Override
    public void onNext(String s) {
        Log.d(tag, "Item: " + s);
    }

    @Override
    public void onCompleted() {
        Log.d(tag, "Completed!");
    }

    @Override
    public void onError(Throwable e) {
        Log.d(tag, "Error!");
    }
};

Subscribe

String[] words = {"Hello", "Hi", “Aloha”};
Subscriber<String> subscriber = new Subscriber<String>() {
    @Override
    public void onNext(String s) {
        Log.d(tag, “Word: " + s);
    }

    @Override
    public void onCompleted() {
        Log.d(tag, "Completed!");
    }

    @Override
    public void onError(Throwable e) {
        Log.d(tag, "Error!");
    }
};

Observable.from(words).subscribe(subscriber);

Event transformation

More operations

[JimmyLv]

通过项目实战以及听 Session 对 RxJava 有了基本的经验，并且更重要的是明白了其背后 Reactive 的理念和思考方式，而 RxJS 则是 JavaScript 世界的一种实现，更多的 RxJS 学习记录将在 #260