feat(transducers): add partitionTime() transducer

thi-ng · Feb 28, 2020 · efafd0b · efafd0b
1 parent 38d511b
commit efafd0b
Show file tree

Hide file tree

Showing 4 changed files with 56 additions and 2 deletions.
diff --git a/packages/transducers/README.md b/packages/transducers/README.md
@@ -1,6 +1,6 @@
 <!-- This file is generated - DO NOT EDIT! -->
 
-# ![@thi.ng/transducers](https://media.thi.ng/umbrella/banners/thing-transducers.svg?1582660659)
+# ![@thi.ng/transducers](https://media.thi.ng/umbrella/banners/thing-transducers.svg?1582933570)
 
 [![npm version](https://img.shields.io/npm/v/@thi.ng/transducers.svg)](https://www.npmjs.com/package/@thi.ng/transducers)
 ![npm downloads](https://img.shields.io/npm/dm/@thi.ng/transducers.svg)
@@ -143,7 +143,7 @@ package.
 yarn add @thi.ng/transducers
 ```
 
-Package sizes (gzipped): ESM: 7.8KB / CJS: 8.4KB / UMD: 7.6KB
+Package sizes (gzipped): ESM: 7.9KB / CJS: 8.4KB / UMD: 7.6KB
 
 ## Dependencies
 
@@ -967,6 +967,7 @@ tx.transduce(tx.map((x) => x*10), tx.push(), tx.range(4))
 - [partitionOf](https://github.com/thi-ng/umbrella/tree/develop/packages/transducers/src/xform/partition-of.ts)
 - [partitionSort](https://github.com/thi-ng/umbrella/tree/develop/packages/transducers/src/xform/partition-sort.ts)
 - [partitionSync](https://github.com/thi-ng/umbrella/tree/develop/packages/transducers/src/xform/partition-sync.ts)
+- [partitionTime](https://github.com/thi-ng/umbrella/tree/develop/packages/transducers/src/xform/partition-time.ts)
 - [partition](https://github.com/thi-ng/umbrella/tree/develop/packages/transducers/src/xform/partition.ts)
 - [peek](https://github.com/thi-ng/umbrella/tree/develop/packages/transducers/src/xform/peek.ts)
 - [pluck](https://github.com/thi-ng/umbrella/tree/develop/packages/transducers/src/xform/pluck.ts)

diff --git a/packages/transducers/README.tpl.md b/packages/transducers/README.tpl.md
@@ -727,6 +727,7 @@ tx.transduce(tx.map((x) => x*10), tx.push(), tx.range(4))
 - [partitionOf](https://github.com/thi-ng/umbrella/tree/develop/packages/transducers/src/xform/partition-of.ts)
 - [partitionSort](https://github.com/thi-ng/umbrella/tree/develop/packages/transducers/src/xform/partition-sort.ts)
 - [partitionSync](https://github.com/thi-ng/umbrella/tree/develop/packages/transducers/src/xform/partition-sync.ts)
+- [partitionTime](https://github.com/thi-ng/umbrella/tree/develop/packages/transducers/src/xform/partition-time.ts)
 - [partition](https://github.com/thi-ng/umbrella/tree/develop/packages/transducers/src/xform/partition.ts)
 - [peek](https://github.com/thi-ng/umbrella/tree/develop/packages/transducers/src/xform/peek.ts)
 - [pluck](https://github.com/thi-ng/umbrella/tree/develop/packages/transducers/src/xform/pluck.ts)

diff --git a/packages/transducers/src/index.ts b/packages/transducers/src/index.ts
@@ -76,6 +76,7 @@ export * from "./xform/partition-by";
 export * from "./xform/partition-of";
 export * from "./xform/partition-sort";
 export * from "./xform/partition-sync";
+export * from "./xform/partition-time";
 export * from "./xform/partition";
 export * from "./xform/peek";
 export * from "./xform/pluck";

diff --git a/packages/transducers/src/xform/partition-time.ts b/packages/transducers/src/xform/partition-time.ts
@@ -0,0 +1,51 @@
+import { iterator } from "../iterator";
+import { partitionBy } from "./partition-by";
+import type { Transducer } from "../api";
+
+/**
+ * Transducer. Yields tumbling, non-overlapping windows/partitions of
+ * input values, with the window size defined by given realtime `period`
+ * (in milliseconds).
+ *
+ * @remarks
+ * Only to be used in async contexts, NOT with {@link (transduce:1)}
+ * directly.
+ *
+ * Also see:
+ * - {@link @thi.ng/rstream# | @thi.ng/rstream}
+ * - {@link @thi.ng/csp# | @thi.ng/csp}.
+ *
+ * @example
+ * ```ts
+ * import { fromInterval, trace } from "@thi.ng/rstream";
+ *
+ * // stream emits
+ * fromInterval(250)
+ *   .transform(partitionTime(1000))
+ *   .subscribe(trace())
+ * // [ 0, 1, 2, 3 ]
+ * // [ 4, 5, 6, 7 ]
+ * // [ 8, 9, 10, 11 ]
+ * // [ 12, 13, 14, 15 ]
+ * // ...
+ * ```
+ *
+ * @param period - window size (in ms)
+ */
+export function partitionTime<T>(period: number): Transducer<T, T[]>;
+export function partitionTime<T>(
+    period: number,
+    src: Iterable<T>
+): IterableIterator<T[]>;
+export function partitionTime<T>(period: number, src?: Iterable<T>): any {
+    return src
+        ? iterator(partitionTime(period), src)
+        : partitionBy(() => {
+              let last = 0;
+              return () => {
+                  const t = Date.now();
+                  t - last >= period && (last = t);
+                  return last;
+              };
+          }, true);
+}