# @effectionx/stream-helpers

> Type-safe stream operators like filter, map, reduce, and forEach

Latest version **0.8.3** (published 2026-06-16) · MIT license · 0 weekly downloads

## Install

```sh
npm install @effectionx/stream-helpers
pnpm add @effectionx/stream-helpers
yarn add @effectionx/stream-helpers
bun add @effectionx/stream-helpers
```

## Health

**Score 65/100 (B)** — status: active.

Positive: has types; esm support; no vulnerabilities; has provenance; high maintenance score.

Warnings: low downloads; pre 1.0.

## Facts

| | |
|---|---|
| Version | 0.8.3 |
| Published | 2026-06-16 |
| First published | 2025-05-19 |
| Weekly downloads | 0 |
| License | MIT |
| TypeScript types | bundled |
| Module format | ESM + CommonJS |
| Dependencies | 4 |
| Unpacked size | 192.6 KB |
| Known vulnerabilities | 0 |
| Install scripts | no |
| Provenance | attested (GitHub Actions) |
| GitHub stars | 12 |
| Author | engineering@frontside.com |
| Maintainers | frontsidejack |
| Keywords | streams |

## Links

- npm: https://www.npmjs.com/package/@effectionx/stream-helpers
- Repository: https://github.com/thefrontside/effectionx
- Homepage: https://github.com/thefrontside/effectionx#readme
- Issues: https://github.com/thefrontside/effectionx/issues
- npm.io page: https://npm.io/package/@effectionx/stream-helpers

## Dependencies (4)

- [remeda](https://npm.io/package/remeda.md) ^2
- [immutable](https://npm.io/package/immutable.md) ^5
- [@effectionx/signals](https://npm.io/package/@effectionx/signals.md) 0.5.4
- [@effectionx/timebox](https://npm.io/package/@effectionx/timebox.md) 0.4.3

## Alternatives

- [byte-size](https://npm.io/package/byte-size.md) — 2.1M weekly downloads
- [speed-limiter](https://npm.io/package/speed-limiter.md) — 16.0K weekly downloads
- [@powersync/node](https://npm.io/package/@powersync/node.md) — 10.9K weekly downloads
- [@ledgerhq/coin-cardano](https://npm.io/package/@ledgerhq/coin-cardano.md) — 1.0K weekly downloads
- [@jayesol/jayeson.lib.streamfinder](https://npm.io/package/@jayesol/jayeson.lib.streamfinder.md) — 1.0K weekly downloads

## Recent versions

- 0.8.3 (latest) — 2026-06-16
- 0.8.2 — 2026-03-22
- 0.8.1 — 2026-02-23
- 0.8.0 — 2026-02-20
- 0.7.3 — 2026-02-15
- 0.7.2 — 2026-02-12
- 0.7.1 — 2026-01-24
- 0.6.1 — 2026-01-21
- 0.6.0 — 2026-01-18
- 0.5.1 — 2026-01-06
- 0.4.1 — 2025-12-08
- 0.4.0 — 2025-12-08
- 0.3.0 — 2025-10-18
- 0.2.0 — 2025-08-22
- 0.1.1 — 2025-08-08
- … 1 more at https://npm.io/package/@effectionx/stream-helpers/versions

## README

# Stream Helpers

A collection of type-safe stream helpers built on top of
[Effection](https://github.com/thefrontside/effection) for efficient and
controlled stream processing.

## Included Helpers

### Filter

The `filter` helper narrows a stream by only passing through items that match a
predicate.

```typescript
import { filter } from "@effectionx/stream-helpers";
import { each } from "effection";

// Example: Synchronous filtering
function* syncExample(source: Stream<number, unknown>) {	

  const gt5 = filter<number>(function* (x) { return x > 5 });

  for (const value of yield* each(gt5(stream))) {
    console.log(value); // Only values > 5
    yield* each.next();
  }
};

// Example: Asynchronous filtering
function* asyncExample(source: Stream<number, unknown>) {

  const evensOf = filter<number>(function* (x) {
    yield* sleep(100); // Simulate async operation
    return x % 2 === 0; // Keep only even numbers
  });

  for (const value of yield* each(evensOf(stream))) {
    console.log(value); // Only even numbers
    yield* each.next();
  }
});
```

### Map

The `map` helper transforms each item in a stream using a provided function.
This is useful for data transformation operations where you need to process each
item individually.

```typescript
import { map } from "@effectionx/stream-helpers";
import { each } from "effection";

function* example(stream: Stream<number, unknown>) {
  const double = map<number>(function* (x) {
    return x * 2;
  });

  for (const value of yield* each(double(stream))) {
    console.log(value); // Each value is doubled
    yield* each.next();
  }
}
```

### Reduce

The `reduce` helper transforms each item in a stream by applying it to
an accumulated value.

```typescript
import { reduce } from "@effectionx/stream-helpers";
import { each } from "effection";

function* example(stream: Stream<number, unknown>) {
  let sum = reduce(function* (total: number, current: number) {
    return total + current;
  }, 0);

  for (let value of yield* each(sum(streamOf([1,2,3]))) {
    console.log(value) // logs 1 -> 3 -> 6
    yield* each.next();
  }
}
```

### Batch

The `batch` helper is useful when you want to convert individual items passing
through the stream into arrays of items. The batches can be created either by
specifying a maximum time or a maximum size. If both are specified, the batch
will be created when either condition is met.

```typescript
import { batch } from "@effectionx/stream-helpers";
import { each } from "effection";

// Example: Batch by size
function* exampleBySize(stream: Stream<number, unknown>) {
  const byThree = batch({ maxSize: 3});

  for (const items of yield* each(byThree(stream))) {
    console.log(batch); // [1, 2, 3], [4, 5, 6], ...
    yield* each.next();
  }
};

// Example: Batch by time
function* exampleByTime(stream: Stream<number, unknown>) {
  const stream = batch({ maxTime: 1000 })(sourceStream);

  for (const batch of yield* each(stream)) {
    console.log(batch); // Items received within 1 second
    yield* each.next();
  }
});

// Example: Combined batching
function* exampleCombined(stream: Stream<number, unknown>) {

  const batched = batch({
    maxSize: 5,
    maxTime: 1000,
  });

  for (const batch of yield* each(batched(stream))) {
    console.log(batch); // Up to 5 items within 1 second
    yield* each.next();
  }
});
```

### Valve

Allows to apply backpressure to the source stream to prevent overwhelming the
downstream consumer. This is useful with any stream that generates items faster
than the consumer can consume them. It was originally designed for use with
Kafka where the producer can cause the service to run out of memory when the
producer produces many faster than the consumer to process the messages. It can
be used as a buffer for any infinite stream.

```typescript
import { valve } from "@effectionx/stream-helpers";
import { each } from "effection";

function* example() {
  const regulated = valve({
    // buffer size threshold when close operation will invoked
    closeAt: 1000,
    *close() {
      // pause the source stream
    },

    // buffer size threshold when open operation will be invoked
    openAt: 100,
    *open() {
      // resume the source stream
    },
  })(stream);

  for (const value of yield* each(regulated)) {
    console.log(value);
    yield* each.next();
  }
}
```

### Drain

The `drain` helper exhausts a stream, discarding all yielded values, and returns
the close value. This is useful when you only care about the final result of a
stream, not the intermediate values.

```typescript
import { drain } from "@effectionx/stream-helpers";

function* example() {
  // Get the response from a request channel (ignoring any progress)
  const channel = yield* transport.send(request);
  const response = yield* drain(channel);
  console.log(response); // The close value
}
```

### First

The `first` helper returns the first value yielded by a stream, or `undefined`
if the stream closes without yielding any values. Use `first.expect()` to throw
an error instead of returning `undefined`.

```typescript
import { first } from "@effectionx/stream-helpers";
import { streamOf } from "@effectionx/stream-helpers";

function* example() {
  const stream = streamOf([1, 2, 3]);
  const value = yield* first(stream);
  console.log(value); // 1

  const empty = streamOf([]);
  const none = yield* first(empty);
  console.log(none); // undefined

  // Use first.expect() to throw if stream is empty
  const required = yield* first.expect(stream); // throws if empty
}
```

### Last

The `last` helper returns the last value yielded by a stream, or `undefined`
if the stream closes without yielding any values. It exhausts the entire stream
to find the last value. Use `last.expect()` to throw an error instead of
returning `undefined`.

```typescript
import { last } from "@effectionx/stream-helpers";
import { streamOf } from "@effectionx/stream-helpers";

function* example() {
  const stream = streamOf([1, 2, 3]);
  const value = yield* last(stream);
  console.log(value); // 3

  const empty = streamOf([]);
  const none = yield* last(empty);
  console.log(none); // undefined

  // Use last.expect() to throw if stream is empty
  const required = yield* last.expect(stream); // throws if empty
}
```

### Take

The `take` helper creates a stream transformer that yields the first `n` values
from the source stream, then closes with the last taken value.

```typescript
import { take, forEach, streamOf } from "@effectionx/stream-helpers";
import { pipe } from "remeda";

function* example() {
  const stream = streamOf([1, 2, 3, 4, 5]);
  // yields 1, 2, then closes with 3
  const closeValue = yield* forEach(function* (value) {
    console.log(value); // 1, then 2
  }, take(3)(stream));
  console.log(closeValue); // 3
}

// Works with pipe
const limited = pipe(source, take(3));
```

### TakeWhile

The `takeWhile` helper creates a stream transformer that yields values while
the predicate returns true. When the predicate returns false, the stream closes
immediately (the failing value is not included).

```typescript
import { takeWhile, forEach, streamOf } from "@effectionx/stream-helpers";
import { pipe } from "remeda";

function* example() {
  const stream = streamOf([1, 2, 3, 4, 5]);
  // yields 1, 2 (stops when value >= 3)
  yield* forEach(function* (value) {
    console.log(value); // 1, then 2
  }, takeWhile((x) => x < 3)(stream));
}

// Works with pipe
const filtered = pipe(source, takeWhile((x) => x.isValid));
```

### TakeUntil

The `takeUntil` helper creates a stream transformer that yields values until
the predicate returns true. When the predicate matches, the stream closes with
the matching value. This is useful for "iterate until a condition is met"
patterns.

```typescript
import { takeUntil, forEach } from "@effectionx/stream-helpers";
import { pipe } from "remeda";

function* example() {
  // Iterate validation progress until we get a terminal status
  const result = yield* forEach(
    function* (progress) {
      showSpinner(progress.status);
    },
    takeUntil((p) => p.status === "valid" || p.status === "invalid")(
      validationStream,
    ),
  );

  // result is the validation object with terminal status
  if (result.status === "valid") {
    console.log("Validation passed!");
  }
}

// Works with pipe
const untilDone = pipe(source, takeUntil((x) => x.done));
```

**Key difference between `takeWhile` and `takeUntil`:**

| Helper      | Yields                  | Closes With    | Use Case                |
| ----------- | ----------------------- | -------------- | ----------------------- |
| `takeWhile` | While predicate is true | `undefined`    | "Keep going while good" |
| `takeUntil` | Until predicate is true | Matching value | "Stop when you find it" |

### ForEach

The `forEach` helper invokes a function for each item passing through a stream.
This is useful when you need to perform side effects or operations on each item
without transforming the stream itself. Unlike other stream helpers that return
transformed streams, `forEach` consumes the entire stream and returns the
stream's close value.

```typescript
import { forEach } from "@effectionx/stream-helpers";
import { createSignal, spawn } from "effection";

function* example() {
  const stream = createSignal<number, void>();

  // Process each item in the stream
  yield* spawn(() =>
    forEach(function* (item) {
      console.log(`Processing: ${item}`);
      // Perform any side effects here
    }, stream)
  );

  stream.send(1);
  stream.send(2);
  stream.send(3);
  stream.close();
}

// Example: Handling stream close value
function* exampleWithCloseValue() {
  const stream = createSignal<string, number>();

  const result = yield* spawn(() =>
    forEach(function* (item) {
      console.log(`Item: ${item}`);
    }, stream)
  );

  stream.send("hello");
  stream.send("world");
  stream.close(42); // Close with value 42

  const closeValue = yield* result;
  console.log(`Stream closed with: ${closeValue}`); // 42
}
```

### Subject

Subject helper converts any stream into a multicast stream that replays the
latest value to new subscribers. It's analogous to
[RxJS BehaviorSubject](https://www.learnrxjs.io/learn-rxjs/subjects/behaviorsubject).

```typescript
import { createSubject } from "@effectionx/stream-helpers";
import { createChannel, spawn } from "effection";

function* example() {
  const subject = createSubject<number>();
  const channel = createChannel<number, void>();
  const downstream = subject(channel);

  // First subscriber
  const sub1 = yield* downstream;

  yield* channel.send(1);
  yield* channel.send(2);

  console.log(yield* sub1.next()); // { done: false, value: 1 }
  console.log(yield* sub1.next()); // { done: false, value: 2 }

  // Late subscriber gets the latest value immediately
  const sub2 = yield* downstream;
  console.log(yield* sub2.next()); // { done: false, value: 2 }
}
```

Use it with a pipe operator to convert any stream into a behavior subject:

```typescript
import { createSubject, map } from "@effectionx/stream-helpers";
import { pipe } from "remeda";

const subject = createSubject<string>();

const stream = pipe(
  source,
  map(function* (x) {
    return x.toString();
  }),
  subject,
);
```

### Passthrough Tracker

Passthrough Tracker stream helper provides a way to know if all items that
passed through the stream have been handled. This is especially helpful when you
want to ensure that all items were processed before completing an operation.

It's different from other stream helpers because you must first call
`createTracker` function which retuns an object. The actual helper is on the
`passthrough` method which you can call and chain as you would with other
helpers.

```typescript
import { each, signal } from "effection";
import { createTracker } from "@ffectionx/stream-helpers"

const source = signal(0);

// create the tracker
const tracker = yield* createTracker();

// create  passthrough stream helper
const track = tracker.passthrough();

for (const value of yield* each(track(source))) {
  // mark items 
  tracker.markOne(value);
  yield* each.next();
}

// will resolve when all items that passed through the stream were seen
yield* tracker;
```

### Composing stream helpers

You can use a simple `pipe()` to compose a series of stream helpers together. In
this example, we use one from [remeda](https://remedajs.com/docs/#pipe),

```typescript
import { batch, filter, forEach, map, valve } from "@effectionx/stream-helpers";
// any standard pipe function should work
import { pipe } from "remeda";

function* example(source: Stream<number, unknown>) {
  // Compose stream helpers using pipe
  const stream = pipe(
    source,
    valve({ open, close, openAt: 100, closeAt: 100 }),
    filter(function* (x) {
      return x > 0;
    }),
    map(function* (x) {
      return x * 20;
    }),
    batch({ maxSize: 50 }),
  );

  // Use forEach to process each value in the composed stream
  yield* forEach(function* (value) {
    console.log(value);
  }, stream);
}
```

## Testing Streams

The library includes testing utilities to help you test your stream processing
code. These are available in `@effectionx/stream-helpers/test-helpers` export.

### Faucet

The `useFaucet` function creates a stream that can be used to test the behavior
of streams that use backpressure. It's particularly useful in tests where you
need a controllable source stream.

```typescript
import { useFaucet } from "@effectionx/stream-helpers/test-helpers";
import { each, run, spawn } from "effection";

await run(function* () {
  const faucet = yield* useFaucet<number>({ open: true });

  // Remember to spawn the stream subscription before sending items to the stream
  yield* spawn(function* () {
    for (let i of yield* each(faucet)) {
      console.log(i);
      yield* each.next();
    }
  });

  // Pass an array of items to send items to the stream one at a time synchronously
  yield* faucet.pour([1, 2, 3]);

  // Pass an operation to control the rate at which items are sent to the stream
  yield* faucet.pour(function* (send) {
    yield* sleep(10);
    yield* send(5);
    yield* sleep(30);
    yield* send(6);
    yield* sleep(10);
    yield* send(7);
  });

  // You can close the faucet to stop items from being sent
  faucet.close();

  // And open it again when needed
  faucet.open();
});
```

When passing a function to `faucet.pour`, the `send` function will return an
operation. This operation will block when the faucet is closed. This is
particularly helpful when testing backpressure, because you can send many event
without worrying about the `open/close` state.

---
_Source: https://npm.io/package/@effectionx/stream-helpers · Machine-readable twin of the npm.io package page. Health data is recomputed on every publish._