Adds throttle

src/Source.ts

@@ -29,6 +29,10 @@ export class Source<T> {
     return new Source(FilteredSourceBuilder(this.builder, predicate))
   }
 
+  throttle(elements: number, perDurationMs: number) {
+    return new Source(ThrottledSourceBuilder(this.builder, elements, perDurationMs))
+  }
+
   into<M>(sink: Sink<T, M>) {
     const self = this
     return {
@@ -218,3 +222,29 @@ function FilteredSourceBuilder<T>(prev: SourceBuilder<T>, predicate: (v: T) => b
     }
   } 
 }
+
+
+function ThrottledSourceBuilder<T>(prev: SourceBuilder<T>, elements: number, perDurationMs: number) {
+  let timestamps: number[] = []
+  return {
+    build() {
+      return [
+        ...prev.build(),
+        new stream.Transform({
+          objectMode: true,
+          transform(v, {}, onNext) {
+            const current = Date.now()
+            timestamps = [...timestamps.filter(t => t > (current - perDurationMs)), current]
+            this.push(v)
+            if(timestamps.length >= elements) {
+              const  timeToWait = (timestamps[0] + perDurationMs) - Date.now()
+              setTimeout(onNext, timeToWait)
+            } else {
+              onNext()
+            }
+          }
+        })
+      ]
+    }
+  }
+}

src/streams.test.ts

@@ -84,4 +84,16 @@ describe('Source', function () {
     )
   })
 
+  it('should throttle elements', async () => {
+    const before = Date.now()
+    const result = await Source.fromArray([1,2,3,4,5])
+      .throttle(2, 10)
+      .into(Sink.sum)
+      .run()
+    const after = Date.now()
+    const duration = after - before
+    assert.equal(result, 15)
+    assert.ok(duration> 20, 'must at least be 20ms but was ' + duration)
+  })
+
 });