等待完成然后發出一個值的RxJS運算子

是否有 RxJS 運算子等待源完成然后發出給定值？如果沒有，我怎么能自己提供？

這將與toArray()等待源完成的作業類似。我不想收集所有發出的值，而是想忽略它們并回傳一個不同的值。

這是一個等效的實作：

observable.pipe(
  ignoreElements(),
  endWith(myValue),
);

或者：

observable.pipe(
  toArray(),
  map(ignore => myValue)
)

經常有我需要這個的情況。我得出的結論，這是危險的承諾轉化then通過-chains的可觀switchMap()或mergeMap()因內觀測可以完成，而完全不發光的值。最近我們遇到了這個問題：

return getEntitiesFromBackend().pipe(
  switchMap(entities => {
    return putEntitiesToObjectStore(entities);
  }),
  switchMap(() => {
    return storeSyncDate();
  })
);

在某些情況下，同步日期沒有存盤，而且很難找出原因。最后，原因是該putEntities...方法為其“放置”操作發出一個值。但在這些情況下，entities陣列是空的，所以根本沒有發出任何值。

這就是我真正想做的 - 翻譯成承諾世界：

return getEntitiesFromBackend()
  .then(entities => {
    return putEntitiesToObjectStore(entities);
  })
  .then(() => {
    return storeSyncDate();
  })
);

我看到的大多數使用switchMap/的代碼mergeMap都沒有這個問題。因為大多數時候你處理的 HTTP 請求只發出一次然后完成。例如，請參見此處。這讓我習慣于將典型的 Promise 模式轉換為 RxJS 世界，switchMap而無需過多考慮它的實際作業和目的。現在，我們使用 IndexedDB，我們的大多數方法都會回傳 observables，為每個 DB 操作發出一個值。switchMap/mergeMap會在這里做噩夢。

這就是為什么我要求這樣的運算子，并想知道為什么我還找不到它，因為它在我們的應用程式中很常見。我可以通過使用上面的替代實作輕松解決這個問題，但不想一遍又一遍地重復這兩個運算子：

return getEntitiesFromBackend().pipe(
  switchMap(entities => {
    return putEntitiesToObjectStore(entities);
  }),
  ignoreElements(),
  endWith(),
  switchMap(() => {
    return storeSyncDate();
  })
);

當然，我可以使用toArray()并忽略下一個運算子中的引數。我不喜歡它，因為它會導致不必要的開銷。

uj5u.com熱心網友回復：

我建議兩個改變。

1. 使用concatMap代替switchMap或mergeMap。concatMap將確保來自getEntitiesFromBackend()Observable 的每個發射將一個接一個地依次轉發，而不是并行 ( mergeMap) 或將被取消 ( switchMap)。有關不同型別的高階映射運算子的簡要介紹，請參見此處。

2. 您可以將運算子與始終回傳默認值的謂詞結合使用，而不是像ignoreElements 這樣的運算子組合。這樣，當源 observable 完成時，將發出默認值。maplastfalse

return getEntitiesFromBackend().pipe(
  concatMap(entities => {
    return putEntitiesToObjectStore(entities).pipe(
      last(() => false, myValue),
    );
  }),
  concatMap((myValue) => {
    return storeSyncDate();
  })
);

uj5u.com熱心網友回復：

在我看來，您想實作這一目標：

source.pipe(
   last(), // emits only last value, when the source completes
   map(() => myValue), // or mapTo(myValue)
)

uj5u.com熱心網友回復：

使用finalize運算子怎么樣？

https://rxjs.dev/api/operators/finalize

uj5u.com熱心網友回復：

這是我的嘗試，但我不知道它是否正常作業（即正確處理錯誤等）

function emitOnComplete(value) {
  return function<T>(source: Observable<T>): Observable<T> {
    return new Observable(subscriber => {
      return source.subscribe({
        next() {},
        complete() {
          subscriber.next(value);
          subscriber.complete();
        }
      });
    });
  };
}

或者我想出了這個，但它未經測驗：

function emitOnComplete(value) {
  return function <T>(source: Observable<T>): Observable<T> {
    return source.pipe(
      ignoreElements(),
      // concat() is used to make sure we emit the value AFTER source completed
      // The tick is: source will never emit any value, because we use ignoreElements() above.
      // So the user will only get `of(value)`, but after the source is completed
      () => concat(source, of(value)),
    );
  }
}

uj5u.com熱心網友回復：

想到了幾個選項。

defaultIfEmpty

這不完全是您所描述的，但它可能適合您的用例。

defaultIfEmpty：如果源 Observable 完成而不發出任何下一個值，則發出給定值，否則鏡像源 Observable。

return getEntitiesFromBackend().pipe(
  switchMap(entities => putEntitiesToObjectStore(entities)),
  defualtIfEmpty(null),
  switchMap(_ => storeSyncDate())
);

創建您自己的運算子

有一個靜態版本pipe可以在沒有應用程式到任何給定流的情況下執行組合。

因此，例如：

const a = source.pipe(
  ignoreElements(),
  endWith(myValue)
);

const b = pipe(
  ignoreElements(),
  endWith(myValue)
);

這里，a是一個可觀察的。Pipe 組合兩個運算子，然后回傳將組合運算子應用于source.

b另一方面只是第一步。b本身就是一個運算子。運算子尚未應用于可觀察物件。

所以你可以做這樣的事情：

source1.pipe(b);
source2.pipe(b);

我已經重復使用了我的b接線員兩次。我們大部分時間都在那里！

RxJS 運算子非常有用，因為它們由根據您的需要自定義運算子的函式回傳。在上述情況下，每次使用 時b， 的值都相同endwith。

我們可以將 b 包裝在一個函式中，以便在每次使用的基礎上進行定制。

const myCustomOperator = myValue => pipe(
  ignoreElements(),
  endWith(myValue)
);

source1.pipe(
  myCustomOperator(22)
);

source2.pipe(
  myCustomOperator(23)
);

這就像任何其他運算子一樣作業，因此它也可以與所有其他標準運算子組合（“管道化”）。

我可以通過使用上面的替代實作輕松解決這個問題，但不想一遍又一遍地重復這兩個運算子：

現在你有了一段可重用的代碼！

我將如何實作這一點：

JavaScript：

function ignoreConcatWith(genObs) {
  return pipe(
    ignoreElements(),
    concatWith(from(genObs()))
  );
}

[...]
  return getEntitiesFromBackend().pipe(
    switchMap(entities => putEntitiesToObjectStore(entities)),
    ignoreConcatWith(() => storeSyncDate())
  );
[...]

我發現看到靜態型別的代碼通常真的有助于我的理解。所以在 TypeScript 中也是同樣的事情：

function ignoreConcatWith<T,R>(genObs: () => ObservableInput<R>): OperatorFunction<T,R> {
  return pipe(
    ignoreElements(),
    concatWith(from(genObs()))
  );
}

[...]
  return getEntitiesFromBackend().pipe(
    switchMap(entities => putEntitiesToObjectStore(entities)),
    ignoreConcatWith(() => storeSyncDate())
  );
[...]

標籤：有角的 承诺 rxjs angular2-observables 反应性x

上一篇：如何從我的物件創建一個數字陣列？

下一篇：如何在BehaviourSubject陣列中搜索單個物件