createAsyncController

Overview

createAsyncController provides switchMap-based request orchestration with built-in status tracking and an AbortController for cancellation. When a new request is triggered via execute(), the previous in-flight request is automatically cancelled. This makes it ideal for search-as-you-type, data fetching, and any scenario where only the latest request matters.

Import

import { createAsyncController } from "rxjs-ai";

Basic usage

import { createAsyncController } from "rxjs-ai";
import { Observable } from "rxjs";

interface SearchResult {
  id: string;
  title: string;
}

// The executor is passed at construction time.
// It receives the request and an AbortSignal, and returns an Observable.
const search = createAsyncController<string, SearchResult[]>(
  (query, signal) =>
    new Observable((subscriber) => {
      fetch(`/api/search?q=${query}`, { signal })
        .then((res) => res.json())
        .then((data) => {
          subscriber.next(data);
          subscriber.complete();
        })
        .catch((err) => subscriber.error(err));
    }),
);

// Subscribe to state changes
search.state$.subscribe((state) => {
  console.log(state.status); // "idle" | "loading" | "success" | "error" | "cancelled"
  console.log(state.data);   // SearchResult[] | null
  console.log(state.error);  // unknown | null
});

// Execute a request — just pass the request value
search.execute("rxjs");

// A second call auto-cancels the first
search.execute("rxjs-ai");

// Manual cancellation
search.cancel();

// Clean up
search.destroy();

State machine

The controller tracks five discrete statuses throughout the request lifecycle:

┌──────┐  execute()  ┌─────────┐  success  ┌─────────┐
│ idle │────────────▶│ loading │──────────▶│ success │
└──────┘             └─────────┘           └─────────┘
                        │   │
               cancel() │   │ error
                        ▼   ▼
                 ┌───────────┐  ┌───────┐
                 │ cancelled │  │ error │
                 └───────────┘  └───────┘

Status	Description
idle	Initial state. No request has been made yet.
loading	A request is in-flight.
success	The most recent request completed successfully. state.data contains the result.
error	The most recent request failed. state.error contains the rejection reason.
cancelled	The most recent request was cancelled, either manually via cancel() or automatically by a subsequent execute() call.

Auto-cancel behaviour

Calling execute() while a request is already in-flight will:

1. Abort the previous request’s AbortSignal.
2. Transition the state to cancelled for the previous request.
3. Immediately begin the new request, transitioning the state to loading.

This uses switchMap internally, so only the latest request’s result is emitted.

API

createAsyncController(executor, options?)

Creates a new async controller with the given executor function.

Argument	Type	Description
executor	(request: Request, signal: AbortSignal) => Observable<Data>	Required. The function that performs the async operation. Receives the request value and an AbortSignal, and must return an Observable<Data>.
options	CreateAsyncControllerOptions	Optional configuration object.

Returned handle — AsyncController

Member	Signature	Description
state$	Observable<AsyncState<Data, ErrorType>>	Emits the current async state on every transition. Replays the latest value to new subscribers.
execute	(request: Request) => void	Triggers a new async operation by passing the request to the executor. If a previous operation is in-flight it is automatically cancelled.
cancel	() => void	Manually cancels the current in-flight request and transitions the state to cancelled.
destroy	() => void	Cancels any in-flight request, completes all internal Subjects, and releases resources.

Types

/** Possible statuses of an async operation. */
type AsyncStatus = "idle" | "loading" | "success" | "error" | "cancelled";

/** The state emitted by the controller's state$ Observable. */
interface AsyncState<Data, ErrorType = unknown> {
  status: AsyncStatus;
  data: Data | null;
  error: ErrorType | null;
  updatedAt: number | null;
}

/** The controller instance returned by createAsyncController. */
interface AsyncController<Request, Data, ErrorType = unknown> {
  state$: Observable<AsyncState<Data, ErrorType>>;
  execute: (request: Request) => void;
  cancel: () => void;
  destroy: () => void;
}