[譯] RxJS: multicast 操做符的祕密

原文連接: blog.angularindepth.com/rxjs-multic…

本文爲 RxJS 中文社區 翻譯文章，如需轉載，請註明出處，謝謝合做！

若是你也想和咱們一塊兒，翻譯更多優質的 RxJS 文章以奉獻給你們，請點擊【這裏】

multicast 操做符有一個祕密。publish 操做符也是如此，它封裝了 multicast 。這個祕密有時候真的挺好用的。

祕密

multicast 和 publish 的文檔中都提到了 ConnectableObservable 。ConnectableObservable 是一種特殊類型的 observable，只有調用它的 connect 方法後，它纔會開始向訂閱者發送通知。然而，multicast 和 publish 操做符並不是永遠返回 ConnectableObservable 。

咱們先來看下 publish 的源碼:

export function publish<T>(
  this: Observable<T>,
  selector?: (source: Observable<T>) => Observable<T>
): Observable<T> | ConnectableObservable<T> {
  return selector ?
    multicast.call(this, () => new Subject<T>(), selector) :
    multicast.call(this, new Subject<T>());
}
複製代碼

能夠很清楚地看出，publish 只是對 multicast 進行了一層很薄的封裝。它建立了 subject 並傳給 multicast，還有一個可選的 selector 函數。最有趣的部分是在 multicast 實現之中，它包含以下代碼:

if (typeof selector === 'function') {
  return this.lift(new MulticastOperator(subjectFactory, selector));
}

const connectable: any = Object.create(this, connectableObservableDescriptor);
connectable.source = this;
connectable.subjectFactory = subjectFactory;
return <ConnectableObservable<T>> connectable;
複製代碼

只有在不傳入 selector 函數的狀況下，multicast 才返回 ConnectableObservable。若是傳入 selector 函數的話，會使用 lift 機制來使得源 observable 建立出適當類型的 observable 。不須要在返回的 observable 上調用 connect 方法，而且在 selector 函數的做用域中會共享源 observable 。

這意味着 multicast (以及 publish) 操做符能夠用來輕鬆實現源 observable 的本地共享。

使用 publish 進行本地共享

咱們來看看使用 publish 的示例。

RxJS 引入了 defaultIfEmpty 操做符，它接收一個值，若是源 observable 爲空的話，會將這個值發出。有時候，可以指定一個默認 observable 的話要比指定單個值有用得多，那麼讓咱們來實現一個 defaultObservableIfEmpty 函數，它能夠與 let 操做符一塊兒使用。

下面的彈珠圖展現了源 observable 爲空時它的行爲:

RxJS 引入了 isEmpty 操做符，當源 observable 完成時，它會發出布爾值以標識源 observable 是否爲空。可是，要在 defaultObservableIfEmpty 實現中使用它的話，須要共享源 observable，由於須要發出值的通知，而 isEmpty 沒法作到這點。publish 操做符使得源 observable 的共享變得簡單，實現以下所示:

function defaultObservableIfEmpty<T>( defaultObservable: Observable<T> ): (source: Observable<T>) => Observable<T> {

  return source => source.publish(shared => shared.merge(
    shared.isEmpty().mergeMap(empty => empty ?
      defaultObservable :
      Observable.empty<T>()
    )
  ));
}
複製代碼

傳給 publish 的 selector 函數接收共享的源 observable 。selector 返回的 observable 是由共享源 observable 和根據源 observable 是否爲空獲得的 observable (若是源 observable 爲空，則爲傳入的默認 observable，不然爲空 observable) 的組合而成。

源 observable 的共享徹底是由 publish 管理的。使用 selector 函數，就可以根據須要屢次訂閱共享 observable，而不會影響源 observable 後面的訂閱。

使用 multicast 進行本地共享

咱們來看另外一個示例，此次使用 multicast 。

RxJS 引入了 takeWhile 操做符，它返回的 observable 會發出源 observable 的值，直到不知足給定條件的值出現，此刻 observable 完成。不知足條件的那個值不會被髮出。咱們來實現一個 takeWhileInclusive 函數，它能夠與 let 操做符一塊兒使用。

下面的彈珠圖展現了值不知足條件時的行爲:

可使用 takeWhile 操做符做爲基礎來實現，當不知足條件時，只須要再鏈接不知足條件的那個值便可。要在 takeWhile 返回的 observable 完成後取得這個值，可使用 ReplaySubject:

function takeWhileInclusive<T>(
  predicate: (value: T) => boolean
): (source: Observable<T>) => Observable<T> {

  return source => source.multicast(
    () => new ReplaySubject<T>(1),
    shared => shared
      .takeWhile(predicate)
      .concat(shared.take(1).filter(t => !predicate(t)))
  );
}
複製代碼

這裏使用了緩衝區大小爲1的 ReplaySubject 來共享源 observable 。當 takeWhile 操做符返回的 observable 完成時，共享的 observable 是串聯的，使用 take(1) 能夠確保只考慮重放的值，而 filter 能夠確保只有當不知足條件時才進行追加。

這種方式可靠嗎？

RxJS 5 是至關新的庫，它的文檔扔在進行中，因此這種方式尚未在文檔中出現，只是在內部使用。公開的 TypeScript 簽名指出了並不是永遠返回的是 ConnectableObservable，也有相對應的單元測試。

RxJS 一般比較靈活，所以實現這些函數還有其餘方式，但上面的示例說明了當須要本地共享源 observable 時，publish 和 multicast 簡單易用，值得考慮。