implement Source.filterAsync

src/source-spec.ts

@@ -161,6 +161,20 @@ describe('Stream', () => {
 
     });
 
+    describe('filterAsync', () => {
+
+      it('produces an empty source from an empty source', () => {
+        const src = Source.empty();
+        return chai.expect(src.filterAsync(() => Promise.resolve(true)).toArray()).to.eventually.deep.equal([]);
+      });
+
+      it('only produces elements for which the predicate returns true', () => {
+        const src = Source.fromArray([1,2,3,4,5,6]);
+        return chai.expect(src.filterAsync(x => Promise.resolve(x % 2 === 0)).toArray()).to.eventually.deep.equal([2,4,6]);
+      });
+
+    });
+
     describe('intoOutputStream', () => {
 
       it('writes no data when the source is empty', () => {

src/source.ts

@@ -120,6 +120,21 @@ export class Source<Output> {
     });
   }
 
+  filterAsync(pred: (output: Output) => Promise<boolean>): Source<Output> {
+    return new Source(<State, Result>(sink: SinkInterface<Output, State, Result>) => {
+      return this.pipe<State, Result>({
+        onStart: () => sink.onStart(),
+        onData: (state, output) => pred(output).then(res => {
+          if (!res) {
+            return Promise.resolve(state);
+          }
+          return sink.onData(state, output);
+        }),
+        onEnd: (state) => sink.onEnd(state)
+      });
+    });
+  }
+
   toArray(): Promise<Array<Output>> {
     return this.fold([], (arr, outp) => arr.concat([outp]));
   }