       d.subscribe(sink) unless reply == Bacon.noMore
     new Property(subscribe)
 
-  distinctUntilChanged: ->
-    @withStateMachine undefined, (prev, event) ->
-      if !event.hasValue()
-        [prev, [event]]
-      else if prev isnt event.value
-        [event.value, [event]]
-      else
-        [prev, []]
-
-  withStateMachine: (initState, f) ->
-    state = initState
-    @withHandler (event) ->
-      fromF = f(state, event)
-      assertArray fromF
-      [newState, outputs] = fromF
-      assertArray outputs
-      state = newState
-      reply = Bacon.more
-      for output in outputs
-        reply = @push output
-        if reply == Bacon.noMore
-          return reply
-      reply
-
   decorateWith: (label, property) ->
     property.sampledBy(this, (propertyValue, streamValue) ->
         result = cloneObject(streamValue)

      =>
        removeSink sink
        unsubscribeFromSource() unless @hasSubscribers()
  toEventStream: -> new EventStream(@subscribe)
  toString: -> "Dispatcher"

class Bus extends EventStream
  constructor: ->

  onError: (f) -> @subscribe (event) ->
    f event.error if event.isError()
  errors: -> @filter(-> false)
  filter: (f) ->
    @withHandler (event) -> 
      if event.filter(f)
        @push event
      else
        Bacon.more
  takeWhile: (f) ->
    @withHandler (event) -> 
      if event.filter(f)
        @push event
      else
        @push end()
        Bacon.noMore
  endOnError: ->
    @withHandler (event) ->
      if event.isError()
        @push event
        @push end()
      else
        @push event
  take: (count) ->
    assert "take: count must >= 1", (count>=1)
    @withHandler (event) ->
      if !event.hasValue()
        @push event
      else if (count == 1)
        @push event
        @push end()
        Bacon.noMore
      else
        count--
        @push event
  map: (f) ->
    @withHandler (event) -> 
      @push event.fmap(f)
  takeUntil: (stopper) =>
    src = this
    @withSubscribe (sink) ->
        unsubscribed = false
        unsubSrc = nop
        unsubStopper = nop
        unsubBoth = -> unsubSrc() ; unsubStopper() ; unsubscribed = true
        srcSink = (event) ->
          if event.isEnd()
            unsubStopper()
          reply = sink event
          if reply == Bacon.noMore
            unsubStopper()
          reply
        stopperSink = (event) ->
          if event.isError()
            Bacon.more
          else if event.isEnd()
            Bacon.noMore
          else
            unsubSrc()
            sink end()
            Bacon.noMore
        unsubStopper = stopper.subscribe(stopperSink)
        unsubSrc = src.subscribe(srcSink) unless unsubscribed
        unsubBoth
  skip : (count) ->
    assert "skip: count must >= 0", (count>=0)
    @withHandler (event) ->
      if !event.hasValue()
        @push event
      else if (count > 0)
        count--
        Bacon.more
      else
        @push event
  distinctUntilChanged: ->
    @withStateMachine undefined, (prev, event) ->
      if !event.hasValue()
        [prev, [event]]
      else if prev isnt event.value
        [event.value, [event]]
      else
        [prev, []]

  withStateMachine: (initState, f) ->
    state = initState
    @withHandler (event) ->
      fromF = f(state, event)
      assertArray fromF
      [newState, outputs] = fromF
      assertArray outputs
      state = newState
      reply = Bacon.more
      for output in outputs
        reply = @push output
        if reply == Bacon.noMore
          return reply
      reply


class EventStream extends Observable
  constructor: (subscribe) ->
