[add] Basic methods for Pipe

Author: TechQuery

package.json

@@ -1,6 +1,6 @@
 {
-    "name": "waterwheel",
-    "version": "1.0.0-alpha.1",
+    "name": "iterable-observer",
+    "version": "1.0.0-beta.0",
     "license": "LGPL-3.0",
     "author": "[email protected]",
     "description": "Observable Proposal implement based on Async Generator (ES 2018) & TypeScript",
@@ -13,28 +13,28 @@
         "generator",
         "iterator"
     ],
-    "homepage": "https://web-cell.dev/WaterWheel/",
+    "homepage": "https://web-cell.dev/iterable-observer/",
     "repository": {
         "type": "git",
-        "url": "git+https://github.com/EasyWebApp/WaterWheel.git"
+        "url": "git+https://github.com/EasyWebApp/iterable-observer.git"
     },
     "bugs": {
-        "url": "https://github.com/EasyWebApp/WaterWheel/issues"
+        "url": "https://github.com/EasyWebApp/iterable-observer/issues"
     },
     "source": "source/index.ts",
     "types": "dist/index.d.ts",
-    "main": "dist/waterwheel.umd.js",
-    "module": "dist/waterwheel.js",
+    "main": "dist/iterable-observer.umd.js",
+    "module": "dist/iterable-observer.js",
     "devDependencies": {
         "@types/jest": "^24.0.23",
-        "husky": "^3.0.9",
+        "husky": "^3.1.0",
         "jest": "^24.9.0",
         "lint-staged": "^9.4.3",
         "microbundle": "^0.11.0",
         "open-cli": "^5.0.0",
         "prettier": "^1.19.1",
         "ts-jest": "^24.2.0",
-        "typedoc": "^0.15.2",
+        "typedoc": "^0.15.3",
         "typescript": "^3.7.2"
     },
     "prettier": {
@@ -60,8 +60,8 @@
     "scripts": {
         "test": "lint-staged && jest",
         "debug": "node --inspect node_modules/jest/bin/jest --runInBand",
-        "pack-docs": "typedoc --name WaterWheel --out docs/ source/",
-        "build": "microbundle  &&  npm run pack-docs",
+        "pack-docs": "typedoc --name \"Iterable Observer\" --out docs/ source/",
+        "build": "rm -rf dist/  &&  microbundle  &&  npm run pack-docs",
         "help": "npm run pack-docs  &&  open-cli docs/index.html",
         "prepublishOnly": "npm test  &&  npm run build"
     },

source/Observable.ts

@@ -3,7 +3,7 @@ import { Defer, makeDefer } from './utility';
 export interface Observer<T = any> {
     next(value: T): void;
     error(reason: string | Error): void;
-    complete(value: T): void;
+    complete(): void;
 }
 
 export type SubscriberFunction = (observer: Observer) => (() => void) | void;
@@ -29,29 +29,68 @@ export class Observable<T = any> {
                 queue.push(makeDefer());
             },
             error(reason) {
-                if (done) return;
-
-                queue[queue.length - 1].reject(reason), (done = true);
+                if (!done)
+                    queue[queue.length - 1].reject(reason), (done = true);
 
                 if (canceler) canceler();
             },
-            complete(value) {
-                if (done) return;
-
-                queue[queue.length - 1].resolve(value), (done = true);
+            complete() {
+                if (!done) queue[queue.length - 1].resolve(), (done = true);
 
                 if (canceler) canceler();
             }
         };
 
         canceler = this.subscriber(observer);
 
-        while (true) {
+        do {
             yield queue[0].promise;
 
             queue.shift();
+        } while (queue[0] || !done);
+    }
+
+    static of<T = any>(...items: T[]) {
+        return new this<T>(({ next, complete }) => {
+            for (const item of items) next(item);
+
+            complete();
+        });
+    }
+
+    subscribe(
+        onNext: Observer<T>['next'],
+        onError?: Observer<T>['error'],
+        onComplete?: Observer<T>['complete']
+    ) {
+        var stop = false;
+
+        (async () => {
+            try {
+                for await (const item of this)
+                    if (!stop) onNext(item);
+                    else break;
+
+                if (onComplete instanceof Function) onComplete();
+            } catch (error) {
+                if (onError instanceof Function) onError(error);
+            }
+        })();
+
+        return {
+            unsubscribe() {
+                stop = true;
+            },
+            get closed() {
+                return stop;
+            }
+        };
+    }
 
-            if (done) break;
-        }
+    static from<T = any>(observable: Observable<T>) {
+        return new this<T>(
+            ({ next, error, complete }) =>
+                observable.subscribe(next, error, complete).unsubscribe
+        );
     }
 }

source/utility.ts

@@ -1,6 +1,6 @@
 export interface Defer<T = any> {
     promise: Promise<T>;
-    resolve: (data: T) => void;
+    resolve: (data?: T) => void;
     reject: (error: Error | string) => void;
 }
 

test/index.spec.ts

@@ -1,46 +1,94 @@
 import { Observable, Observer } from '../source';
 
-describe('Observable', () => {
-    var observable: Observable, timer: any;
+function createExample() {
+    var timer: any,
+        count = 0;
 
     const cleaner = jest.fn(() => clearInterval(timer));
 
-    it('should not call Subscriber Function while constructing', () => {
-        var count = 0;
+    const subscriber = jest.fn(({ next, complete }: Observer) => {
+        timer = setInterval(() => (++count < 6 ? next(count) : complete()), 0);
 
-        const subscriber = jest.fn(({ next, complete }: Observer) => {
-            timer = setInterval(
-                () => (++count < 5 ? next(count) : complete(count)),
-                0
-            );
+        return cleaner;
+    });
+
+    return { subscriber, cleaner };
+}
+
+describe('Observable', () => {
+    describe('Single', () => {
+        const { subscriber, cleaner } = createExample();
+        var observable: Observable;
 
-            return cleaner;
+        it('should not call Subscriber Function while constructing', () => {
+            observable = new Observable<number>(subscriber);
+
+            expect(subscriber).toBeCalledTimes(0);
         });
 
-        observable = new Observable<number>(subscriber);
+        it('should iterate Limited items asynchronously', async () => {
+            const list = [];
 
-        expect(subscriber).toBeCalledTimes(0);
-    });
+            for await (const item of observable) list.push(item);
 
-    it('should iterate Limited items asynchronously', async () => {
-        const list = [];
+            expect(list).toEqual(expect.arrayContaining([1, 2, 3, 4, 5]));
 
-        for await (const item of observable) list.push(item);
+            expect(cleaner).toBeCalledTimes(1);
+        });
 
-        expect(list).toEqual(expect.arrayContaining([1, 2, 3, 4, 5]));
+        it('should throw Error after error() called', () => {
+            const observable = new Observable(({ error }) => {
+                error(new Error('test'));
+            });
 
-        expect(cleaner).toBeCalledTimes(1);
+            expect(
+                (async () => {
+                    for await (const item of observable);
+                })()
+            ).rejects.toStrictEqual(new Error('test'));
+        });
     });
 
-    it('should throw Error after error() called', () => {
-        const observable = new Observable(({ error }) => {
-            error(new Error('test'));
+    describe('Pipe', () => {
+        it('should construct an Observable from Static data', async () => {
+            const observable = Observable.of<number>(1, 2, 3),
+                list = [];
+
+            for await (const item of observable) list.push(item);
+
+            expect(list).toEqual(expect.arrayContaining([1, 2, 3]));
         });
 
-        expect(
-            (async () => {
-                for await (const item of observable);
-            })()
-        ).rejects.toStrictEqual(new Error('test'));
+        it('should invoke handlers after subscribing', async () => {
+            const { subscriber, cleaner } = createExample();
+
+            const observable = new Observable<number>(subscriber),
+                onNext = jest.fn();
+
+            await new Promise(resolve =>
+                observable.subscribe(onNext, null, resolve)
+            );
+
+            expect(onNext).toHaveBeenNthCalledWith(1, 1);
+            expect(onNext).toHaveBeenNthCalledWith(2, 2);
+            expect(onNext).toHaveBeenNthCalledWith(3, 3);
+            expect(onNext).toHaveBeenNthCalledWith(4, 4);
+            expect(onNext).toHaveBeenNthCalledWith(5, 5);
+
+            expect(cleaner).toBeCalledTimes(1);
+        });
+
+        it('should construct an Observable from an exist Observable', async () => {
+            const { subscriber } = createExample();
+
+            const old = new Observable<number>(subscriber);
+
+            const observable = Observable.from<number>(old),
+                list = [];
+
+            for await (const item of observable) list.push(item);
+
+            expect(list).toEqual(expect.arrayContaining([1, 2, 3, 4, 5]));
+        });
     });
 });