<commit_before>_ = require("underscore")

module.exports = RedisSharelatex =
	createClient: (opts = {port: 6379, host: "localhost"})->
		if !opts.retry_max_delay?
			opts.retry_max_delay = 5000 # ms
		
		if opts.password?
			opts.auth_pass = opts.password
			delete opts.password
		if opts.endpoints?
			standardOpts = _.clone(opts)
			delete standardOpts.endpoints
			delete standardOpts.masterName
			client = require("redis-sentinel").createClient opts.endpoints, opts.masterName, standardOpts
		else
			standardOpts = _.clone(opts)
			delete standardOpts.port
			delete standardOpts.host
			client = require("redis").createClient opts.port, opts.host, standardOpts
		return client
		
	createRobustSubscriptionClient: (opts, heartbeatOpts = {}) ->
		sub = RedisSharelatex.createClient(opts)
		pub = RedisSharelatex.createClient(opts)
		
		heartbeatInterval = heartbeatOpts.heartbeat_interval or 1000 #ms
		reconnectAfter = heartbeatOpts.reconnect_after or 5000 #ms
		
		id = require("crypto").pseudoRandomBytes(16).toString("hex")
		heartbeatChannel = "heartbeat-#{id}"
		lastHeartbeat = Date.now()
		
		sub.subscribe heartbeatChannel, (error) ->
			if error?
				console.error "ERROR: failed to subscribe to #{heartbeatChannel} channel", error
		sub.on "message", (channel, message) ->
			if channel == heartbeatChannel
				lastHeartbeat = Date.now()
		
		reconnectIfInactive = () ->
			timeSinceLastHeartbeat = Date.now() - lastHeartbeat
			if timeSinceLastHeartbeat > reconnectAfter
				console.warn "No heartbeat for #{timeSinceLastHeartbeat}ms, reconnecting"
				sub.connection_gone("no heartbeat for #{timeSinceLastHeartbeat}ms")
				# Reset timer after triggering a reconnect to avoid potential cascading failure.
				lastHeartbeat = Date.now()
		
		setInterval () ->
			pub.publish heartbeatChannel, "PING"
			reconnectIfInactive()
		, heartbeatInterval
		
		return sub
			
		
<commit_msg>Handle reconnect logic better to (hopefully) avoid duplicate connections<commit_after>_ = require("underscore")

module.exports = RedisSharelatex =
	createClient: (opts = {port: 6379, host: "localhost"})->
		if !opts.retry_max_delay?
			opts.retry_max_delay = 5000 # ms
		
		if opts.password?
			opts.auth_pass = opts.password
			delete opts.password
		if opts.endpoints?
			standardOpts = _.clone(opts)
			delete standardOpts.endpoints
			delete standardOpts.masterName
			client = require("redis-sentinel").createClient opts.endpoints, opts.masterName, standardOpts
		else
			standardOpts = _.clone(opts)
			delete standardOpts.port
			delete standardOpts.host
			client = require("redis").createClient opts.port, opts.host, standardOpts
		return client
		
	createRobustSubscriptionClient: (opts, heartbeatOpts = {}) ->
		sub = RedisSharelatex.createClient(opts)
		pub = RedisSharelatex.createClient(opts)
		
		heartbeatInterval = heartbeatOpts.heartbeat_interval or 1000 #ms
		reconnectAfter = heartbeatOpts.reconnect_after or 5000 #ms
		
		id = require("crypto").pseudoRandomBytes(16).toString("hex")
		heartbeatChannel = "heartbeat-#{id}"
		lastHeartbeat = Date.now()
		
		sub.subscribe heartbeatChannel, (error) ->
			if error?
				console.error "ERROR: failed to subscribe to #{heartbeatChannel} channel", error
		sub.on "message", (channel, message) ->
			if channel == heartbeatChannel
				lastHeartbeat = Date.now()
		
		reconnectIfInactive = () ->
			timeSinceLastHeartbeat = Date.now() - lastHeartbeat
			if timeSinceLastHeartbeat > reconnectAfter
				console.warn "[#{new Date()}] No heartbeat for #{timeSinceLastHeartbeat}ms on #{heartbeatChannel}, reconnecting"
				# If the client realises it isn't connected then will be trying to
				# restablish the connection, so there's nothing for us to do. If
				# it still thinks it's connected, then disconnect it and start to reconnect.
				if sub.connected
					sub.end()
					# We ended the connection, but want to start it up again, so set
					# the internal closing variable:
					sub.closing = false
					# Trigger the reconnect:
					sub.connection_gone("no heartbeat for #{timeSinceLastHeartbeat}ms")
				# Reset timer after triggering a reconnect to avoid potential cascading failure.
				lastHeartbeat = Date.now()
		
		setInterval () ->
			pub.publish heartbeatChannel, "PING"
			reconnectIfInactive()
		, heartbeatInterval
		
		return sub
			
		

{'fraction_non_alphanumeric': 0.09655012734429266, 'fraction_numerical': 0.008335262792313036, 'mean_word_length': 4.112426035502959, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 17, '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': '1541450', 'n_tokens_mistral': 1534, 'n_tokens_neox': 1250, 'n_words': 336}