API · Select

`stopWhen()`

Emit values until a synchronous condition matches, including the value that matched.

Signature

stopWhen(fn: (value: T, context: C) => unknown): Exstream<T, C>

Example

exstream([1, 2, 3, 4])
  .stopWhen((value) => value === 3)
  .valuesSync()
// [1, 2, 3]

Parameters

fn

Type (value, context) => unknownRequired

Runs synchronously after each successful value has been emitted. A truthy result stops this branch.

Behavior

The matching value is always emitted before the branch is destroyed. If the predicate never matches, all values pass through. stopWhen() preserves order and context, adds no independent queue, and propagates downstream pressure.

Destroying this branch releases its upstream consumer. Shared sources continue for other forks; an otherwise unneeded source can stop. This is branch termination, not a fatal pipeline failure.

Errors

Existing record errors pass through and do not call fn. If fn throws, the exception escapes the synchronous consumer boundary; handle fallible logic before this operator. Promises are not awaited and are truthy, so an async predicate stops on the first value.

Forms

stream.stopWhen(predicate)
exstream.pipeline().stopWhen(predicate)
exstream.stopWhen(predicate, stream)
stream.through(exstream.stopWhen(predicate))

take(), find(), filter()

stopWhen()