How to improve RxJS code to apply a function only for the first emission

Question

I have observable that emitting values every N seconds and I have. I want to take the first emission and fire a function on that.

I have this code:

      sideEffect = false;
      observable$
      .pipe(
        tap((data) => {
          // executing this only for the first emission
          if (!sideEffect) {
            this.sideEffect(data.props);
            sideEffect = true;
          }
        })
      )
      .subscribe((data) => {
        // process all upcoming emissions
      });

Is there anyway to make this code better using RxJS operators without defining any local variables?

Answer 1

I would proceed creating 2 separate Observables and then merge them, so that we can have a single subscription. I think this is the most idiomatic way of implementing this logic via rxjs.

The code would look something like this

const firstEmission$ = observable$
      .pipe(
        take(1),  // you can use also the first() operator which is the same as take(1)
        tap((data) => this.sideEffect(data.props))
      );
const otherEmissions$ = observable$
      .pipe(
        skip(1),
      );
merge(firstEmission$, otherEmissions$).subscribe((data) => {
        // process all upcoming emissions
      });

Answer 2

There is a scan operator that can solve your problem:

const stream$ = from([1,2,3,4]).pipe(
  scan((firstValue, value) => {
    if (!firstValue) {
      firstValue = value;
      console.log('performing side effects for value: ', value);
    }
    return value;
  }, null),
  map(value => { /* your further logic here */ return value; })
);

scan is called for every emission and stores the value inside itself. It's similar to reduce function in arrays. Also be aware that "falsy" values might behave incorrectly (because of !firstValue check)