 
     _.each toSink, (s) =>
       @stats.awaitOrderingRemoved s.message
-
-      s.sink.onNext s.message
-      s.sink.onCompleted()
-      s.errors.onCompleted()
+      @_doSinkMessage {sequenceNumber: justProcessedMsg.sequenceNumber, cached: true}, s
 
   checkAvaialbleForProcessingAndLockLocally: (msg) ->
     @isProcessed msg

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

  orderBySequenceNumber: (msg, lock, recycleBin) ->
    sink = new Rx.Subject()
    errors = new Rx.Subject()

    @_getLastProcessedSequenceNumber msg
    .then (lastProcessedSN) =>
      if msg.sequenceNumber is (lastProcessedSN + 1)
        sink.onNext msg
        sink.onCompleted()
        errors.onCompleted()
      else if msg.sequenceNumber <= lastProcessedSN
        @_messageHasLowSequenceNumber msg
        recycleBin.onNext msg
        sink.onCompleted()

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

    if cached?
      Q({sequenceNumber: cached, cached: true})
    else
      @sphere.getLastProcessedSequenceNumber msg.resource
      .then (sn) =>
