feat(task): added the pipe operator

Author: hrajchert

src/operators/basic.ts

@@ -0,0 +1,60 @@
+import { Task, UncaughtError } from '../task';
+
+export type IMapFn <A, B> = (a: A) => B;
+export type ITaskChainFn<A, B, E> = (value: A) => Task<B, E>;
+export type IPipeFn<T1, T2, E1, E2> = (a: Task<T1, E1>) => Task<T2, E2>;
+
+export function map<T1, T2, E> (fn: IMapFn<T1, T2>) {
+    return function (input: Task<T1, E>): Task<T2, E | UncaughtError> {
+        return new Task((outerResolve, outerReject) => {
+            input.fork(
+                outerReject,
+                value => {
+                    try {
+                        const result = fn(value);
+                        outerResolve(result);
+                    } catch (error) {
+                        outerReject(new UncaughtError(error));
+                    }
+                }
+            );
+        });
+    };
+}
+
+export function chain<T1, T2, E1, E2> (fn: ITaskChainFn<T1, T2, E2>) {
+    return function (input: Task<T1, E1>): Task<T2, E1 | E2 | UncaughtError> {
+        return new Task((outerResolve, outerReject) => {
+            input.fork(
+                outerReject,
+                value => {
+                    try {
+                        fn(value).fork(outerReject, outerResolve);
+                    }
+                    catch (err) {
+                        outerReject(new UncaughtError(err));
+                    }
+                }
+            );
+        });
+    };
+}
+
+export function catchError<T1, T2, E1, E2> (fn: ITaskChainFn<E1, T2, E2>) {
+    return function (input: Task<T1, E1>): Task<T1 | T2, E2 | UncaughtError> {
+        return new Task((outerResolve, outerReject) => {
+            input.fork(
+                err => {
+                    try {
+                        fn(err).fork(outerReject, outerResolve);
+                    }
+                    catch (err) {
+                        outerReject(new UncaughtError(err));
+                    }
+                },
+                outerResolve
+            );
+        });
+    };
+}
+

src/task.ts

@@ -1,5 +1,5 @@
-export type IMapFn <A, B> = (a: A) => B;
-export type ITaskChainFn<A, B, E> = (value: A) => Task<B, E>;
+import { catchError, chain, IMapFn, IPipeFn, ITaskChainFn, map } from './operators/basic';
+
 
 export class UncaughtError extends Error {
     type: 'UncaughtError';
@@ -95,52 +95,45 @@ export class Task <T, E> {
         });
     }
 
-    map<TResult> (fn: IMapFn<T, TResult>): Task<TResult, E | UncaughtError> {
+    pipe (): Task<T, E>;
+    pipe<T1, E1> (f1: IPipeFn<T, T1, E, E1>): Task<T1, E1 | UncaughtError>;
+    pipe<T1, T2, E1, E2> (f1: IPipeFn<T, T1, E, E1>, f2: IPipeFn<T1, T2, E1, E2>): Task<T2, E2 | UncaughtError>;
+    pipe<T1, T2, T3, E1, E2, E3> (f1: IPipeFn<T, T1, E, E1>, f2: IPipeFn<T1, T2, E1, E2>, f3: IPipeFn<T2, T3, E2, E3>): Task<T3, E3 | UncaughtError>;
+    pipe<T1, T2, T3, T4, E1, E2, E3, E4> (f1: IPipeFn<T, T1, E, E1>, f2: IPipeFn<T1, T2, E1, E2>, f3: IPipeFn<T2, T3, E2, E3>, f4: IPipeFn<T3, T4, E3, E4>): Task<T4, E4 | UncaughtError>;
+    pipe<T1, T2, T3, T4, T5, E1, E2, E3, E4, E5> (f1: IPipeFn<T, T1, E, E1>, f2: IPipeFn<T1, T2, E1, E2>, f3: IPipeFn<T2, T3, E2, E3>, f4: IPipeFn<T3, T4, E3, E4>, f5: IPipeFn<T4, T5, E4, E5>): Task<T5, E5 | UncaughtError>;
+    pipe<T1, T2, T3, T4, T5, T6, E1, E2, E3, E4, E5, E6> (f1: IPipeFn<T, T1, E, E1>, f2: IPipeFn<T1, T2, E1, E2>, f3: IPipeFn<T2, T3, E2, E3>, f4: IPipeFn<T3, T4, E3, E4>, f5: IPipeFn<T4, T5, E4, E5>, f6: IPipeFn<T5, T6, E5, E6>): Task<T6, E6 | UncaughtError>;
+    pipe<T1, T2, T3, T4, T5, T6, T7, E1, E2, E3, E4, E5, E6, E7> (f1: IPipeFn<T, T1, E, E1>, f2: IPipeFn<T1, T2, E1, E2>, f3: IPipeFn<T2, T3, E2, E3>, f4: IPipeFn<T3, T4, E3, E4>, f5: IPipeFn<T4, T5, E4, E5>, f6: IPipeFn<T5, T6, E5, E6>, f7: IPipeFn<T6, T7, E6, E7>): Task<T7, E7 | UncaughtError>;
+    pipe<T1, T2, T3, T4, T5, T6, T7, T8, E1, E2, E3, E4, E5, E6, E7, E8> (f1: IPipeFn<T, T1, E, E1>, f2: IPipeFn<T1, T2, E1, E2>, f3: IPipeFn<T2, T3, E2, E3>, f4: IPipeFn<T3, T4, E3, E4>, f5: IPipeFn<T4, T5, E4, E5>, f6: IPipeFn<T5, T6, E5, E6>, f7: IPipeFn<T6, T7, E6, E7>, f8: IPipeFn<T7, T8, E7, E8>): Task<T8, E8 | UncaughtError>;
+    pipe<T1, T2, T3, T4, T5, T6, T7, T8, T9, E1, E2, E3, E4, E5, E6, E7, E8, E9> (f1: IPipeFn<T, T1, E, E1>, f2: IPipeFn<T1, T2, E1, E2>, f3: IPipeFn<T2, T3, E2, E3>, f4: IPipeFn<T3, T4, E3, E4>, f5: IPipeFn<T4, T5, E4, E5>, f6: IPipeFn<T5, T6, E5, E6>, f7: IPipeFn<T6, T7, E6, E7>, f8: IPipeFn<T7, T8, E7, E8>, f9: IPipeFn<T8, T9, E8, E9>): Task<T9, E9 | UncaughtError>;
+    pipe<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, E1, E2, E3, E4, E5, E6, E7, E8, E9, E10> (f1: IPipeFn<T, T1, E, E1>, f2: IPipeFn<T1, T2, E1, E2>, f3: IPipeFn<T2, T3, E2, E3>, f4: IPipeFn<T3, T4, E3, E4>, f5: IPipeFn<T4, T5, E4, E5>, f6: IPipeFn<T5, T6, E5, E6>, f7: IPipeFn<T6, T7, E6, E7>, f8: IPipeFn<T7, T8, E7, E8>, f9: IPipeFn<T8, T9, E8, E9>, f10: IPipeFn<T9, T10, E9, E10>): Task<T10, E10 | UncaughtError>;
+    pipe<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, E1, E2, E3, E4, E5, E6, E7, E8, E9, E10, E11> (f1: IPipeFn<T, T1, E, E1>, f2: IPipeFn<T1, T2, E1, E2>, f3: IPipeFn<T2, T3, E2, E3>, f4: IPipeFn<T3, T4, E3, E4>, f5: IPipeFn<T4, T5, E4, E5>, f6: IPipeFn<T5, T6, E5, E6>, f7: IPipeFn<T6, T7, E6, E7>, f8: IPipeFn<T7, T8, E7, E8>, f9: IPipeFn<T8, T9, E8, E9>, f10: IPipeFn<T9, T10, E9, E10>, f11: IPipeFn<T10, T11, E10, E11>): Task<T11, E11 | UncaughtError>;
+    pipe<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, E1, E2, E3, E4, E5, E6, E7, E8, E9, E10, E11, E12> (f1: IPipeFn<T, T1, E, E1>, f2: IPipeFn<T1, T2, E1, E2>, f3: IPipeFn<T2, T3, E2, E3>, f4: IPipeFn<T3, T4, E3, E4>, f5: IPipeFn<T4, T5, E4, E5>, f6: IPipeFn<T5, T6, E5, E6>, f7: IPipeFn<T6, T7, E6, E7>, f8: IPipeFn<T7, T8, E7, E8>, f9: IPipeFn<T8, T9, E8, E9>, f10: IPipeFn<T9, T10, E9, E10>, f11: IPipeFn<T10, T11, E10, E11>, f12: IPipeFn<T11, T12, E11, E12>): Task<T12, E12 | UncaughtError>;
+    pipe<TFinal, EFinal> (...fns: any[]): Task<TFinal, EFinal | UncaughtError> {
         return new Task((outerResolve, outerReject) => {
-            this.fork(
-                outerReject,
-                value => {
+            const newTask = fns.reduce(
+                (task, f) => {
                     try {
-                        const result = fn(value);
-                        outerResolve(result);
-                    } catch (error) {
-                        outerReject(new UncaughtError(error));
+                        return f(task);
+                    } catch (err) {
+                        return Task.reject(new UncaughtError(err));
                     }
-                }
+                },
+                this
             );
+            return newTask.fork(outerReject, outerResolve);
         });
     }
 
+    map<TResult> (fn: IMapFn<T, TResult>): Task<TResult, E | UncaughtError> {
+        return map<T, TResult, E>(fn)(this);
+    }
+
     chain<TResult, EResult> (fn: ITaskChainFn<T, TResult, EResult>): Task<TResult, E | EResult | UncaughtError> {
-        return new Task((outerResolve, outerReject) => {
-            this.fork(
-                outerReject,
-                value => {
-                    try {
-                        fn(value).fork(outerReject, outerResolve);
-                    }
-                    catch (err) {
-                        outerReject(new UncaughtError(err));
-                    }
-                }
-            );
-        });
+        return chain<T, TResult, E, EResult>(fn)(this);
     }
 
     catch<TResult, EResult> (fn: ITaskChainFn<E, TResult, EResult>): Task<T | TResult, EResult | UncaughtError> {
-        return new Task((outerResolve, outerReject) => {
-            this.fork(
-                err => {
-                    try {
-                        fn(err).fork(outerReject, outerResolve);
-                    }
-                    catch (err) {
-                        outerReject(new UncaughtError(err));
-                    }
-                },
-                outerResolve
-            );
-        });
+        return catchError<T, TResult, E, EResult>(fn)(this);
     }
 
     fork (errorFn: (error: E) => any, successFn: (value: T) => any): void {
@@ -152,3 +145,4 @@ export class Task <T, E> {
         );
     }
 }
+

test/task.test.ts

@@ -1,8 +1,8 @@
+import { catchError, chain, map } from '../src/operators/basic';
 import { Task, UncaughtError } from '../src/task';
 import { assertFork, jestAssertNever, jestAssertUntypedNeverCalled } from './jest-helper';
 
 
-
 describe('Task', () => {
     describe('fork', () => {
         it('should be lazy (dont call if not forked)', cb => {
@@ -417,4 +417,156 @@ describe('Task', () => {
             );
         });
     });
+    describe('pipe', () => {
+        it('Should return the same task if no function is passed', (cb) => {
+            // GIVEN: a resolved value
+            const task = Task.resolve(0);
+
+            // WHEN: we call pipe with no transformation
+            const result = task.pipe();
+
+            // THEN: the success function should be called without transformation
+            result.fork(
+                jestAssertNever(cb),
+                assertFork(cb, x => expect(x).toBe(0))
+            );
+        });
+
+        it('Should work with one function', (cb) => {
+            // GIVEN: a resolved value
+            const task = Task.resolve(0);
+
+            // WHEN: we pipe the value
+            const result = task.pipe(
+                map(n => '' + n)
+            );
+
+            // THEN: the success function should be called with the transformed value
+            result.fork(
+                jestAssertNever(cb),
+                assertFork(cb, x => expect(x).toBe('0'))
+            );
+        });
+
+        it('Should work with multiple functions', (cb) => {
+            // GIVEN: a resolved value
+            const task = Task.resolve(0);
+
+            // WHEN: we pipe the value
+            const result = task.pipe(
+                map(n => '' + n),
+                map(s => s + '!'),
+                map(s => s + '!'),
+                map(s => s + '!')
+            );
+
+            // THEN: the success function should be called with the transformed value
+            result.fork(
+                jestAssertNever(cb),
+                assertFork(cb, x => expect(x).toBe('0!!!'))
+            );
+        });
+
+        it('Should work with multiple functions converting success types correctly', (cb) => {
+            // GIVEN: a resolved value
+            const task = Task.resolve(0);
+
+            // WHEN: we pipe the value
+            const result = task.pipe(
+                map(n => '' + n),
+                map(s => parseInt(s)),
+                map(n => '' + n),
+                map(s => parseInt(s))
+            );
+
+            // THEN: the success function should be called with the transformed value
+            result.fork(
+                jestAssertNever(cb),
+                assertFork(cb, x => expect(x).toBe(0))
+            );
+        });
+        it('Should be lazy (dont call if not forked)', (cb) => {
+            // GIVEN: a resolved value
+            const task = Task.resolve(0);
+
+            // and a manually created pipeable function
+            const dontCall = (t: Task<number, never>) => {
+                // This should not be called
+                expect(true).toBe(false);
+                return t;
+            };
+
+            // WHEN: we pipe the value but dont fork
+            const result = task.pipe(
+                dontCall
+            );
+
+            // THEN: the content of the task is never called
+            setTimeout(cb, 20);
+        });
+
+        it('Should handle pipeable methods that throw', (cb) => {
+            // GIVEN: a resolved value
+            const task = Task.resolve(0);
+
+            // and a manually created pipeable function
+            const pipeThatThrows = (t: Task<number, never>): Task<number, never> => {
+                throw 'oops';
+            };
+
+            // WHEN: we pipe the value
+            const result = task.pipe(
+                pipeThatThrows
+            );
+
+             // THEN: the error function should be called with the new error
+             result.fork(
+                assertFork(cb, x => {expect(x).toBeInstanceOf(UncaughtError); }),
+                jestAssertUntypedNeverCalled(cb)
+            );
+        });
+
+        it('Should be able to recover from pipeable methods that throw', (cb) => {
+            // GIVEN: a resolved value
+            const task = Task.resolve(0);
+            const task2 = Task.resolve('0');
+            // and a manually created pipeable function
+            const pipeThatThrows = (t: Task<number, never>): Task<number, never> => {
+                throw 'oops';
+            };
+
+            // WHEN: we pipe the value
+            const result = task.pipe(
+                pipeThatThrows,
+                catchError(err => task2)
+            );
+
+            // THEN: the success function should be called with the catched value
+            result.fork(
+                jestAssertNever(cb),
+                assertFork(cb, x => expect(x).toBe('0'))
+            );
+        });
+
+        it('Should be able to propagate errors correctly', (cb) => {
+            // GIVEN: a resolved value
+            const task = Task.resolve(0);
+            const rej = Task.reject('buu');
+            const task2 = Task.resolve('0');
+
+            // WHEN: we pipe the value
+            const result = task.pipe(
+                chain(val => rej),
+                // err should be of type `string | UncauchtError`
+                // The first because of rej, the second one because of chain
+                catchError(err => task2)
+            );
+
+            // THEN: the success function should be called with the catched value
+            result.fork(
+                jestAssertNever(cb),
+                assertFork(cb, x => expect(x).toBe('0'))
+            );
+        });
+    });
 });

tsconfig.json

@@ -17,6 +17,9 @@
         ]
     },
     "include": [
-        "src"
+        "src/**/*.ts",
+        "test/**/*.ts",
+        "node_modules/@types"
+
     ]
 }