rxjs 只在第一次执行tap

Question

我只想在获得第一个发射值时执行 tap()

就像是：

 Observable
  .pipe(
     tap(() => { /* execute only when I get the first emitted value */ })
  )
  .subscribe(() => {
     // .....
  })

原文由 Huantao 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

您可以在地图运算符中使用索引，例如 concatMap 。与其他方法不同，这对于所选索引是完全灵活的。假设你想点击第二次发射 index === 1 或任何谓词，如 index % 2 === 0

 // these are because of using rxjs from CDN in code snippet, ignore them
const {of, interval} = rxjs;
const {take, tap, concatMap} = rxjs.operators;

// main code
const stream = interval(250).pipe(take(4))

stream.pipe(
  concatMap((value, index) => index === 0
    ? of(value).pipe(
        tap(() => console.log('tap'))
      )
    : of(value)
  )
)
.subscribe(x => console.log(x));

 <script src="https://unpkg.com/@reactivex/rxjs@6.x/dist/global/rxjs.umd.js"></script>

原文由 jal 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 2

如果我正确地理解了您想要执行的想法 tap() 仅在流订阅开始时而不是其他时间。 这是我的自定义运算符：

 import { Observable, of } from 'rxjs';
import { switchMap, tap } from 'rxjs/operators';

export function startWithTap<T>(callback: () => void) {
  return (source: Observable<T>) =>
    of({}).pipe(tap(callback), switchMap((o) => source));
}

---

例如，此运算符的用法是：

 this.api.getData().pipe(
  startWithTap(() => this.loading.start()),
)

这是我的实际代码示例，当有人订阅由 api 服务（通过 httpClient）创建的 Observable 时，加载开始。

---

更新

Use this instead of the above implementation, because this one only uses defer , instead of using of , tap and switchMap :

 export function startWithTap<T>(callback: () => void) {
  return (source: Observable<T>) =>
    defer(() => {
      callback();
      return source;
    });
}

原文由 Goga Koreli 发布，翻译遵循 CC BY-SA 4.0 许可协议