Building a custom tapOnce operator

If you work with rxjs in your codebase, chances are you've reached for the tap operator at some point.
The tap operator leaves the stream untouched while letting you run a function (or side-effect) with each emitted value as its argument. A common use case is logging values to the console:

source.pipe(
  tap(val => console.log(val))
);

But what about those situations where you only want that side-effect to run for the very first emission? There's no built-in operator for that in rxjs — but that’s not a problem, because we can create our own.

Anatomy of a custom operator

If you’ve never looked under the hood of an rxjs operator, here’s the essence: an operator is simply a function that accepts an observable and returns another observable. That’s the whole contract.
In practice, we usually transform the input observable or perform some operation on it before handing it back. For a deeper dive into authoring custom operators, check out this article or this one.
Following that pattern, a minimal custom operator looks something like this:

function basicCustomOperator<T>() {
    return function(source: Observable<T>) {        
        return source;
    };
}

You would then use it in your code like this:

source.pipe(
  basicCustomOperator()
)

Now that we have the basic shape of a custom operator in mind, let’s build tapOnce.

Implementing tapOnce

One straightforward idea is to leverage the existing take operator. We can apply take(1) to the source, subscribe to that inner observable, and then pass the original source through to the rest of the pipe.

function tapOnce<T>(fn: (value)=> void) {
    return function(source: Observable<T>) {
        source
            .pipe(
                take(1),
                tap(value => fn(value))
            )
            .subscribe();

        return source;
    };
}

On the surface, this looks reasonable, and it will behave correctly in many scenarios. However, it comes with a few caveats.
For one, it establishes a subscription on the source. That can lead to surprising behavior — for instance, the side-effect might fire even if the outer observable is never subscribed to.

const sourceSubject = new Subject();

const source = sourceSubject.pipe(
  tapOnce(x => console.log(`tapOnce ${x}`)),
);

sourceSubject.next("1");
sourceSubject.next("2");

// tapOnce will execute even if we didn't subscribe to sourceSubject

Additionally, because we hold onto the same source reference, multiple subscriptions to the resulting observable will only trigger the effect once.

const sourceSubject = new Subject();

const source = sourceSubject.pipe(
  tapOnce(x => console.log(`tapOnce ${x}`))
);

source.subscribe();
source.subscribe();

sourceSubject.next("1");
sourceSubject.next("2");

// tapOnce will execute on first subscribe, when we would expect to run for both

There’s another edge case when the operator is combined with takeUntil.

source.pipe(
  tapOnce(x=> console.log(`tapOnce ${x}`)),
  takeUntil(takeUntilSource)
).subscribe(x=> console.log(x))

source.subscribe();

takeUntilSource.next()
source.next('1')
source.next('2')
source.next('3')

If takeUntil emits before the source produces its first value, you’d reasonably expect tapOnce to stay silent. But that’s not what happens — the inner subscription in our operator fires regardless. The root cause is that this internal subscription is unaware of the takeUntil placed later in the pipe.

Fixing it with defer

Let’s explore a better solution. What we really need is an operator that runs our code only when the source is actually subscribed, and that also isolates state between multiple subscriptions. Fortunately, rxjs ships with the defer operator, which handles both concerns.
Here’s the improved version of tapOnce:

function tapOnce<T>(fn: (value) => void) {
    return (source: Observable<any>) =>
        defer(() => {
            let first = true;
            return source.pipe(
                tap<T>((payload) => {
                    if (first) {
                        fn(payload);
                    }
                    first = false;
                })
            );
        });
}

Let’s walk through what this code does. By returning a defer() observable, we ensure the inner logic is instantiated only at subscription time.

defer(() => {
  // ...
  return source.pipe(
    // ...
  );
});

Moreover, defer generates a fresh observable for each subscription, so the first flag stays per-subscription. That guarantees each subscriber gets its own tracking state. You can dig into the details here.

In the end, all we’re doing is checking whether this is the first emission, and deciding whether to invoke the provided function.

// ...
let first = true;
return source.pipe(
  tap((payload) => {
    if (first) {
      fn(payload);
    }
    first = false; 
  })
);

With that, we now have a tapOnce operator that behaves consistently across all the scenarios we examined.

Thanks for reading! Feel free to experiment with the code here:

StackBlitz