import { DifferenceStreamWriter, LinearUnaryOperator } from '../graph.js' import { StreamBuilder } from '../d2.js' import type { IStreamBuilder, PipedOperator } from '../types.js' import type { DifferenceStreamReader } from '../graph.js' import type { MultiSet } from '../multiset.js' /** * Operator that filters elements from the input stream */ export class FilterOperator extends LinearUnaryOperator { #f: (data: T) => boolean constructor( id: number, inputA: DifferenceStreamReader, output: DifferenceStreamWriter, f: (data: T) => boolean, ) { super(id, inputA, output) this.#f = f } inner(collection: MultiSet): MultiSet { return collection.filter(this.#f) } } /** * Filters elements from the input stream * @param f - The predicate to filter elements */ export function filter(f: (data: T) => boolean): PipedOperator { return (stream: IStreamBuilder): IStreamBuilder => { const output = new StreamBuilder( stream.graph, new DifferenceStreamWriter(), ) const operator = new FilterOperator( stream.graph.getNextOperatorId(), stream.connectReader(), output.writer, f, ) stream.graph.addOperator(operator) return output } }