transaction grouper

gistfile1.lua

local tslice = function(orig, first, last)
	local copy = {}
	while i=first, last do
		table.insert(copy, orig[i])
	end
	return copy
end

local function transactionize = function(self, transaction_blocks)
	local transaction_coroutines = {}
	for i,naked_callback in pairs(transaction_blocks) do
		table.insert(transaction_coroutines, ccreate(naked_callback))
	end
	local my_key = self:getKey()

	--transaction function
	local res, err = redis:transaction({cas=true, watch=self:getKey()}, function(redis)
		--WATCH ...
		while i<#transaction_coroutines do
			local transaction_callback = transaction_coroutines[i]
			assert(cresume(transaction_callback, redis, ...))
			if cstatus(transaction_callback)~='dead' then
				i = i + 1
			else
				table.remove(transaction_coroutines, i)
			end
		end
		
		redis:multi()
		
		local queued_commands_offset = {}
		while i<#transaction_coroutines do
			local transaction_callback = transaction_coroutines[i]
			local already_queued = redis:commands_queued()
			assert(cresume(transaction_callback))
			if cstatus(transaction_callback) ~= 'dead' then
				queued_commands_offset[transaction_callback]={ already_queued, redis:commands_queued() }
				i = i + 1
			else
				table.remove(transaction_coroutines, i)
			end
		end
	end)

	if not res then return nil, err end
	for i, transaction_callback in ipairs(transaction_coroutines) do
		cresume(transaction_callback, tslice(res, unpack(queued_commands_offset[transaction_callback])))
		--we no longer care about the coroutine's status. we're done.
	end
	return res
end

A transaction block for this function might look like this: https://gist.github.com/776659