[ngrx/store-2] ngrx/store 用到的 Observable 跟 Subject

2018鐵人賽

洋蔥

2017-12-18 14:44:05

4829 瀏覽

分享至

Observable

如果您寫 Angular, 那您一定對 Observable 不陌生, 當您對後端做 GET, POST, PUT, DELETE 要求時所使用的 HttpClient 就是 Observable. 因為整個 ReactiveX 非常龐大, 甚至包含了其他程式語言, 這裡我們只簡單的介紹針對 Javascript RxJS 而且在 ngrx/store 常用到的基本定義以及運算子, 完整的 RxJs 請參考 RxJS API

產生 Observable

Observable 簡單來說就是資料流, 至於如何產生 Observable 呢? 一般而言有幾種情況

同步靜態的如純量, Array, Set 或是 Object

Rx.Observable.of(true);
Rx.Observable.from([1,2,3,4]);
const user = [
    { name: 'John', age: 36},
    { name: 'Jason', age: 24},
    { name: 'Jennifer', age: 33}
]
Rx.Observable.from(user);

同步動態的如利用 Event, 例如 mouse event, input event 或是非同步時間等等

Rx.Observable.fromEvent(mouse, 'mousemove');
Rx.Observable.interval(1000);

從非同步的 Promise 產生

// getUser from github
function getUser(username) {
    return $.ajax({
        url: 'https://api.github.com/users/' + username,
        dataType: 'jsonp'
    }).promise(); 
}
Rx.Observable.fromPromise(getUser('myname'));

自行建立 Observable

new Rx.Observable(observer => {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    setTimeout(() => {
        observer.next(4);
        observer.complete();
    }, 1000);
});

Data Provider, Observable, Subscriber 跟 Observer

有幾個比較重要的基本定義

Data Provider: 也就是資料流的來源, 像上面的 of, from, fromPromise... 的對象
Observable: 就是觀察資料流的物件, 它有一個重大的任務, 當資料來源變更時, 需要馬上告訴他的客戶 (Observer, Subscriber) 我的資料變動了, 這個是稱為 push 的方式, 而 push 的方式就是當資料變動時啟動 Observer next 的 method, 所以客戶自己要定義接到這個 next 時， onNext callback 要做什麼因應, 接到 error 時 onError 要做什麼...
Observer: 也就是 Observable 的客戶，主要任務是定義一些 callback 處理當 Observable 啟動 next, error 或是 complete 時要做的事
Subscriber 或 Subscription: 是 Observer 的實作, 除了定義 Observer 要做的 callback 外，另外還可以隨時做 unsubscribe, 而這也是跟 Promise 很不同的地方

public abstract class Subscriber<T> implements Observer<T>, Subscription

用一個例子來說明

const source$ = new Rx.Observable(observer => {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    observer.error(new Error('error: error message'));
    setTimeout(() => {
        observer.next(4);
        observer.complete();
    }, 1000);
});

source$.subscribe(
    val => {
        console.log(val);
    },
    err => {
        console.log(err);
    },
    complete => {
        console.log('completed');
    }
);

codepen

Data Provider: 是簡單的 1, 2, 3
Observable: 也就是 source$, 他要告訴 Observer 資料的變化, 像是 observer.next(1), ...
Subscription: 定義了相對應的個別 callback
所以以上的程式會產生

還是覺得很神奇？那我們自己來打造一個簡單的 Observable 好了

var Observable = (function (subscriber) {
  this._subscriber = subscriber;
})

Observable.prototype.subscribe = function(next, error, complete) {
  this._subscriber(next);
}

const source$ = new Observable(observer => {
  observer(1);
  observer(2);
  observer(3);
});

const subs = source$.subscribe(
                  (v) => {
                      console.log('subscribed: ' + v)
                  },
                  (err) => {
                      console.log('error with' + err)
                  },
                  (complete) => {
                      console.log('complete');
                  }
)

codepen

輸出的結果會是

Subject

最後談一下 Subject, 因為 ngrx/store 裡的 store 跟 dispatcher 都是 Subject (BehaviorSubject)
Subject 是 Observable 跟 Observer 的延伸, 也就是同時具有 Observable 跟 Observer 的特徵, 他可以被 Subscriber subscribe 也同時 subscribe 到其他的 Observable. 我們可以當它是代理人, 舉個例子

var subject$ = new Rx.Subject();

// acting as Observable
const sub1 = subject$.subscribe({
  next: (v) => console.log('observerA: ' + v)
});
const sub2 = subject$.subscribe({
  next: (v) => console.log('observerB: ' + v)
});

// acting as Observer
subject$.next(1);
subject$.next(2);

codepen

BehaviorSubject

BehaviorSubject 延伸 Subject 再加上它會保留一下最後的一筆資料，也就是即使過了資料產生的時間，後來的訂閱者還是可以拿到最後產生時的資料，也因為如此，在產生新的 BehaviorSubject 時，需要給初值。

如上面例子

var subject$ = new Rx.Subject();

// acting as Observable
const sub1 = subject$.subscribe({
  next: (v) => console.log('observerA: ' + v)
});
const sub2 = subject$.subscribe({
  next: (v) => console.log('observerB: ' + v)
});

// acting as Observer
subject$.next(1);
subject$.next(2);

const sub3 = subject$.subscribe({
    next: (v) => console.log('observerC: ' + v)
});

codepen
sub3 不會得到更新
但是換成 BehaviorSubject

var subject$ = new Rx.BehaviorSubject(0);

// acting as Observable
const sub1 = subject$.subscribe({
  next: (v) => console.log('observerA: ' + v)
});
const sub2 = subject$.subscribe({
  next: (v) => console.log('observerB: ' + v)
});

// acting as Observer
subject$.next(1);
subject$.next(2);

const sub3 = subject$.subscribe({
    next: (v) => console.log('observerC: ' + v)
});

codepen

sub3 還是會拿到最後的值 (2)，產生 ObservableC: 2

有了 Observable 的基本認識, 接下來我們來談一些運算子 Operators