Add parallelForEach and transform stream utilities

Author: jdalton

registry/lib/streams.d.ts

@@ -1,5 +1,10 @@
 declare type AnyIterable<T> = Iterable<T> | AsyncIterable<T>
 declare const Streams: {
+  parallelForEach<T>(
+    concurrency: number,
+    func: (data: T) => Promise<void>,
+    iterable: AsyncIterable<T>
+  ): Promise<void>
   parallelMap<T, R>(
     concurrency: number
   ): {
@@ -20,8 +25,26 @@ declare const Streams: {
     func: (data: T) => R | Promise<R>,
     iterable: AnyIterable<T>
   ): AsyncIterableIterator<R>
+  transform(concurrency: number): {
+    <T, R>(
+      func: (data: T) => R | Promise<R>,
+      iterable: AnyIterable<T>
+    ): AsyncIterableIterator<R>
+    <T, R>(
+      func: (data: T) => R | Promise<R>
+    ): (iterable: AnyIterable<T>) => AsyncIterableIterator<R>
+  }
+  transform<T, R>(
+    concurrency: number,
+    func: (data: T) => R | Promise<R>
+  ): (iterable: AnyIterable<T>) => AsyncIterableIterator<R>
+  transform<T, R>(
+    concurrency: number,
+    func: (data: T) => R | Promise<R>,
+    iterable: AnyIterable<T>
+  ): AsyncIterableIterator<R>
 }
-declare namespace Promises {
+declare namespace Streams {
   export { AnyIterable }
 }
 export = Streams

registry/lib/streams.js

@@ -1,5 +1,7 @@
 'use strict'
 
+const { apply: ReflectApply } = Reflect
+
 let _streamingIterables
 /*@__NO_SIDE_EFFECTS__*/
 function getStreamingIterables() {
@@ -10,11 +12,26 @@ function getStreamingIterables() {
 }
 
 /*@__NO_SIDE_EFFECTS__*/
-function parallelMap(concurrency, func) {
+async function parallelForEach(concurrency, func, iterable) {
+  for await (const _ of parallelMap(concurrency, func, iterable)) {
+    /* empty block */
+  }
+}
+
+/*@__NO_SIDE_EFFECTS__*/
+function parallelMap(...args) {
+  const streamingIterables = getStreamingIterables()
+  return ReflectApply(streamingIterables.parallelMap, undefined, args)
+}
+
+/*@__NO_SIDE_EFFECTS__*/
+function transform(...args) {
   const streamingIterables = getStreamingIterables()
-  return streamingIterables.parallelMap(concurrency, func)
+  return ReflectApply(streamingIterables.transform, undefined, args)
 }
 
 module.exports = {
-  parallelMap
+  parallelForEach,
+  parallelMap,
+  transform
 }