RxJS take, takeUntil, takeWhile, first, last, single, skip and tap Operators

Many streams are long or never end. These operators decide which values you keep and when to stop. They are the everyday tools of RxJS.

What you will learn

• tap for side effects
• take, takeUntil, takeWhile, takeLast
• first, last, single
• skip, skipWhile, skipUntil, skipLast
• distinctUntilChanged

The idea in one picture

Imagine values arriving on a conveyor belt: 1, 2, 3, 4, 5, 6.

• take operators grab the values you want from the start (or end), then stop.
• skip operators let some values pass by unused.
• first, last, single pick one value.
• tap peeks at values without changing them.

tap: look without touching

tap runs a function for each value and passes the value along unchanged. Use it for logging, debugging and small side effects such as turning a loading flag off:

import { of, tap, map } from 'rxjs';

of(1, 2, 3)
  .pipe(
    tap((v) => console.log('before map', v)),
    map((v) => v * 10),
    tap({ next: (v) => console.log('after map', v), complete: () => console.log('done') }),
  )
  .subscribe();

Do not put important logic in tap. Logic that other code depends on belongs in map, subscribe or a service.

The take family

// src/app/rxjs3/take-demo.ts
import { Subject, from, interval, takeLast, takeUntil, takeWhile, take, toArray, lastValueFrom } from 'rxjs';

const collect = <T>(source: import('rxjs').Observable<T>) => lastValueFrom(source.pipe(toArray()));

export const numbers$ = from([1, 2, 3, 4, 5, 6]);

export const firstTwo = () => collect(numbers$.pipe(take(2))); // [1, 2]
export const whileSmall = () => collect(numbers$.pipe(takeWhile((n) => n < 4))); // [1, 2, 3]
export const lastTwo = () => collect(numbers$.pipe(takeLast(2))); // [5, 6]

// takeUntil stops when another Observable (the "notifier") emits.
// This is the classic way to stop a stream when a component is destroyed.
export function tickUntil(stop$: Subject<void>) {
  return interval(1000).pipe(takeUntil(stop$));
}

• `take(n)`: the first n values, then completes. It also unsubscribes from the source, which makes take(1) a standard way to read one value.
• `takeWhile(predicate)`: values until the test first fails, then completes. Pass true as the second argument to include the failing value.
• `takeLast(n)`: waits for the stream to complete, then emits the last n values. It never emits on an endless stream.
• `takeUntil(notifier$)`: values until notifier$ emits. Put `takeUntil` last in the pipe, so that inner subscriptions created by earlier operators are cleaned up too.

In modern Angular, takeUntilDestroyed() replaces the manual destroy$ subject for components.

first, last and single

// src/app/rxjs3/pick-demo.ts
import { EMPTY, first, from, last, lastValueFrom, single } from 'rxjs';

export const firstEven$ = from([1, 2, 3, 4]).pipe(first((n) => n % 2 === 0)); // 2
export const lastEven$ = from([1, 2, 3, 4]).pipe(last((n) => n % 2 === 0)); // 4
export const onlyBig$ = from([1, 2, 30]).pipe(single((n) => n > 10)); // 30

export const safeFirst$ = EMPTY.pipe(first(undefined, 'default')); // 'default'
export const unsafeFirst = () => lastValueFrom(EMPTY.pipe(first()));

• first() emits the first value and completes. With a predicate, the first value that passes.
• last() waits for completion and emits the last value (that passes).
• single() expects exactly one matching value. If there are none, or several, it throws an error.
• If the stream is empty and you did not give a default value, first() and last() throw EmptyError. Pass a default: first(undefined, 'default').
• take(1) is similar to first(), but never errors on an empty stream. It just completes.

The skip family

// src/app/rxjs3/skip-demo.ts
import { Subject, from, lastValueFrom, skip, skipLast, skipUntil, skipWhile, toArray } from 'rxjs';

const collect = <T>(source: import('rxjs').Observable<T>) => lastValueFrom(source.pipe(toArray()));
const numbers$ = from([1, 2, 3, 4, 5, 6]);

export const dropTwo = () => collect(numbers$.pipe(skip(2))); // [3, 4, 5, 6]
export const dropWhileSmall = () => collect(numbers$.pipe(skipWhile((n) => n < 4))); // [4, 5, 6]
export const dropLastTwo = () => collect(numbers$.pipe(skipLast(2))); // [1, 2, 3, 4]

// skipUntil ignores values until the notifier emits
export function afterStart(source$: Subject<number>, start$: Subject<void>) {
  return source$.pipe(skipUntil(start$));
}

• `skip(n)` ignores the first n values. Useful for ignoring the initial value of a BehaviorSubject or form control.
• `skipWhile(predicate)` ignores values while the test is true, and lets everything through after it first fails.
• `skipLast(n)` holds back the last n values, so it can only emit when more arrive.
• `skipUntil(notifier$)` ignores values until the notifier emits once.

distinctUntilChanged

distinctUntilChanged() ignores a value that is the same as the previous one. It prevents duplicate work, for example a search for the same text twice in a row. For objects, pass a comparison: distinctUntilChanged((a, b) => a.id === b.id).

Test every claim

// src/app/rxjs3/operators.spec.ts
import { Subject, distinctUntilChanged, from, lastValueFrom, of, single, tap, toArray, EmptyError } from 'rxjs';
import { firstTwo, lastTwo, tickUntil, whileSmall } from './take-demo';
import { firstEven$, lastEven$, onlyBig$, safeFirst$, unsafeFirst } from './pick-demo';
import { afterStart, dropLastTwo, dropTwo, dropWhileSmall } from './skip-demo';

describe('take, skip and pick operators', () => {
  it('take family', async () => {
    expect(await firstTwo()).toEqual([1, 2]);
    expect(await whileSmall()).toEqual([1, 2, 3]);
    expect(await lastTwo()).toEqual([5, 6]);
  });

  it('takeUntil stops the interval', () => {
    vi.useFakeTimers();
    const stop$ = new Subject<void>();
    const ticks: number[] = [];
    let completed = false;
    tickUntil(stop$).subscribe({ next: (n) => ticks.push(n), complete: () => (completed = true) });
    vi.advanceTimersByTime(2500);
    stop$.next();
    vi.advanceTimersByTime(5000);
    expect(ticks).toEqual([0, 1]);
    expect(completed).toBe(true);
    vi.useRealTimers();
  });

  it('first, last, single', async () => {
    expect(await lastValueFrom(firstEven$)).toBe(2);
    expect(await lastValueFrom(lastEven$)).toBe(4);
    expect(await lastValueFrom(onlyBig$)).toBe(30);
    expect(await lastValueFrom(safeFirst$)).toBe('default');
    await expect(unsafeFirst()).rejects.toBeInstanceOf(EmptyError);
  });

  it('single fails when several values match', async () => {
    await expect(lastValueFrom(from([20, 30]).pipe(single((n) => n > 10)))).rejects.toThrow();
  });

  it('skip family', async () => {
    expect(await dropTwo()).toEqual([3, 4, 5, 6]);
    expect(await dropWhileSmall()).toEqual([4, 5, 6]);
    expect(await dropLastTwo()).toEqual([1, 2, 3, 4]);
  });

  it('skipUntil ignores values until the notifier emits', () => {
    const source$ = new Subject<number>();
    const start$ = new Subject<void>();
    const seen: number[] = [];
    afterStart(source$, start$).subscribe((v) => seen.push(v));
    source$.next(1);
    source$.next(2);
    start$.next();
    source$.next(3);
    expect(seen).toEqual([3]);
  });

  it('distinctUntilChanged removes repeats, tap does not change values', async () => {
    const log: number[] = [];
    const result = await lastValueFrom(
      of(1, 1, 2, 2, 2, 1).pipe(distinctUntilChanged(), tap((v) => log.push(v)), toArray()),
    );
    expect(result).toEqual([1, 2, 1]);
    expect(log).toEqual([1, 2, 1]);
  });
});

Use takeUntilDestroyed in a component

// src/app/rxjs3/clock.ts
import { Component, signal } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { interval, map } from 'rxjs';

@Component({
  selector: 'app-clock',
  template: `<p>Seconds on this page: {{ seconds() }}</p>`,
})
export class Clock {
  protected readonly seconds = signal(0);

  constructor() {
    interval(1000)
      .pipe(map((n) => n + 1), takeUntilDestroyed())
      .subscribe((n) => this.seconds.set(n));
  }
}

// src/app/app.ts
import { Component } from '@angular/core';
import { Clock } from './rxjs3/clock';

@Component({
  imports: [Clock],
  selector: 'app-root',
  templateUrl: './app.html',
  styleUrl: './app.css',
})
export class App {}

<!-- src/app/app.html -->
<main>
  <app-clock />
</main>

Try it yourself

1. Use take(5) and scan to sum the first five numbers of an interval.
2. Make a button that triggers a Subject and stops a ticking clock with takeUntil.
3. Replace .pipe(first()) with .pipe(take(1)) on an empty stream and compare.

Quick check

1. Which operator completes an endless stream after three values?
2. Why can first() throw an error?
3. Where should takeUntil go in a pipe?

Next, learn to accumulate, delay and debounce in scan, reduce, debounce and delay.