<commit_before>Settings = require('settings-sharelatex')
logger = require('logger-sharelatex')
Keys = require('./UpdateKeys')
redis = require("redis-sharelatex")

UpdateManager = require('./UpdateManager')
Metrics = require('./Metrics')
RateLimitManager = require('./RateLimitManager')

module.exports = DispatchManager =
	createDispatcher: (RateLimiter) ->
		client = redis.createClient(Settings.redis.realtime)
		worker = {
			client: client
			_waitForUpdateThenDispatchWorker: (callback = (error) ->) ->
				timer = new Metrics.Timer "worker.waiting"
				worker.client.blpop "pending-updates-list", 0, (error, result) ->
					timer.done()
					return callback(error) if error?
					return callback() if !result?
					[list_name, doc_key] = result
					[project_id, doc_id] = Keys.splitProjectIdAndDocId(doc_key)
					# Dispatch this in the background
					backgroundTask = (cb) ->
						UpdateManager.processOutstandingUpdatesWithLock project_id, doc_id, (error) ->
							logger.error err: error, project_id: project_id, doc_id: doc_id, "error processing update" if error?
							cb()
					RateLimiter.run backgroundTask, callback
						
			run: () ->
				return if Settings.shuttingDown
				worker._waitForUpdateThenDispatchWorker (error) =>
					if error?
						logger.error err: error, "Error in worker process"
						throw error
					else
						worker.run()
		}
		
		return worker
		
	createAndStartDispatchers: (number) ->
		RateLimiter = new RateLimitManager(number)<commit_msg>stop OpRangeNotAvailable errors going to sentry<commit_after>Settings = require('settings-sharelatex')
logger = require('logger-sharelatex')
Keys = require('./UpdateKeys')
redis = require("redis-sharelatex")
Errors = require("./Errors")

UpdateManager = require('./UpdateManager')
Metrics = require('./Metrics')
RateLimitManager = require('./RateLimitManager')

module.exports = DispatchManager =
	createDispatcher: (RateLimiter) ->
		client = redis.createClient(Settings.redis.realtime)
		worker = {
			client: client
			_waitForUpdateThenDispatchWorker: (callback = (error) ->) ->
				timer = new Metrics.Timer "worker.waiting"
				worker.client.blpop "pending-updates-list", 0, (error, result) ->
					timer.done()
					return callback(error) if error?
					return callback() if !result?
					[list_name, doc_key] = result
					[project_id, doc_id] = Keys.splitProjectIdAndDocId(doc_key)
					# Dispatch this in the background
					backgroundTask = (cb) ->
						UpdateManager.processOutstandingUpdatesWithLock project_id, doc_id, (error) ->
							# log everything except OpRangeNotAvailable errors, these are normal
							if error? and not (error instanceof Errors.OpRangeNotAvailableError)
								logger.error err: error, project_id: project_id, doc_id: doc_id, "error processing update"
							cb()
					RateLimiter.run backgroundTask, callback
						
			run: () ->
				return if Settings.shuttingDown
				worker._waitForUpdateThenDispatchWorker (error) =>
					if error?
						logger.error err: error, "Error in worker process"
						throw error
					else
						worker.run()
		}
		
		return worker
		
	createAndStartDispatchers: (number) ->
		RateLimiter = new RateLimitManager(number)
{'fraction_non_alphanumeric': 0.11118090452261306, 'fraction_numerical': 0.000628140703517588, 'mean_word_length': 3.8184568835098336, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3846807', 'n_tokens_mistral': 1150, 'n_tokens_neox': 917, 'n_words': 225}