全面教程：在 RxJS 中建立流 | Linux 中國

繼承可觀察物件

在有可用的運運算元前，這是一種實現自定義運運算元的方式。RxJS 在內部擴充套件了 可觀察物件 。Subject 就是一個例子，另一個是 publisher 運運算元。它傳回一個 ConnectableObservable 物件，該物件提供額外的方法 connect。

實現 Subscribable 介面

有時候你已經用一個物件來儲存狀態，並且能夠傳送值。如果你實現了 Subscribable 介面，你可以把它轉換成一個可觀察物件。Subscribable 介面中只有一個 subscribe 方法。

interface Subscribable<T> {  subscribe(observerOrNext?: PartialObserver<T> | ((value: T) => void), error?: (error: any) => void, complete?: () => void): Unsubscribable}

結合和選擇現有的流

知道怎麼建立一個獨立的流還不夠。有時候你有好幾個流但其實只需要一個。有些函式也可作為運運算元，所以我不打算在這裡深入展開。推薦看看 Max NgWizard K[10] 所寫的一篇 文章[11]，它還包含一些有趣的動畫。

還有一個建議：你可以透過拖拽元素的方式互動式的使用結合操作，參考 RxMarbles[12]。

ObservableInput 型別

期望接收流的運運算元和函式通常不單獨和可觀察物件一起工作。相反，它們實際上期望的引數型別是 ObservableInput，定義如下：

type ObservableInput<T> = SubscribableOrPromise<T> | ArrayLike<T> | Iterable<T>;

這意味著你可以傳遞一個 promises 或者陣列卻不需要事先把他們轉換成可觀察物件。

defer

主要的目的是把一個 observable 物件的建立延遲（defer）到有人想要訂閱的時間。在以下情況，這很有用：

最後一點包含了一個並不起眼的用例：Promises（defer 也可以傳回一個 promise 物件）。看看這個用到了 fetch API 的例子：

function getUser(id) {
  console.log("fetching data");
  return fetch(`https://server/user/${id}`);
}
const userPromise = getUser(1);
console.log("I don't want that request now");
// 其它地方
userPromise.then(response => console.log("done");

// 輸出
// fetching data
// I don't want that request now
// done

const user$ = defer(() => getUser(1));
console.log("I don't want that request now");
// 其它地方
user$.subscribe(response => console.log("done");

// 輸出
// I don't want that request now
// fetching data
// done

iif(
  () => new Date().getHours() < 12,
  of("AM"),
  of("PM")
)
.subscribe();

// 結果
// AM before noon, PM afterwards

const stream1$ = of(1, 2).pipe(
  tap(i => { if(i>1) throw 'error'}) //fail after first element
);
const stream2$ = of(3,4);
onErrorResumeNext(stream1$, stream2$)
  .subscribe(console.log);

// 結果
// 1 3 4 |

function handleResponses([user, account]) {
  // 執行某些任務
}
forkJoin(
  fetch("https://server/user/1"),
  fetch("https://server/account/1")
)
.subscribe(handleResponses);

merge(
  interval(1000).pipe(mapTo("Stream 1"), take(2)),
  interval(1200).pipe(mapTo("Stream 2"), take(2)),
  timer(0, 1000).pipe(mapTo("Stream 3"), take(2)),
  2 //two concurrent streams
)
.subscribe();
// 只訂閱流 1 和流 2
// 輸出
// Stream 1 -> after 1000ms
// Stream 2 -> after 1200ms
// Stream 1 -> after 2000ms
// 流 1 結束後，開始訂閱流 3
// 輸出
// Stream 3 -> after 0 ms
// Stream 2 -> after 400 ms (2400ms from beginning)
// Stream 3 -> after 1000ms
merge(
  interval(1000).pipe(mapTo("Stream 1"), take(2)),
  interval(1200).pipe(mapTo("Stream 2"), take(2))
  1
)
// 等效於
concat(
  interval(1000).pipe(mapTo("Stream 1"), take(2)),
  interval(1200).pipe(mapTo("Stream 2"), take(2))
)
// 輸出
// Stream 1 -> after 1000ms
// Stream 1 -> after 2000ms
// Stream 2 -> after 3200ms
// Stream 2 -> after 4400ms

zip(
  interval(1000),
  interval(1200),
)
.subscribe();

// 結果
// [0, 0] [1, 1] [2, 2] ...

combineLatest(
  interval(1000),
  interval(1200),
)
.subscribe();

// 結果
// [0, 0] [1, 0] [1, 1] [2, 1] ...

zip(
  interval(1000),
  interval(1200),
  (e1, e2) -> e1 + e2
)
.subscribe();

// 結果
// 0 2 4 6 ...

race(
  interval(1000),
  of("foo")
)
.subscribe();

// 結果
// foo |