API · Transform

`filter()`

Keep successful values whose synchronous predicate returns a truthy result.

Signature

filter<S extends T>(
  fn: (value: T, context: CallbackContext<T, C>) => value is S,
): Exstream<S, C>

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

Example

const paid = exstream(orders).filter((order) => order.status === 'paid')

Parameters

fn

Type (value, context) => unknownRequired

Called once for every successful value. Truthy results keep the value; falsy results drop it. A TypeScript type-predicate return narrows the output type. The optional context is materialized only when the callback declares its second parameter.

Behavior

Kept values are emitted unchanged and in input order. Dropped values do not consume an output slot: the operator continues requesting upstream records until downstream receives a kept value or the source ends. It has no separate buffer.

When requested, context.input identifies the value that created the context, context.signal follows branch cancellation, and custom fields added upstream are preserved. Kept values retain that same context object.

The predicate is not awaited. A returned promise is an object and therefore truthy, so it keeps the value regardless of its eventual result. Use mapAsync() followed by filter() when the decision requires asynchronous work.

Existing record errors pass through without invoking the predicate.

Narrowing

const values: Array<string | null> = ['Ada', null]

const names = exstream(values).filter((value): value is string => value !== null)
// Exstream<string, C>

Errors

A thrown predicate error becomes a contextual record error associated with the current input. Handle it with errors(), skipErrors(), or another error policy. Fatal graph failures bypass the predicate and abort the branch.

Forms

filter() is available on streams and reusable pipelines. Its standalone form accepts the stream directly or returns a curried operator:

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

map(), flatMap(), skipErrors()

filter()