Implement filter function

package.json

@@ -1,6 +1,6 @@
 {
   "name": "chained-promise",
-  "version": "0.2.1",
+  "version": "0.2.2",
   "description": "Functional programming tools for recurring promises",
   "main": "dist/src/index.js",
   "repository": {

src/ChainedPromise.ts

@@ -15,6 +15,11 @@
  */
 import fix from './fix';
 
+type SkipToken<T> = {
+  data: symbol,
+  next: ChainedPromise<T>
+};
+
 /**
  * An extended promise for recurring promises with multiple compositions.
  *
@@ -49,16 +54,22 @@ class ChainedPromise<T> extends Promise<T> {
    * Initializes fields common to both {@link ChainedPromise#constructor}
    * and {@link ChainedPromise#from} code path.
    */
-  private _initialize() {
+  private initialize() {
     this.flatMapChain = [];
   }
 
   /**
    * Constructs next {@link ChainedPromise} that carries over settings and
    * composition properties of the current one.
    */
-  _nextPromise<U>(v: T): ChainedPromise<U> {
-    const nextPromise = ChainedPromise.from(this.next(v), this.next);
+  private nextPromise<U>(v: T|SkipToken<T>): ChainedPromise<U> {
+    let nextPromise;
+    if ((v as SkipToken<T>).data &&
+        (v as SkipToken<T>).data === ChainedPromise.SKIP) {
+      nextPromise = ChainedPromise.from((v as SkipToken<T>).next, this.next);
+    } else {
+      nextPromise = ChainedPromise.from(this.next(v as T), this.next);
+    }
     nextPromise.flatMapChain = this.flatMapChain;
     return ((nextPromise as any) as ChainedPromise<U>);
   }
@@ -70,7 +81,7 @@ class ChainedPromise<T> extends Promise<T> {
   constructor(executor, next = ChainedPromise.nextFieldPicker<T>('next')) {
     super(executor);
     this.next = next;
-    this._initialize();
+    this.initialize();
   }
 
   /**
@@ -132,13 +143,19 @@ class ChainedPromise<T> extends Promise<T> {
    * ChainedPromise.DONE} symbol.
    */
   forEach<V>(fn: (v: T) => void): Promise<V> {
-    return fix<T, V>((v: T, complete) => {
-      fn(v);
-      const nextPromise = this.next(v);
+    return fix<T, V>((v: T|SkipToken<T>, complete) => {
+      let nextPromise;
+      if ((v as SkipToken<T>).data &&
+          (v as SkipToken<T>).data === ChainedPromise.SKIP) {
+        nextPromise = (v as SkipToken<T>).next;
+      } else {
+        fn(v as T);
+        nextPromise = this.next(v as T);
+      }
       if (nextPromise[ChainedPromise.DONE] !== undefined) {
         return complete(nextPromise[ChainedPromise.DONE]);
       } else {
-        return this._nextPromise(v);
+        return this.nextPromise(v as T);
       }
     })(this);
   }
@@ -164,7 +181,7 @@ class ChainedPromise<T> extends Promise<T> {
    * Overrides Promise.then to compose with extra functions. See {@link
    * ChainedPromise} for the specifics of available compositions.
    */
-  then<TResult1 = T, TResult2 = never>(
+  then<TResult1 = T | SkipToken<T>, TResult2 = never>(
       onFulfilled?: ((value: T) => TResult1 | PromiseLike<TResult1>)|null|
       undefined,
       onRejected?: (result: any) => TResult2 |
@@ -181,8 +198,14 @@ class ChainedPromise<T> extends Promise<T> {
       return super.then(onFulfilled as any, onRejected as any) as any;
     } else {
       const firstFlatMapped = super.then(this.flatMapChain[0]);
-      let flatMapped = this.flatMapChain.slice(1).reduce(
-          (x, y) => x.then(y), firstFlatMapped);
+      let flatMapped = this.flatMapChain.slice(1).reduce((x, y) => {
+        return x.then((res) => {
+          if (res.data && res.data === ChainedPromise.SKIP) {
+            return res;
+          }
+          return y(res);
+        });
+      }, firstFlatMapped);
       return flatMapped.then(onFulfilled, onRejected);
     }
   }
@@ -275,13 +298,27 @@ class ChainedPromise<T> extends Promise<T> {
         });
   }
 
+  /**
+   * Filters for values that evaluates to `true`.
+   */
+  filter(fn: (t: T) => boolean): ChainedPromise<T|SkipToken<T>> {
+    return this.map<T|SkipToken<T>>((v) => {
+      if (!fn(v)) {
+        return {data: ChainedPromise.SKIP, next: this.next(v)};
+      }
+      return v;
+    });
+  }
+
   /**
    * Symbol to indicate the end of promise chains. Having
    * `{[ChainedPromise.DONE]: <some value>}` as a next value will indicate the
    * end of the chain, and will cause fixed promises such as
    * {@link ChainedPromise#forEach} to resolve to the given value.
    */
   static DONE = Symbol('ChainedPromise.DONE');
+
+  private static SKIP = Symbol('ChainedPromise.SKIP');
 }
 
 export default ChainedPromise;

test/ChainedPromise.spec.ts

@@ -375,4 +375,27 @@ describe('ChainedPromise', function() {
           .catch(done);
     });
   });
+  describe('filter', () => {
+    it('skips data', (done) => {
+      const thirdPromise =
+          Promise.resolve({data: 3, next: {[ChainedPromise.DONE]: 'done'}});
+      const secondPromise = Promise.resolve({data: 2, next: thirdPromise});
+      const testChainedPromise =
+          new ChainedPromise<{data: number}>((resolver) => {
+            resolver({data: 1, next: secondPromise});
+          });
+      testChainedPromise
+          .map((v) => ({...v, data: v.data + 1}))  // [2, 3, 4]
+          .filter((v) => v.data % 2 == 0)          // [2, 4]
+          .map((v) => ({...v, data: v.data * 2}))  // [4, 8]
+          .collect<{data: number}, string>()
+          .then((result) => {
+            expect((result[0] as {data: number}).data).toEqual(4);
+            expect((result[1] as {data: number}).data).toEqual(8);
+            expect(result[2]).toEqual('done');
+            expect(result.length).toEqual(3);
+            done();
+          });
+    });
+  });
 });