 
     @_getLastProcessedSequenceNumber msg
     .then (lastProcessedSN) =>
-      if msg.sequenceNumber is (lastProcessedSN.sequenceNumber + 1)
+      if msg.payload.sequenceNumber is (lastProcessedSN.sequenceNumber + 1)
         @_doSinkMessage lastProcessedSN,
           message: msg
           sink: sink
           errors: errors
           recycleBin: recycleBin
-      else if msg.sequenceNumber <= lastProcessedSN.sequenceNumber
+      else if msg.payload.sequenceNumber <= lastProcessedSN.sequenceNumber
         @_messageHasLowSequenceNumber msg
         recycleBin.onNext msg
         sink.onCompleted()

      available

  hasLocalLock: (msg) ->
    _.contains(@localLocks, msg.id)

  takeLocalLock: (m) ->
    @stats.locallyLocked = @stats.locallyLocked + 1
    @localLocks.push m.id

  releaseLocalLock: (m) ->
    if @hasLocalLock(m)
      @stats.locallyLocked = @stats.locallyLocked - 1

    @localLocks = _.without @localLocks, m.id

  isProcessed: (msg) ->
    # TODO
    Q(_.contains(@processedMessages, msg.id))

  lockMessage: (msg) ->
    # TODO


  _checkAwaiting: (justProcessedMsg) ->
    [toSink, stillAwaiting] = _.partition @awaitingMessages, (a) =>
      @_snCacheKey(a.message.payload.resource) is @_snCacheKey(justProcessedMsg.payload.resource) and a.message.payload.sequenceNumber is (justProcessedMsg.payload.sequenceNumber + 1)

    @awaitingMessages = stillAwaiting

    _.each toSink, (s) =>
      @stats.awaitOrderingRemoved s.message
      @_doSinkMessage {sequenceNumber: justProcessedMsg.payload.sequenceNumber, cached: true}, s

  getSourceInfo: () ->
    @sphere.getSourceInfo()

  getMessageSource: () ->
    [observer, observable] = @sphere.getMessageSource()

    newObservable = observable
    .map (msg) =>
      {payload: msg, persistence: this}

    [observer, newObservable]

  checkAvaialbleForProcessingAndLockLocally: (msg) ->
    @isProcessed msg
