     "#{res.typeId}-#{res.id}"
 
   _getLastProcessedSequenceNumber: (msg) ->
-    cached = @sequenceNumberCache.get @_snCacheKey(msg.resource)
+    cached = @sequenceNumberCache.get @_snCacheKey(msg.payload.resource)
 
     if cached?
       Q({sequenceNumber: cached, cached: true})
     else
-      @sphere.getLastProcessedSequenceNumber msg.resource
+      @sphere.getLastProcessedSequenceNumber msg.payload.resource
       .then (sn) =>
-        alreadyInCache = @sequenceNumberCache.get @_snCacheKey(msg.resource)
+        alreadyInCache = @sequenceNumberCache.get @_snCacheKey(msg.payload.resource)
 
         if not alreadyInCache? or alreadyInCache < sn
-          @sequenceNumberCache.set @_snCacheKey(msg.resource), sn
+          @sequenceNumberCache.set @_snCacheKey(msg.payload.resource), sn
 
         {sequenceNumber: sn, cached: false}
 
 class MessageProcessor
-  constructor: (@stats, @messageService, @persistenceService, options) ->
-    @messageProcessors = options.processors # Array[Message => Promise[Something]]
-    @tickDelay = options.tickDelay or 500
+  constructor: (@stats, options) ->
+    @messageProcessors = options.processors # Array[(SourceInfo, Message) => Promise[Anything]]
+    @messageSources = options.messageSources
+    @heartbeatInterval = options.heartbeatInterval or 500
 
     @recycleBin = @_createRecycleBin()
     @errors = @_createErrorProcessor(@recycleBin)
     @unrecoverableErrors = @_createUnrecoverableErrorProcessor(@recycleBin)
 
   run: () ->
-    all = @_allMessages()
+    heartbeat = Rx.Observable.interval @heartbeatInterval
+    .filter (tick) =>
+      not @stats.applyBackpressureAtTick(tick)
+
+    messageSources = _.map @messageSources, (source) ->
+      [sourceObserver, sourceObservable] = source.getMessageSource()
+      heartbeat.subscribe sourceObserver
+      sourceObservable
+
+    all = Rx.Observable.merge messageSources
+    .map (msg) =>
+      @stats.incommingMessage msg
+
     maybeLocked = @_filterAndLockNewMessages(all)
     processed = @_doProcessMessages(maybeLocked)
 

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

      @persistenceService.reportSuccessfullProcessing msg
      .then (msg) ->
        subj.onNext(msg)
        subj.onCompleted()
      .fail (error) ->
        @errors.onNext {message: msg, error: error, processor: "Reporting successful processing"}
        subj.onComplete()
      .done()

  _processMessage: (processors, msg) ->
    try
      promises = _.map processors, (processor) ->
        processor msg.persistence.getSourceInfo(), msg.payload

      Q.all promises
    catch error
      Q.reject(error)

  _filterAndLockNewMessages: (messages) ->
    [newMessages, other, errors]  = @_split messages, (msg) ->
      msg.persistence.checkAvaialbleForProcessingAndLockLocally msg

    other.subscribe @recycleBin
    errors.subscribe @unrecoverableErrors
