55 lines
1.4 KiB
JavaScript
55 lines
1.4 KiB
JavaScript
import {
|
|
getOperator,
|
|
initOptions,
|
|
ProcessingMode
|
|
} from "./core";
|
|
import { Lazy } from "./lazy";
|
|
import { assert, cloneDeep, isEmpty } from "./util";
|
|
class Aggregator {
|
|
#pipeline;
|
|
#options;
|
|
constructor(pipeline, options) {
|
|
this.#pipeline = pipeline;
|
|
this.#options = initOptions(options);
|
|
}
|
|
/**
|
|
* Returns an {@link Iterator} for lazy evaluation of the pipeline.
|
|
*
|
|
* @param collection An array or iterator object
|
|
* @returns {Iterator} an iterator object
|
|
*/
|
|
stream(collection, options) {
|
|
let iter = Lazy(collection);
|
|
const opts = options ?? this.#options;
|
|
const mode = opts.processingMode;
|
|
if (mode & ProcessingMode.CLONE_INPUT) iter.map(cloneDeep);
|
|
const stages = new Array();
|
|
if (!isEmpty(this.#pipeline)) {
|
|
for (const opExpr of this.#pipeline) {
|
|
const opKeys = Object.keys(opExpr);
|
|
const opName = opKeys[0];
|
|
const call = getOperator("pipeline", opName, opts);
|
|
assert(
|
|
opKeys.length === 1 && !!call,
|
|
`invalid pipeline operator ${opName}`
|
|
);
|
|
stages.push(opName);
|
|
iter = call(iter, opExpr[opName], opts);
|
|
}
|
|
}
|
|
if (mode & ProcessingMode.CLONE_OUTPUT) iter.map(cloneDeep);
|
|
return iter;
|
|
}
|
|
/**
|
|
* Return the results of the aggregation as an array.
|
|
*
|
|
* @param collection
|
|
*/
|
|
run(collection, options) {
|
|
return this.stream(collection, options).value();
|
|
}
|
|
}
|
|
export {
|
|
Aggregator
|
|
};
|