@frostnode/waitfor

Predicate-based wait and poll operator for RxJS. Drop it into any pipe to re-subscribe to a completed source on a configurable schedule, with rich timing and retry strategies, optional pause control, and tab-visibility awareness.

Features

• Two polling types: repeat and interval for different completion semantics
• Timing strategies: constant, linear, exponential, random and dynamic (state-driven)
• Pause controls: external notifier and/or whenHidden (browser tab visibility)
• Flexible retries: cap attempts on consecutive errors or on total errors
• Input validation: guards against bad time inputs
• Cross-platform: browser and Node.js
• RxJS compatibility: RxJS v7 and v8
• Module formats: CJS, ESM, and UMD

Installation

npm install @frostnode/waitfor --save

rxjs is a peer dependency. In NestJS apps it is already present.

Why

Polling is the right tool when a source does not push updates: HTTP endpoints that report status, on-chain transaction receipts, Redis Streams entries, order-book snapshots between match cycles, queue depth probes. The operator re-subscribes to your source after each completion, distinguishes "normal delay between polls" from "error retry", and lets you pause emissions on demand. First emission and in-flight cycle completion are preserved across pauses.

Usage

Default Configuration

Plug-and-play — just add the operator to your pipe.

import { poll } from '@frostnode/waitfor';
import { takeWhile } from 'rxjs';

request$
  .pipe(
    poll(), // Poll every second with exponential retry strategy (7s max)
    takeWhile(({ status }) => status !== 'done', true)
  )
  .subscribe({ next: console.log });

Strategy-Based Configuration

Use built-in strategies for easy timing control.

import { poll } from '@frostnode/waitfor';
import { takeWhile } from 'rxjs';

request$
  .pipe(
    poll({
      type: 'interval', // Drops uncompleted source after delay
      delay: {
        strategy: 'random',
        time: [1000, 3000], // Random delay between 1 and 3 seconds
      },
      retry: {
        limit: Infinity, // Will never throw
      },
    }),
    takeWhile(({ status }) => status !== 'done', true)
  )
  .subscribe({ next: console.log });

Advanced Dynamic Strategies

Implement complex polling strategies with dynamic timing based on poll state.

import { poll } from '@frostnode/waitfor';
import { takeWhile } from 'rxjs';

request$
  .pipe(
    poll({
      delay: {
        /* Adaptive polling based on response data */
        strategy: 'dynamic',
        time: ({ value }) => (value?.items.length ? 1000 : 500),
      },
      retry: {
        /* Custom exponential backoff with jitter */
        strategy: 'dynamic',
        time: ({ consecutiveRetryCount }) => {
          const exponential = Math.pow(2, consecutiveRetryCount - 1) * 1000;
          const jitter = Math.random() * 200;

          return exponential + jitter;
        },
        limit: 6,
      },
    }),
    takeWhile(({ status }) => status !== 'done', true)
  )
  .subscribe({ next: console.log });

Pause With Notifier

Control polling from outside by passing an Observable<boolean> as pause.notifier: emit true to pause and false to resume. If the notifier never emits, polling starts (same as resume). To start paused, use an observable that emits true initially (e.g. new BehaviorSubject(true)). You can combine it with whenHidden: true so both your stream and tab visibility affect pausing. If the notifier observable errors, the error is caught: polling does not error or complete because of notifier failures, and the previous pause state is kept.

import { poll } from '@frostnode/waitfor';
import { fromEvent } from 'rxjs';
import { scan, takeWhile } from 'rxjs';

const click$ = fromEvent(document, 'click').pipe(
  scan((isPaused) => !isPaused, false)
);

request$
  .pipe(
    poll({
      pause: {
        notifier: click$,
        whenHidden: false, // set true (default) to also pause when tab hidden
      },
    }),
    takeWhile(({ status }) => status !== 'done', true)
  )
  .subscribe({ next: console.log });

// Now, click anywhere on the page to toggle pause/resume

API Reference

poll(config?: PollConfig)

Creates a polling operator that manages the source observable's subscription lifecycle.

PollConfig

interface PollConfig {
  /**
   * Defines the polling behavior:
   * - 'repeat': Polls after current source completes
   * - 'interval': Polls in intervals, dropping any ongoing source operations
   * @default 'repeat'
   */
  type?: 'repeat' | 'interval';

  /**
   * Configuration for polling delays (between successful operations)
   */
  delay?: {
    /**
     * Strategy type for delay timing. Built-in strategies (except dynamic)
     * calculate time per state's `pollCount`.
     * @default 'constant'
     */
    strategy: 'constant' | 'linear' | 'exponential' | 'random' | 'dynamic';

    /**
     * Time (ms) depending on strategy:
     * - constant: number
     * - linear: number
     * - exponential: number
     * - random: [min, max]
     * - dynamic: (state) => number | [min, max]
     * @default 1000
     */
    time:
      | number
      | [min: number, max: number]
      | (state: PollState) => number | [min: number, max: number];
  };

  /**
   * Configuration for retry behavior (on errors)
   */
  retry?: {
    /**
     * Strategy type for retry timing. Built-in strategies (except dynamic)
     * calculate time per state:
     * - consecutiveOnly: true → uses `consecutiveRetryCount`
     * - consecutiveOnly: false → uses `retryCount`
     * @default 'exponential'
     * @note Required if `time` is provided, otherwise uses default
     */
    strategy: 'constant' | 'linear' | 'exponential' | 'random' | 'dynamic';

    /**
     * Time (ms) depending on strategy:
     * - constant: number
     * - linear: number
     * - exponential: number
     * - random: [min, max]
     * - dynamic: (state) => number | [min, max]
     * @default 1000
     * @note Required if `strategy` is provided, otherwise uses default
     */
    time:
      | number
      | [min: number, max: number]
      | (state: PollState) => number | [min: number, max: number];

    /**
     * Maximum number of retry attempts before throwing an error.
     * Use `Infinity` to keep retrying indefinitely.
     * @default 3
     */
    limit?: number;

    /**
     * Controls how retries are counted:
     * - true: Only consecutive errors count toward retry limit
     *   (resets counter on success)
     * - false: All errors count toward retry limit regardless of
     *   successful responses between them
     * @default true
     */
    consecutiveOnly?: boolean;
  };


  /**
   * Configuration for pause behavior
   */
  pause?: {
    /**
     * Observable that emits true to pause, false to resume
     * @default false
     * @note
     * - Interrupts polling/retrying cycles
     * - Can pause first emission
     * - Defaults to false if no initial emission
     * - Errors from this notifier are caught and do NOT error/complete polling
     */
    notifier?: Observable<boolean>;
    /**
     * [Browser only] When true, polling pauses if tab isn't visible
     * @default true
     * @note
     * - Polling/retrying cycles finish before pausing
     * - First emission guaranteed
     */
    whenHidden?: boolean;
  };
}

PollState

State object passed to delay/retry time producer functions:

interface PollState<T> {
  /** Latest value from the source. For `interval` polling type,
   * first emission is undefined. */
  value: T | undefined;

  /** Latest error when retrying */
  error: any | undefined;

  /** Total number of successful poll operations */
  pollCount: number;

  /** Total number of retry attempts */
  retryCount: number;

  /** Current number of consecutive retry attempts */
  consecutiveRetryCount: number;
}

/** Note: pollCount + retryCount = total attempts */

Contributing

Issues and pull requests are welcome. Check open issues before opening a duplicate.

License

MIT