Iterable Observer

Observable Proposal implement based on Async Generator (ES 2018) & TypeScript

Usage

Basic

import { Observable } from 'iterable-observer';

var count = 0,
    list = [];

const observable = new Observable(({ next, complete }) => {
    const timer = setInterval(
        () => (++count < 5 ? next(count) : complete(count)),
        0
    );
    return () => clearInterval(timer);
});

(async () => {
    for await (const item of observable) list.push(item);

    console.log(list); // [1, 2, 3, 4, 5]
})();
Copy

Enhance Run-time platforms

Transform events

import { Observable } from 'iterable-observer';

const reader = new FileReader(),
    { files } = document.querySelector('input[type="file"]');

reader.readAsBlob(files[0]);

(async () => {
    for await (const { loaded } of Observable.fromEvent(reader, 'progress'))
        console.log((loaded / file.size) * 100 + '%');
})();
Copy

Transform streams

import { Observable } from 'iterable-observer';

(async () => {
    const { body } = await fetch('https://example.com/path/to/blob'),
        chunks: Uint8Array[] = [];

    for await (const chunk of Observable.fromStream(body)) chunks.push(chunk);

    const blob = new Blob(chunks);

    console.log(blob);
}();
Copy

Concurrent Task to Serial Queue

import { createQueue } from 'iterable-observer';
import Koa from 'koa';
import BodyParser from 'koa-bodyparser';

const { process, observable } = createQueue(),
    app = new Koa();

(async () => {
    for await (const { defer, data } of observable)
        defer.resolve(JSON.stringify(data));
})();

app.use(BodyParser)
    .use(async context => (context.body = await process(context.request.body)))
    .listen(80);
Copy