RxJS Operators

Custom RxJS operators for reactive programming. These operators extend RxJS with additional functionality for common patterns like filtering, error handling, and debugging.

Installation

npm install @dasch-ng/rxjs-operators

Available Operators

filterNil

Filters out null and undefined values from the stream with proper TypeScript type narrowing.

Use case: Clean data streams by removing nil values while maintaining type safety.

Example

import { of } from 'rxjs';
import { filterNil } from '@dasch-ng/rxjs-operators';

of(1, null, 2, undefined, 3).pipe(filterNil()).subscribe(console.log);
// Output: 1, 2, 3

View API Documentation →

---

filterEmpty

Filters out empty strings from the stream.

Use case: Remove empty strings from user input or form values.

Example

import { of } from 'rxjs';
import { filterEmpty } from '@dasch-ng/rxjs-operators';

of('hello', '', 'world', '  ', 'foo').pipe(filterEmpty()).subscribe(console.log);
// Output: 'hello', 'world', 'foo'

View API Documentation →

---

filterError

Filters out error values from the stream, only allowing successful emissions.

Use case: Handle errors gracefully in observable chains without terminating the stream.

Example

import { of, throwError } from 'rxjs';
import { filterError } from '@dasch-ng/rxjs-operators';

of(1, 2, 3)
  .pipe(
    mergeMap((x) => (x === 2 ? throwError(() => new Error('Error!')) : of(x))),
    filterError(),
  )
  .subscribe(console.log);
// Output: 1, 3 (error is filtered out)

View API Documentation →

---

tapCatch

Combines tap and catchError operators into a single convenient operator.

Use case: Execute side effects on successful emissions and handle errors in one operation.

Example

import { of, throwError } from 'rxjs';
import { tapCatch } from '@dasch-ng/rxjs-operators';

throwError(() => new Error('Oops!'))
  .pipe(
    tapCatch({
      next: (value) => console.log('Got value:', value),
      error: (err, caught) => {
        console.error('Error occurred:', err);
        return of('fallback value');
      },
    }),
  )
  .subscribe(console.log);
// Console output:
// Error occurred: Error: Oops!
// fallback value

View API Documentation →

---

debugOperator

Debug operator for logging observable emissions, errors, and completions.

Use case: Debugging observable streams during development.

Example

import { of } from 'rxjs';
import { debugOperator } from '@dasch-ng/rxjs-operators';

of(1, 2, 3).pipe(debugOperator('MyStream')).subscribe();
// Console output:
// [MyStream] Next: 1
// [MyStream] Next: 2
// [MyStream] Next: 3
// [MyStream] Complete

View API Documentation →

---

filterString

Filters out non-string values and optionally empty strings.

Use case: Ensure only valid string values pass through the stream.

Example

import { of } from 'rxjs';
import { filterString } from '@dasch-ng/rxjs-operators';

of('hello', 123, 'world', null, '', 'foo').pipe(filterString()).subscribe(console.log);
// Output: 'hello', 'world', 'foo'

View API Documentation →

---

pollUntil

Reads a value repeatedly until it is ready, emits it once and completes. The first read is synchronous, so an already available value is emitted right away. Errors with rxjs' TimeoutError after timeout ms.

Use case: Wait for something initialised outside your control, such as an auth token or a third-party SDK global, before sending requests.

Options: interval (default 100), timeout (default 10000, Infinity waits forever), ready (default: not null, undefined, '' or false).

Example

import { switchMap } from 'rxjs';
import { pollUntil } from '@dasch-ng/rxjs-operators';

pollUntil(() => auth.token, { timeout: 5000 })
  .pipe(switchMap((token) => http.get(url, { headers: { Authorization: `Bearer ${token}` } })))
  .subscribe();

pollUntil(() => window.someSdk, { interval: 250 }).subscribe((sdk) => sdk.init());

Common Patterns

Combining Operators

RxJS operators can be combined in powerful ways:

import { of } from 'rxjs';
import { map } from 'rxjs/operators';
import { filterNil, tapCatch } from '@dasch-ng/rxjs-operators';

const userInput$ = of('  hello  ', null, '  world  ', undefined);

userInput$
  .pipe(
    filterNil(), // Remove null/undefined
    map((s) => s.trim()), // Trim whitespace
    tapCatch({
      next: (value) => console.log('Processing:', value),
      error: (err) => of(''),
    }),
  )
  .subscribe((result) => {
    console.log('Result:', result);
  });

Error Handling

Use tapCatch for graceful error handling with logging:

import { tapCatch } from '@dasch-ng/rxjs-operators';
import { of } from 'rxjs';

apiCall$
  .pipe(
    tapCatch({
      next: (data) => {
        analytics.track('api_success', data);
      },
      error: (err) => {
        logger.error('API call failed', err);
        return of({ data: [], error: err.message });
      },
    }),
  )
  .subscribe(handleResponse);

Stream Debugging

Use debugOperator to understand complex observable chains:

import { debugOperator } from '@dasch-ng/rxjs-operators';

complexObservable$
  .pipe(debugOperator('Before Map'), map(transformData), debugOperator('After Map'), filter(isValid), debugOperator('After Filter'))
  .subscribe();

Type Safety

All operators are fully typed and provide proper TypeScript inference:

import { of } from 'rxjs';
import { filterNil } from '@dasch-ng/rxjs-operators';

const numbers$: Observable<number | null> = of(1, null, 2);

const filtered$ = numbers$.pipe(filterNil());
// Type is now Observable<number> (null is removed from type)

filtered$.subscribe((num) => {
  // num is number, not number | null
  console.log(num.toFixed(2)); // No TypeScript error
});

API Reference

For complete API documentation with all parameters and return types, see the API Reference.

Source Code

View the source code on GitHub.

Contributing

Found a bug or want to contribute? Check out the contributing guidelines.