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 element in the input stream */ export class MapOperator extends LinearUnaryOperator { #f: (data: T) => U constructor( id: number, inputA: DifferenceStreamReader, output: DifferenceStreamWriter, f: (data: T) => U, ) { super(id, inputA, output) this.#f = f } inner(collection: MultiSet): MultiSet { return collection.map(this.#f) } } /** * Applies a function to each element in the input stream * @param f - The function to apply to each element */ export function map(f: (data: T) => O): PipedOperator { return (stream: IStreamBuilder): IStreamBuilder => { const output = new StreamBuilder( stream.graph, new DifferenceStreamWriter(), ) const operator = new MapOperator( stream.graph.getNextOperatorId(), stream.connectReader(), output.writer, f, ) stream.graph.addOperator(operator) return output } }