Array function based Steam & EventStream implementation. Also handles promises.

ArrayLikeObservable.es6

class Observable {

  constructor(fn) {

    this._fn = fn;
    this._children = [];

    const handler = {
      get(target, prop) {
        return prop in target || !(prop in Array.prototype)
          ? target[prop]
          : target._getHandler(prop);
      }
    };

    return new Proxy(this, handler);

  }

  static fromEvent(target, event, options = {}) {

    const observable = new Observable()
    const eventHandler = v => observable.push(v);

    target.addEventListener(event, eventHandler, options);
    observable.cleanup = () => target.removeEventListener(event, eventHandler, options);

    return observable;

  }

  set cleanup(fn) {
    this._cleanup = fn;
  }

  push(value) {

    if (this._complete) return;

    // Resolve if necessary
    const promise = value instanceof Promise ? value : Promise.resolve(value);

    promise.then(value_ => {

      // Execute function with value_ or use value_ as result directly
      const result = this._fn ? this._fn(value_) : value_;

      // Pass value on to children
      this._children.forEach(observable => observable.push(result))

    })

  }

  end() {
    this._cleanup && this._cleanup();
    this._children.forEach(stream => stream.end());
    this._children = [];
    this._complete = true;
  }

  _getHandler(method) {

    // Return a function which uses 'method' as array function
    return (...args) => this._chain(v => {

      const result = this._box(v)[method](...args);

      // When the result of the array function is another array, return the first value
      // Return the whole value otherwise
      return Array.isArray(result) ? result.shift() : result;

    })

  }

  _box(v) {
    return typeof v === 'undefined' ? [] : [v];
  }

  _chain(fn) {
    const observable = new Observable(fn);
    this._children.push(observable);
    return observable;
  }

}

export default Observable;

example.es6

const SPACE = 'SPACE';
const KEYS = { 32: SPACE };

const keyStream = EventStream.of(document.documentElement, 'keypress')
const spaceKeyStream = keyStream
	.map(e => e.which)
	.map(code => KEYS[code])
	.filter(key => key === SPACE);

spaceKeyStream.forEach(v => console.log(v));