rxjs выполняет тап только в первый раз

Question

Я хочу выполнить tap (), только когда получу первое значение

Что-то вроде:

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

Answer 1

Помимо уже упомянутых опций вы также используете multicast.

multicast(new Subject(), s => concat(
  s.pipe(
    take(1),
    tap(v => console.log('tap', v)),
  ),
  s
)

Живая демоверсия: https://stackblitz.com/edit/rxjs-shvuxm

Answer 2

Вы можете использовать индекс в операторах карты как concatMap. В отличие от других подходов, это абсолютно гибкий подход к выбранному индексу. Допустим, вы хотите нажать на 2-е излучение 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>

Answer 3

Если я правильно понял вашу идею, вы хотите выполнить 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()),
)

Это мой пример кода, где загрузка начинается, когда кто-то подписывается на Observable, созданный сервисом API (посредством httpClient).

Answer 4

(Обновление моего ранее неправильного ответа)

Основываясь на предоставленных комментарии и ссылке на картан, он уже проделал работу по созданию оператора, который делает это, он находится в пакете 'rxjs-etc'. Решением, основанным на его операторе, является установка 'rxjs-etc', затем:

import { initial } from 'rxjs-etc/operators';

observable$.pipe(
    initial(src$ => src$.pipe(tap(() => {/* execute only on first value */})))
).
subscribe(() => {
    // ..... 
})

Работа StackBlitz пример.