How to use zen-observable - 10 common examples

export default function points() {
    return new Observable(observer => {
        let cancelMouse = listen(document, 'mousemove').subscribe(
            event => observer.next(point(event.pageX, event.pageY, event.target))
        );
        let cancelFinger = listen(document, 'touchmove').subscribe(event => {
            // sum all the vectors and return the angle
            var touch = event.changedTouches[0];
            observer.next(point(touch.pageX, touch.pageY, event.target));
        });
        return () => {
            cancelMouse();
            cancelFinger();
        };
    });
}

export function Interval(tick = 1000, tickData = Symbol('tick')) {
  return new Observable(observer => {
    let timer = () => setTimeout(() => {
      if (typeof tickData === 'function') tickData = tickData()
      observer.next(tickData);
      timer()
      // observer.complete();
    }, tick);
    timer()
  
    // On unsubscription, cancel the timer
    return () => clearTimeout(timer);

  })
}

export function startWith(obs, val) {
  return new Observable(observer => {
    observer.next(val) // immediately output this value
    const handler = obs.subscribe(x => observer.next(x))
    return () => handler()
  })
}

function renderStream(element, instance, state, stateMap) {
  // this is a separate function because scope gets messy when being recursive
  var isNewStream = false; // assume no stream switching by default
  // this is the first ping of data throughout the app
  var source = Observable.of(state);
  var addToStream = function addToStream(_source) {
    // visit each source and merge with source
    if (_source) return source = merge(source, _source);
  };
  var markNewStream = function markNewStream() {
    return isNewStream = true;
  };
  var newInstance = render(source, addToStream, markNewStream)(element, instance, state, stateMap);
  return { source: source, instance: newInstance, isNewStream: isNewStream };
}

var addToStream = function addToStream(_source) {
    // visit each source and merge with source
    if (_source) return source = merge(source, _source);
  };
  var markNewStream = function markNewStream() {

var subReducer = fn(obj);
        // there are two forms of return the subreducer can have
        // straight stream form
        // or object form where we need to scan it into string
        if (subReducer.source && subReducer.reducer) {
          // object form
          subReducer = scan(subReducer.source, subReducer.reducer || function (_, n) {
            return n;
          }, _this.initialState[k]);
        }
        return subReducer.map(function (x) {
          return _defineProperty({}, k, x);
        }); // map to its particular namespace
      });
      var source = merge.apply(undefined, _toConsumableArray(sources));
      var reducer = function reducer(acc, n) {
        return _extends({}, acc, n);
      };
      return { source: source, reducer: reducer };
    }
  }]);

test('isObservable', t => {
	t.false(isObservable(null));
	t.false(isObservable({}));
	t.false(isObservable(new Listr([])));
	t.false(isObservable(new Promise(() => {})));
	t.true(isObservable(new RxObservable(() => {})));
	t.true(isObservable(new ZenObservable(() => {})));
});

it('turns an ES7 observable into a stream', () => {
    expect(Kefir.fromESObservable(Observable.of(1, 2))).to.be.observable.stream()
  })

it('outputs a compatible Observable', done => {
    const a = stream()
    const values = []
    const observable = Observable.from(a)
    observable.subscribe({
      next(x) {
        values.push(x)
      },
      complete(x) {
        expect(values).toEqual([1, 2, 3])
        done()
      },
    })
    send(a, [1, 2, 3, ''])
  })

function Test ({props}) {
      return new Observable((observer) => {
        observer.next(props.input)
      })
    }