|
|
@@ -10,6 +10,7 @@ Metrics = require "./Metrics"
|
|
|
Errors = require "./Errors"
|
|
|
DocumentManager = require "./DocumentManager"
|
|
|
RangesManager = require "./RangesManager"
|
|
|
+Profiler = require "./Profiler"
|
|
|
|
|
|
module.exports = UpdateManager =
|
|
|
processOutstandingUpdates: (project_id, doc_id, callback = (error) ->) ->
|
|
|
@@ -20,13 +21,17 @@ module.exports = UpdateManager =
|
|
|
callback()
|
|
|
|
|
|
processOutstandingUpdatesWithLock: (project_id, doc_id, callback = (error) ->) ->
|
|
|
+ profile = new Profiler("processOutstandingUpdatesWithLock", {project_id, doc_id})
|
|
|
LockManager.tryLock doc_id, (error, gotLock, lockValue) =>
|
|
|
return callback(error) if error?
|
|
|
return callback() if !gotLock
|
|
|
+ profile.log("tryLock")
|
|
|
UpdateManager.processOutstandingUpdates project_id, doc_id, (error) ->
|
|
|
return UpdateManager._handleErrorInsideLock(doc_id, lockValue, error, callback) if error?
|
|
|
+ profile.log("processOutstandingUpdates")
|
|
|
LockManager.releaseLock doc_id, lockValue, (error) =>
|
|
|
return callback(error) if error?
|
|
|
+ profile.log("releaseLock").end()
|
|
|
UpdateManager.continueProcessingUpdatesWithLock project_id, doc_id, callback
|
|
|
|
|
|
continueProcessingUpdatesWithLock: (project_id, doc_id, callback = (error) ->) ->
|
|
|
@@ -38,13 +43,20 @@ module.exports = UpdateManager =
|
|
|
callback()
|
|
|
|
|
|
fetchAndApplyUpdates: (project_id, doc_id, callback = (error) ->) ->
|
|
|
+ profile = new Profiler("fetchAndApplyUpdates", {project_id, doc_id})
|
|
|
RealTimeRedisManager.getPendingUpdatesForDoc doc_id, (error, updates) =>
|
|
|
return callback(error) if error?
|
|
|
if updates.length == 0
|
|
|
return callback()
|
|
|
- async.eachSeries updates,
|
|
|
- (update, cb) -> UpdateManager.applyUpdate project_id, doc_id, update, cb
|
|
|
- callback
|
|
|
+ profile.log("getPendingUpdatesForDoc")
|
|
|
+ doUpdate = (update, cb)->
|
|
|
+ UpdateManager.applyUpdate project_id, doc_id, update, (err) ->
|
|
|
+ profile.log("applyUpdate")
|
|
|
+ cb(err)
|
|
|
+ finalCallback = (err) ->
|
|
|
+ profile.log("async done").end()
|
|
|
+ callback(err)
|
|
|
+ async.eachSeries updates, doUpdate, finalCallback
|
|
|
|
|
|
applyUpdate: (project_id, doc_id, update, _callback = (error) ->) ->
|
|
|
callback = (error) ->
|
|
|
@@ -66,14 +78,19 @@ module.exports = UpdateManager =
|
|
|
HistoryManager.recordAndFlushHistoryOps project_id, doc_id, appliedOps, historyOpsLength, callback
|
|
|
|
|
|
lockUpdatesAndDo: (method, project_id, doc_id, args..., callback) ->
|
|
|
+ profile = new Profiler("lockUpdatesAndDo", {project_id, doc_id})
|
|
|
LockManager.getLock doc_id, (error, lockValue) ->
|
|
|
+ profile.log("getLock")
|
|
|
return callback(error) if error?
|
|
|
UpdateManager.processOutstandingUpdates project_id, doc_id, (error) ->
|
|
|
return UpdateManager._handleErrorInsideLock(doc_id, lockValue, error, callback) if error?
|
|
|
+ profile.log("processOutStandingUpdates")
|
|
|
method project_id, doc_id, args..., (error, response_args...) ->
|
|
|
return UpdateManager._handleErrorInsideLock(doc_id, lockValue, error, callback) if error?
|
|
|
+ profile.log("method")
|
|
|
LockManager.releaseLock doc_id, lockValue, (error) ->
|
|
|
return callback(error) if error?
|
|
|
+ profile.log("releaseLock").end()
|
|
|
callback null, response_args...
|
|
|
# We held the lock for a while so updates might have queued up
|
|
|
UpdateManager.continueProcessingUpdatesWithLock project_id, doc_id
|