Building a pipeline

From a list of tasks, select the first two unfinished tasks and trim the whitespace around their titles. Use separate operators to select items, transform them, and limit the count.

Operator order

import { from, type Operator } from "mogami-stream";

type Task = { title: string; done: boolean };

const tasks: Task[] = [
  { title: " Review draft ", done: false },
  { title: "Send invoice", done: true },
  { title: " Book room ", done: false },
  { title: "Update calendar", done: false },
];

const titles = from(tasks)
  .filter((task) => !task.done)
  .map((task) => task.title.trim())
  .take(2);

const result = await titles.toArray();
console.log(result); // ["Review draft", "Book room"]

filter selects unfinished tasks, map trims their titles, and take(2) limits the result to two items. Completed tasks never reach map, so their titles are not processed.

Operators are applied in the order of the method calls. Here, map changes each task into a string, so task.done is no longer available afterward. Filter by task state before that transformation.

Values and types at each stage

In Stream<T>, T is the type of the value at that stage. Following each stage's output tells you what the next callback receives.

Stage	Value	Type
from(tasks)	A task object	Stream<Task>
filter(...)	An unfinished task object	Stream<Task>
map(...)	A trimmed title	Stream<string>
take(2)	At most two titles	Stream<string>
.toArray()	An array of titles	Promise<string[]>

In this example, filter selects items, while map changes both the value and its type. take leaves the value type unchanged. After await, result is a string[].

take placement and item counts

take(2) counts values reaching its position in the pipeline. To select two unfinished tasks, put it after filter, as in the first example. If fewer than two unfinished tasks exist, the result contains fewer than two items.

Limiting the input first considers only the first two entries in the original array:

const firstTwoInputs = from(tasks)
  .take(2)
  .filter((task) => !task.done)
  .map((task) => task.title.trim());

console.log(await firstTwoInputs.toArray()); // ["Review draft"]

The second task is already complete, so the result contains only "Review draft". The later "Book room" task is not read. Place take according to whether you want to limit input items or matching items.

Array collection and per-item processing

Use .toArray() when you need the results together in an array. Here, take(2) limits that array to two items. Without a count limit, make sure all results can fit in memory.

If you do not need an array, process the same titles stream one item at a time with forEach:

await titles.forEach((title) => console.log(title));
// Review draft
// Book room

This call starts a separate execution that reads the tasks array again. forEach does not collect a result array and waits for each callback to finish, including asynchronous callbacks.

See Terminal consumers for aggregation and predicate-based early termination, or Concurrency, capacity, and demand for concurrent asynchronous work. For external I/O, see Cancellation and cleanup for stopping work and releasing resources.

.pipe() and custom operators

Use .pipe(a, b) for multiple custom operators. Each takes the preceding source and returns a new stream; fluent methods can continue afterward.

function trimTitles(): Operator<string, string> {
  return (source) => from(source).map((title) => title.trim());
}

function titleLengths(): Operator<string, number> {
  return (source) => from(source).map((title) => title.length);
}

const lengths = from(tasks)
  .filter((task) => !task.done)
  .map((task) => task.title)
  .pipe(trimTitles(), titleLengths())
  .take(2);

console.log(await lengths.toArray()); // [12, 9]

Entry points and type inference

mogami-stream/core provides standalone operators, .pipe(), and terminal methods. It omits fluent transformation methods such as .map(), allowing bundlers to remove unused operators.

import { from, map, take } from "mogami-stream/core";

const values = await from([1, 2, 3])
  .pipe(
    map((value) => value * 2),
    take(2),
  )
  .toArray();

console.log(values); // [2, 4]

Both entries share execution and cleanup. Pass a core stream to the default entry's from() to use fluent transformations; importing that entry alone does not change existing streams.

Fluent chains have no stage limit. Each .pipe() or standalone pipe call infers up to eight operators; split longer pipelines across calls. Value-independent operators such as const first = take(1) remain reusable across element types.