 
       subj
 
-  _allMessages: () =>
-    Rx.Observable.interval(@tickDelay)
-    .filter (tick) =>
-      not @stats.applyBackpressureAtTick(tick)
-    .flatMap =>
-      @messageService.getMessages()
-    .map (msg) =>
-      @stats.incommingMessage msg
-
   _createRecycleBin: () ->
     recycleBin = new Rx.Subject()
 
     recycleBin
     .subscribe (msg) =>
-      @persistenceService.releaseLocalLock msg
+      msg.persistence.releaseLocalLock msg
       @stats.messageFinished msg
 
     recycleBin

      # We've done it!! We processed message successfully! (let's hope we will also unlock it successfully too)
      @stats.yay msg

      alreadyInCache = @sequenceNumberCache.get @_snCacheKey(msg.resource)

      if not alreadyInCache? or alreadyInCache < msg.sequenceNumber
        @sequenceNumberCache.set @_snCacheKey(msg.resource), msg.sequenceNumber
        @_checkAwaiting msg

      msg

  _messageHasLowSequenceNumber: (msg) ->
    console.info "WARN: message has appeared twice for processing (something wrong in the pipeline)", msg

  _doSinkMessage: (sn, box) ->
    doSink = () ->

    .flatMap (msg) =>
      subj = new Rx.Subject()

      msg.persistence.unlockMessage msg
      .then (msg) ->
        subj.onNext(msg)
        subj.onCompleted()
      .fail (error) =>
        @errors.onNext {message: msg, error: error, processor: "Unlocking the message"}
        subj.onComplete()
      .done()
