Consuming Observables with a Generator

consuming-observable-with-generator.js

(function() { 'use strict';

function consume(observable, gen) {

    return new Promise((resolve, reject) => {
            
        if (typeof gen === "function") {
            // Prime the generator to accept values
            gen = gen();
            gen.next();
        }

        observable.subscribe({
            next(v) { send(v, "next") },
            error(e) { send(e, "throw") },
            complete(x) { send(x, "return") },
        });

        function send(v, type) {

            try { 

                let result = gen[type](v);

                if (result.done)
                    resolve(result.value);

            } catch (e) { reject(e) }
        }

    });
}

function listen(element, eventName, capture) {
    return new Observable(sink => {
        let handler = e => sink.next(e);
        element.addEventListener(eventName, handler, Boolean(capture));
        return _=> element.removeEventListener(eventName, handler, Boolean(capture));
    });
}

let clicks = listen(document.body, "click");

consume(clicks, function*() {

    let evt;

    while (evt = yield)
        console.log(`[Click] ${ evt.x }:${ evt.y }`);
});

})();