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

    @pushing = false
    @ended = false
    @prevError = undefined
    @unsubscribeFromSource = nop

  hasSubscribers: ->
    @subscriptions.length > 0

    else
      @push event

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

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