const Observable = function(producer) {

  this.subscribe = producer

}

​

​

Observable.interval = function(period){

​

  const intervalProducer = function(observer) {

    let counter = 0

​

    const unsubscribe = function() {

      clearInterval(timer)

    }

​

    const timer = setInterval(() => {

      try {

        observer.next(counter++)

      } catch(err) {

        unsubscribe() // <- on error we also want to unsusbcribe

        observer.error(err)

      }

    }, period)

​

    return {

      unsubscribe

    }

  }

​

 return new Observable(intervalProducer)

}

​

​

Observable.prototype.takeUntil = function(predicate){

  // Unwrap the producer from the observable

  const originalProducer = this.subscribe;

  // Hijack it so that it will push values until the predicate predicate test pass

  const newProducer = function(observer){

    const sequence = originalProducer({

      next (value) {

        if (predicate(value) !== true) {

          observer.next(value)

        } else {

          sequence.unsubscribe()

        }

      },

      error (err) {observer.error(err)},

      complete () {observer.complete()}

    })

  }

​

  // Return a new observable with the hijacked producer.

  return new Observable(newProducer)

}

​

​

const isGreaterThenThree = (num) => num > 3 === true

​

const countTillThree$ = Observable.interval(1000)

  .takeUntil(isGreaterThenThree)

​

const sub = countTillThree$.subscribe({

  next(value){console.log(value)},

  complete(){console.log('done')},

  error(err){console.log(err)}

})

​