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?

Does this answer your question? [rxjs execute tap only at the first time](https://stackoverflow.com/questions/54099238/rxjs-execute-tap-only-at-the-first-time) — frido, Oct 28 '20 at 11:17

score 4 · Answer 1 · answered Oct 28 '20 at 09:20

4

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
      });

answered Oct 28 '20 at 09:20

Picci

16,775
13
70
113

question: why do you need to `merge` them? and why do you need to `skip(1)` in `otherEmissions$`? - oh is it because the first one is coming from the `firstEmission$` and so you just `merge` to `process them all` use-case, right? – Sergino Oct 28 '20 at 10:14
oh, BTW may be you may have any ideas on that https://stackoverflow.com/questions/64566051/how-to-make-observable-to-repeat – Sergino Oct 28 '20 at 10:22
@sreginogemoh `merge` is used to create a single observable, merging the 2 ones we created, and so have a single subscription. If you so not merge them, you need to subscribe to them separately. `skip(1) is required to avoid that the merged Observable emits twice the first item emitted by `observable$`. – Picci Oct 28 '20 at 13:44

score 1 · Answer 2 · answered Oct 28 '20 at 12:43

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)

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

2 Answers2