     @pushing = false
     @ended = false
     @prevError = undefined
-    @unsubscribeFromSource = nop
+    @unsubSrc = undefined
 
   hasSubscribers: ->
     @subscriptions.length > 0

      subscription = { sink: sink }
      @subscriptions.push(subscription)
      if @subscriptions.length == 1
        unsubSrc = @_subscribe @handleEvent
        @unsubscribeFromSource = ->
          unsubSrc()
          @unsubscribeFromSource = nop
      assertFunction @unsubscribeFromSource
      =>
        @removeSub subscription
        @unsubscribeFromSource() unless @hasSubscribers()

    else
      @push event

  unsubscribeFromSource: ->
    @unsubSrc() if @unsubSrc
    @unsubSrc = undefined

  subscribe: (sink) =>
    if @ended
      sink end()
