pub/sub bus lua

bus.lua

-- luacheck:ignore 
max = {
	req = 0,
	seq = 0,
}

local function next_req()
	max.req = max.req + 1
	return max.req
end

local function next_seq()
	max.seq = max.seq + 1
	return max.seq
end

fiber = require 'fiber'
clock = require 'clock'

box.cfg{
	listen = 3301,
}

box.schema.create_space('bus', {
	format = {
		{ name = "req",    type = "unsigned" },
		{ name = "seq",    type = "unsigned" },
		{ name = "source", type = "string" },
		{ name = "ctime",  type = "unsigned" },
		{ name = "atime",  type = "unsigned" },
	},
	if_not_exists = true,
})

box.space.bus:create_index('primary', {
	parts = { "req", "seq" },
	type = "tree",
	if_not_exists = true,
})

do
	-- initialize:
	local last = box.space.bus.index.primary
		:pairs({}, { iterator = box.index.REQ })
		:take_n(1)
		:totable()[1]

	if last then
		max = {
			req = last.req,
			seq = last.seq,
		}
	end
end

local reqseq_cond = fiber.cond()

function insert(ctime)
	box.space.bus:insert{
		next_req(),
		max.seq,
		"1",
		ctime,
		clock.realtime64(),
	}

	reqseq_cond:broadcast()
end

function update(ctime)
	box.space.bus:insert{
		max.req,
		next_seq(),
		"1",
		ctime,
		clock.realtime64(),
	}

	reqseq_cond:broadcast()
end

function get(req, seq)
	if req == max.req and seq == max.seq then
		if not reqseq_cond:wait(0.001) then
			-- timeout exceed
			return
		end
	end
	return box.space.bus.index.primary
		:pairs({ req, seq }, { iterator = box.index.GE })
		:grep(function(t) return t.req > req or t.seq > seq end)
		:take_n(100)
		:totable()
end

require'console'.start()
os.exit()