[add] fromEvent() & related interfaces

[add] toPromise() utility

Author: TechQuery

.npmignore

@@ -0,0 +1,5 @@
+.editorconfig
+test/
+.rts2_cache_*/
+docs/
+.travis.yml

.travis.yml

@@ -0,0 +1,22 @@
+branches:
+    only:
+        - master
+
+language: node_js
+node_js:
+    - lts/*
+cache:
+    directories:
+        - node_modules
+
+install:
+    - npm install
+script:
+    - npm run ${SCRIPT}
+deploy:
+    provider: pages
+    on:
+        branch: master
+    skip_cleanup: true
+    local_dir: ${FOLDER}
+    github_token: ${TOKEN}

ReadMe.md

@@ -1,16 +1,23 @@
-# WaterWheel
+# Iterable Observer
 
 [Observable Proposal][1] implement based on [Async Generator (ES 2018)][2] & [TypeScript][3]
 
+[![Build Status](https://travis-ci.com/EasyWebApp/iterable-observer.svg?branch=master)][4]
+[![](https://data.jsdelivr.com/v1/package/npm/iterable-observer/badge?style=rounded)][5]
+
+[![NPM](https://nodei.co/npm/iterable-observer.png?downloads=true&downloadRank=true&stars=true)][6]
+
 ## Usage
 
+### Basic
+
 ```javascript
-import { Observable } from 'waterwheel';
+import { Observable } from 'iterable-observer';
 
 var count = 0,
     list = [];
 
-const observable = new Observable(({ next, complete }: Observer) => {
+const observable = new Observable(({ next, complete }) => {
     const timer = setInterval(
         () => (++count < 5 ? next(count) : complete(count)),
         0
@@ -26,6 +33,27 @@ const observable = new Observable(({ next, complete }: Observer) => {
 })();
 ```
 
+### Enhance Run-time platforms
+
+```javascript
+import { Observable } from 'iterable-observer';
+
+const reader = new FileReader(),
+    {
+        files: [file]
+    } = document.querySelector('input[type="file"]');
+
+reader.readAsBlob(file);
+
+(async () => {
+    for await (const { loaded } of Observable.fromEvent(reader, 'progress'))
+        console.log((loaded / file.size) * 100 + '%');
+})();
+```
+
 [1]: https://github.com/tc39/proposal-observable
 [2]: https://tc39.es/ecma262/#sec-asyncgeneratorfunction-objects
 [3]: https://www.typescriptlang.org/
+[4]: https://travis-ci.com/EasyWebApp/iterable-observer
+[5]: https://www.jsdelivr.com/package/npm/iterable-observer
+[6]: https://nodei.co/npm/iterable-observer/

source/Observable.ts

@@ -6,15 +6,42 @@ export interface Observer<T = any> {
     complete(): void;
 }
 
+export interface Subscription {
+    unsubscribe(): void;
+    readonly closed: boolean;
+}
+
+export interface Subscribable<T = any> {
+    [Symbol.observable](): Subscribable<T>;
+    subscribe(
+        onNext: Observer<T>['next'],
+        onError?: Observer<T>['error'],
+        onComplete?: Observer<T>['complete']
+    ): Subscription;
+}
+
 export type SubscriberFunction = (observer: Observer) => (() => void) | void;
 
-export class Observable<T = any> {
+export type EventHandler = (data: any) => void;
+
+export interface EventTrigger {
+    addEventListener?(name: string, handler: EventHandler): void;
+    removeEventListener?(name: string, handler: EventHandler): void;
+    on?(name: string, handler: EventHandler): this;
+    off?(name: string, handler: EventHandler): this;
+}
+
+export class Observable<T = any> implements Subscribable {
     private subscriber: SubscriberFunction;
 
     constructor(subscriber: SubscriberFunction) {
         this.subscriber = subscriber;
     }
 
+    [Symbol.observable]() {
+        return this;
+    }
+
     async *[Symbol.asyncIterator]() {
         var queue: Defer<T>[] = [makeDefer<T>()],
             canceler: (() => void) | void,
@@ -58,6 +85,18 @@ export class Observable<T = any> {
         });
     }
 
+    async toPromise() {
+        const stack = [];
+
+        for await (const item of this) {
+            stack.push(item);
+
+            if (stack.length > 2) stack.shift();
+        }
+
+        return stack[0];
+    }
+
     subscribe(
         onNext: Observer<T>['next'],
         onError?: Observer<T>['error'],
@@ -87,10 +126,30 @@ export class Observable<T = any> {
         };
     }
 
-    static from<T = any>(observable: Observable<T>) {
+    static from<T = any>(observable: Subscribable<T>) {
         return new this<T>(
             ({ next, error, complete }) =>
                 observable.subscribe(next, error, complete).unsubscribe
         );
     }
+
+    static fromEvent<T = any>(target: EventTrigger, name: string) {
+        return new this<T>(({ next, error }) => {
+            if (typeof target.on === 'function')
+                target.on(name, next).on('error', error);
+            else {
+                target.addEventListener(name, next);
+                target.addEventListener('error', error);
+            }
+
+            return () => {
+                if (typeof target.off === 'function')
+                    target.off(name, next).off('error', error);
+                else {
+                    target.removeEventListener(name, next);
+                    target.removeEventListener('error', error);
+                }
+            };
+        });
+    }
 }

test/index.spec.ts

@@ -1,4 +1,5 @@
 import { Observable, Observer } from '../source';
+import { EventEmitter } from 'events';
 
 function createExample() {
     var timer: any,
@@ -59,6 +60,12 @@ describe('Observable', () => {
             expect(list).toEqual(expect.arrayContaining([1, 2, 3]));
         });
 
+        it('should convert to a Promise', () => {
+            const observable = Observable.of<number>(1, 2, 3);
+
+            expect(observable.toPromise()).resolves.toBe(3);
+        });
+
         it('should invoke handlers after subscribing', async () => {
             const { subscriber, cleaner } = createExample();
 
@@ -91,4 +98,29 @@ describe('Observable', () => {
             expect(list).toEqual(expect.arrayContaining([1, 2, 3, 4, 5]));
         });
     });
+
+    describe('3th Party Platform interfaces', () => {
+        it('should listen Event Emitters', async () => {
+            const target = new EventEmitter();
+            var count = 0;
+
+            const observable = Observable.fromEvent(target, 'test'),
+                list = [];
+
+            const timer = setInterval(() => {
+                if (++count < 4) target.emit('test', count);
+                else {
+                    clearInterval(timer);
+                    target.emit('error', 'example');
+                }
+            }, 0);
+
+            try {
+                for await (const item of observable) list.push(item);
+            } catch (error) {
+                expect(list).toEqual(expect.arrayContaining([1, 2, 3]));
+                expect(error).toBe('example');
+            }
+        });
+    });
 });