   debounce: (delay) -> @delayChanges((changes) -> changes.debounce(delay))
   throttle: (delay) -> @delayChanges((changes) -> changes.throttle(delay))
   delayChanges: (f) -> addPropertyInitValueToStream(this, f(@changes()))
+  takeUntil: (stopper) ->
+    changes = this.changes().takeUntil(stopper)
+    addPropertyInitValueToStream(this, changes)
 
 addPropertyInitValueToStream = (property, stream) ->
   getInitValue = (property) ->

  isError: -> false
  hasValue: -> false
  filter: (f) -> true
  onDone : (listener) -> listener()

class Next extends Event
  constructor: (valueF, sourceEvent) ->

        ws = waiters
        waiters = null
        w() for w in ws
    @push = (event) =>
      if not pushing
        return if event is prevError
