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 applies a function to each multi-set in the input stream */ export class TapOperator extends LinearUnaryOperator { #f: (data: MultiSet) => void constructor( id: number, inputA: DifferenceStreamReader, output: DifferenceStreamWriter, f: (data: MultiSet) => void, ) { super(id, inputA, output) this.#f = f } inner(collection: MultiSet): MultiSet { this.#f(collection) return collection } } /** * Invokes a function for each multi-set in the input stream. * This operator doesn't modify the stream and is used to perform side effects. * @param f - The function to invoke on each multi-set * @returns The input stream */ export function tap(f: (data: MultiSet) => void): PipedOperator { return (stream: IStreamBuilder): IStreamBuilder => { const output = new StreamBuilder( stream.graph, new DifferenceStreamWriter(), ) const operator = new TapOperator( stream.graph.getNextOperatorId(), stream.connectReader(), output.writer, f, ) stream.graph.addOperator(operator) return output } }