   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) ->

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

class Bus extends EventStream
  constructor: ->

      combine(sampler, nop, pushPropertyValue).changes().takeUntil(sampler.end())
  sample: (interval) =>
    @sampledBy Bacon.interval(interval, {})
  changes: => new EventStream (sink) =>
    @subscribe (event) =>
      sink event unless event.isInitial()
  withHandler: (handler) ->
    new Property(new PropertyDispatcher(@subscribe, handler).subscribe)
  withSubscribe: (subscribe) -> new Property(new PropertyDispatcher(subscribe).subscribe)
  toProperty: => this

class Dispatcher
