@push-stream-std/push-window (0.0.9)
Installation
@push-stream-std:registry=https://hub.kl1.tenere.ai/api/packages/push-stream-std/npm/npm install @push-stream-std/push-window@0.0.9"@push-stream-std/push-window": "0.0.9"About this package
@push-stream-std/push-window
Aggregate stream values into sliding, tumbling, or time-based windows.
Published as @push-stream-std/push-window.
Synopsis
import { sliding, tumbling, time } from '@push-stream-std/push-window';
import pushable from '@push-stream-std/push-pushable';
const source = pushable();
const sink = {
paused: false,
write(v) { console.log(v); },
end() { console.log('done'); }
};
// Sliding window: emits last N values after each new value
source.pipe(sliding(3)).pipe(sink);
source.push(1); // [1]
source.push(2); // [1, 2]
source.push(3); // [1, 2, 3]
source.push(4); // [2, 3, 4]
source.end(); // done
// Tumbling window: emits groups of N values, flushing remainder on end
source.pipe(tumbling(3)).pipe(sink);
// Input: 1,2,3,4,5 → Output: [1,2,3], [4,5]
// Time window: emits values collected within windowMs intervals
source.pipe(time(1000, { emitEmpty: false })).pipe(sink);
// Emits arrays of values received each second
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-window
Why
A family of through streams that buffer upstream values and emit them as arrays (windows). Sliding windows emit the last size values after every new write, continuously overlapping. Tumbling windows emit non-overlapping chunks of size values, flushing any remainder on end. Time windows batched values received within a configurable time interval. The sliding window uses a head pointer pattern for amortized O(1) reads.
API
sliding(size)
Creates a sliding window through stream.
- size
number– Maximum number of values to keep in the window. Must be a positive integer. - Returns A through stream emitting arrays of length ≤
size.
tumbling(size)
Creates a tumbling window through stream.
- size
number– Number of values per window chunk. Must be a positive integer. - Returns A through stream emitting arrays of exactly
sizevalues (except possibly the last chunk on end).
time(windowMs, options)
Creates a time-based window through stream.
- windowMs
number– Time interval in milliseconds. Must be a positive integer. - options
object(optional):- emitEmpty
boolean(defaulttrue) – Whether to emit empty arrays when no values arrived during a window.
- emitEmpty
- Returns A through stream emitting arrays of values collected within each interval.
All returned streams have pipe(), write(), end(), abort(), resume(), paused, and ended.
Tests
npm test
Tests cover sliding window accumulation and eviction, tumbling window fixed-size chunks and end flush, time window interval batching, empty window emission, and abort cleanup.
Requirements
Node.js 22 or later. ESM only.