function SafeObserver (observer) {
  var safeObserver = {};
  var isSubscribed = true; //this will track our subscribtion state
  var _unsubscribe;

  // this is for convinience.
  // so we can subscribe like this: stream.subscribe(val => { ...some code })
  if(typeof observer === 'function'){
    observer = { next: observer };
  }

  safeObserver.next = function(value) {
    if (!isSubscribed || !observer.next) {
      return;
    }

    try {
      observer.next(value);
    } catch (e) {
      // we want to unsubscribe only if there is an error
      safeObserver.unsubscribe();
    }
  }

  safeObserver.error = function(err) {
    if (!isSubscribed || !observer.error) {
      return;
    }

    try {
      observer.error(err);
    } catch (e) {}

    // we will unsubscribe no matter what happens
    safeObserver.unsubscribe();
  }

  safeObserver.complete = function(err) {
    if (!isSubscribed || !observer.complete) {
      return;
    }

    try {
      observer.complete();
    } catch (e) {}

    // we will unsubscribe no matter what happens
    safeObserver.unsubscribe();
  }

  safeObserver.unsubscribe = function() {
    isSubscribed = false;

    if(_unsubscribe) {
      _unsubscribe();
    }
  }

  safeObserver.setUnsubscribe = function(unsub) {
    _unsubscribe = unsub;
  }

  return safeObserver;
}

function Observable (subscriber) {
  return {
    subscribe: observer => subscriber(SafeObserver(observer)),
    pipe: function (...fns) {
      var initialSource = this;
      return fns.reduce((source, fn) => fn(source), initialSource);
    }
  } 
}

// === Factories ===
function fromEvent(element, eventType) {
  return Observable(function(observer) {
    var handler = ev => observer.next(ev);
    element.addEventListener(eventType, handler);
    
    observer.setUnsubscribe(() => {
      element.removeEventListener(eventType, handler);
    });
    
    return observer.unsubscribe;
  });
}

function interval(delay) {
  return Observable(function(observer) {
    var counter   = 0;
    var callback  = () => observer.next(counter++);
    var _interval = setInterval(callback, delay);
    
    observer.setUnsubscribe(() => clearInterval(_interval));
    
    return observer.unsubscribe;
  });
}

// === Operators ===
function map(transformFn) {
  return source$ => Observable(observer => 
    source$.subscribe(value => observer.next(
      transformFn(value)
    ))
  );
}

function filter(predicateFn) {
  return source$ => Observable(observer => 
    source$.subscribe(value => {
      if(predicateFn(value)){
        observer.next(value);
      }
    })
  );
}

function reduce(reducerFn, seed) {
  return source$ => Observable(observer => {
    var state = seed;
    source$.subscribe(value => {
      state = reducerFn(state, value);
      observer.next(state);
    })
  });
}

function take(number) {
  return source$ => Observable(function(observer) {
    var count = 0;
    var unsubscribeSource = source$.subscribe(function(value) {
      count++;
      observer.next(value);        
      if (count === number) {
        observer.complete();
        unsubscribeSource();
      }
    });
  });
}

function stopWhen(predicateFn) {
  return source$ => Observable(function(observer) {
    var unsubscribeSource =source$.subscribe(function(value) {
      observer.next(value);
      if(predicateFn(value)){
        observer.complete();
        unsubscribeSource();
      }
    })
  });
}

function emitUntil(notifier$) {
  return source$ => Observable(function(observer) {
    var unsubscribeSource = source$.subscribe(observer);
    notifier$.subscribe(function() {
      unsubscribeSource();
      observer.complete();
    });
  });
}

// === Demo ===
var inputText     = document.getElementById('thing');
var textDisplay   = document.querySelector('.formname');