
---
-- Peque�o wrapper que maneja el caso de desconexion del redis
-- actualmente la libreria redis-lua no hace nada cuando se desconecta, lo cual es un poco molesto
-- con este wrapper se consigue automaticamente reconectar y reintentar el comando fallido
--
-- (c) 2011-2012 Datagnasco
--
--
local Redis = require "redis"
local socket = require "socket"
local json = require "cjson"
local hrtime = require "hrtime"

--how many commands we are going to wait before dump histogram
local HISTOGRAM_INTERVAL = 5000

--how much time in ms we consider that a command is delayed
local DELAY_THRESHOLD = 3000


--Removes functions at any level
local function SanitizeTable(t)
	local result = {}
	for k, v in pairs(t) do
		if type(v) == "table" then
			result[k] = SanitizeTable(v)
		else
			result[k] = tostring(v)
		end
	end
	return result
end


local function WrapRedis (ip, port, pwd)
	local logFn
	if type(ip) == "table" then
		local options = ip
		ip = options.ip
		port = options.port
		pwd = options.pwd
		logFn = options.logFn
	end
	
	local rediscore = Redis.connect (ip, port, pwd)
	local retry = true
	
	local commandsSinceLastLog = 0
	local timeDistribution = {}
	
	rediscore.error = function (err)
		if err ~= "closed" and err ~= "Socket is not connected" and not err:match("^connection error") then
			rediscore.network.socket:close()
			retry = false
		end
		
		local errFunction = rediscore.error
		rediscore = Redis.connect (ip, port, pwd)
		rediscore.error = errFunction
	
		error(err)
	end
	
	local redis = {}
	setmetatable (redis, {__index = function (t, k)
		local f = function (self, ...)
			local params = {...}
			local ok, err, result
			local retries = 5
			
			retry = true
			while not ok and retries > 0 do
				ok, err = pcall (function ()
					
					if logFn then
						local startTime = hrtime()
						result = {rediscore[k](rediscore, unpack (params))}
						local endTime = hrtime(startTime)
						
						local duration = endTime:ms()
						
						if duration >= DELAY_THRESHOLD then
							logFn("WrapRedis (%s): delay of %s ms in command %s with parameters %s", tostring(redis), duration, k, json.encode(SanitizeTable(params)))
						end
						
						timeDistribution[k] = timeDistribution[k] or {}
						duration = duration.."ms"
						timeDistribution[k][duration] = (timeDistribution[k][duration] or 0) + 1
						
						commandsSinceLastLog = commandsSinceLastLog + 1
						if commandsSinceLastLog >= HISTOGRAM_INTERVAL then
							commandsSinceLastLog = 0
							logFn("WrapRedis (%s): Time distribution of last %d commands: %s", tostring(redis), HISTOGRAM_INTERVAL, json.encode(timeDistribution))
							timeDistribution = {}
						end
					else
						result = {rediscore[k](rediscore, unpack (params))}
					end
					
				end)
				
				if ok then
					return unpack (result)
				elseif not retry then
					--we only retry on connection errors
					if logFn then
						logFn("WrapRedis (%s): error in command %s with parameters %s, because of %s", tostring(redis), k, json.encode(SanitizeTable(params)), err)
					end
					break
				end
				socket.sleep(.5)
				retries = retries - 1
				
				if logFn then
					logFn("WrapRedis (%s): retrying command %s with parameters %s, because of err %s", tostring(redis), k, json.encode(SanitizeTable(params)), err)
				end
				
			end
			
			if not ok then
				error (err)
			end
		end
		rawset(t, k, f)
		return f
	end})
	
	return redis
end

return WrapRedis
