|
@@ -144,38 +144,6 @@ module.exports = UpdatesManager =
|
|
|
UpdatesManager._processUncompressedUpdatesForDocWithLock project_id, doc_id, temporary, cb
|
|
UpdatesManager._processUncompressedUpdatesForDocWithLock project_id, doc_id, temporary, cb
|
|
|
async.parallelLimit jobs, 5, callback
|
|
async.parallelLimit jobs, 5, callback
|
|
|
|
|
|
|
|
- # flush all outstanding changes
|
|
|
|
|
- flushAll: (limit, callback = (error, result) ->) ->
|
|
|
|
|
- RedisManager.getProjectIdsWithHistoryOps (error, project_ids) ->
|
|
|
|
|
- return callback(error) if error?
|
|
|
|
|
- logger.log {count: project_ids?.length, project_ids: project_ids}, "found projects"
|
|
|
|
|
- jobs = []
|
|
|
|
|
- project_ids = _.shuffle project_ids # randomise to avoid hitting same projects each time
|
|
|
|
|
- selectedProjects = if limit < 0 then project_ids else project_ids[0...limit]
|
|
|
|
|
- for project_id in selectedProjects
|
|
|
|
|
- do (project_id) ->
|
|
|
|
|
- jobs.push (cb) ->
|
|
|
|
|
- UpdatesManager.processUncompressedUpdatesForProject project_id, (err) ->
|
|
|
|
|
- return cb(null, {failed: err?, project_id: project_id})
|
|
|
|
|
- async.series jobs, (error, result) ->
|
|
|
|
|
- return callback(error) if error?
|
|
|
|
|
- failedProjects = (x.project_id for x in result when x.failed)
|
|
|
|
|
- succeededProjects = (x.project_id for x in result when not x.failed)
|
|
|
|
|
- callback(null, {failed: failedProjects, succeeded: succeededProjects, all: project_ids})
|
|
|
|
|
-
|
|
|
|
|
- getDanglingUpdates: (callback = (error, doc_ids) ->) ->
|
|
|
|
|
- RedisManager.getAllDocIdsWithHistoryOps (error, all_doc_ids) ->
|
|
|
|
|
- return callback(error) if error?
|
|
|
|
|
- RedisManager.getProjectIdsWithHistoryOps (error, all_project_ids) ->
|
|
|
|
|
- return callback(error) if error?
|
|
|
|
|
- # function to get doc_ids for each project
|
|
|
|
|
- task = (cb) -> async.concatSeries all_project_ids, RedisManager.getDocIdsWithHistoryOps, cb
|
|
|
|
|
- # find the dangling doc ids
|
|
|
|
|
- task (error, project_doc_ids) ->
|
|
|
|
|
- dangling_doc_ids = _.difference(all_doc_ids, project_doc_ids)
|
|
|
|
|
- logger.log {all_doc_ids: all_doc_ids, all_project_ids: all_project_ids, project_doc_ids: project_doc_ids, dangling_doc_ids: dangling_doc_ids}, "checking for dangling doc ids"
|
|
|
|
|
- callback(null, dangling_doc_ids)
|
|
|
|
|
-
|
|
|
|
|
getDocUpdates: (project_id, doc_id, options = {}, callback = (error, updates) ->) ->
|
|
getDocUpdates: (project_id, doc_id, options = {}, callback = (error, updates) ->) ->
|
|
|
UpdatesManager.processUncompressedUpdatesWithLock project_id, doc_id, (error) ->
|
|
UpdatesManager.processUncompressedUpdatesWithLock project_id, doc_id, (error) ->
|
|
|
return callback(error) if error?
|
|
return callback(error) if error?
|