@push-stream-std/push-sorted-merge (0.0.6)
Installation
@push-stream-std:registry=https://hub.kl1.tenere.ai/api/packages/push-stream-std/npm/npm install @push-stream-std/push-sorted-merge@0.0.6"@push-stream-std/push-sorted-merge": "0.0.6"About this package
@push-stream-std/push-sorted-merge
Merge multiple sorted push-streams with dynamic source addition into a single sorted output.
Published as @push-stream-std/push-sorted-merge.
Synopsis
import { sortedMerge } from '@push-stream-std/push-sorted-merge';
import { values } from '@push-stream-std/push-pushable';
import { collect } from '@push-stream-std/push-concat';
const merger = sortedMerge((a, b) => a - b);
merger.add(values([2, 5, 8]));
merger.add(values([1, 4, 7]));
merger.add(values([3, 6, 9]));
merger.pipe(collect((err, results) => {
console.log(results); // [1, 2, 3, 4, 5, 6, 7, 8, 9]
}));
merger.cap(); // finalize sources and begin emitting
Install
npm config set @push-stream-std:registry https://hub.kl1.tenere.ai/api/packages/push-stream-std/npm/
npm config set -- //hub.kl1.tenere.ai/api/packages/push-stream-std/npm/:_authToken "$PACKAGE_TOKEN"
npm install @push-stream-std/push-sorted-merge
Why
A push-stream operator that merges multiple sorted input streams into a single globally sorted output stream. Sources are added dynamically via add() during an initialization phase. After all sources are added, calling cap() marks the end of the initialization phase, flushes buffered values, and begins emitting the globally sorted stream. Each source's values are buffered separately; the smallest value across all active sources is selected on each emission step. Uses O(1) amortized buffer operations with head pointer pattern. If no comparator is provided, a natural </> comparison is used.
API
sortedMerge([comparator])
comparator(function, optional) —function(a, b)— Comparison function. Defaults to natural comparison using<and>operators, returning -1, 0, or 1.
Returns a merge stream object with the following methods:
stream.add(source)
Adds a sorted push-stream source to the merge. Must be called before cap(). Returns the stream for chaining. Sources added after cap() are ignored.
stream.cap()
Finalizes the set of sources, flushes initialization buffers, and begins emitting sorted values downstream. Must be called after all sources are added. Returns the stream.
stream.pipe(sink)
Standard push-stream pipe. Connects a sink to receive the sorted merged output.
stream.end([err])
Ends the merge stream. If called during initialization, the end is buffered and applied after cap().
stream.abort(err)
Aborts all active sources and the merge stream with the given error.
Tests
npm test
Tests cover merging pre-sorted streams, dynamic source addition, natural comparison, custom comparator, cap() behavior, initialization buffering, and empty streams.
Requirements
Node.js 22 or later. ESM only.