quik-history-provider/qhp.lua

local alien = require("alien")
json = require "JSON"
date = require "date"
require "string"
struct = require "alien.struct"

w32sleep = alien.kernel32.Sleep
w32sleep:types { ret = "void", abi = "stdcall", "int" }


zmq_pollitem = alien.defstruct {
	{"socket", "pointer"},
	{"fd", "int"},
	{"events", "short"},
	{"revents", "short"}
}

zmq_ctx_new = alien.libzmq.zmq_ctx_new
zmq_ctx_new:types { ret = "pointer", abi="stdcall" }

zmq_ctx_term = alien.libzmq.zmq_ctx_term
zmq_ctx_term:types { ret = "int", abi="stdcall", "pointer" }

zmq_bind = alien.libzmq.zmq_bind
zmq_bind:types { ret = "int", abi="stdcall", "pointer", "string" }

zmq_close = alien.libzmq.zmq_close
zmq_close:types { ret = "int", abi="stdcall", "pointer" }

zmq_disconnect = alien.libzmq.zmq_disconnect
zmq_disconnect:types { ret = "int", abi="stdcall", "pointer", "string" }

zmq_setsockopt = alien.libzmq.zmq_setsockopt
zmq_setsockopt:types { ret = "int", abi="stdcall", "pointer", "int", "pointer", "int" }

zmq_errno = alien.libzmq.zmq_errno
zmq_errno:types { ret = "int", abi="stdcall" }

zmq_msg_close = alien.libzmq.zmq_msg_close
zmq_msg_close:types { ret = "int", abi="stdcall", "pointer" }

zmq_msg_data = alien.libzmq.zmq_msg_data
zmq_msg_data:types { ret = "pointer", abi="stdcall", "pointer" }

zmq_msg_get = alien.libzmq.zmq_msg_get
zmq_msg_get:types { ret = "int", abi="stdcall", "pointer", "int" }

zmq_msg_init_size = alien.libzmq.zmq_msg_init_size
zmq_msg_init_size:types { ret = "int", abi="stdcall", "pointer", "int" }

zmq_msg_init = alien.libzmq.zmq_msg_init
zmq_msg_init:types { ret = "int", abi="stdcall", "pointer" }

zmq_msg_more = alien.libzmq.zmq_msg_more
zmq_msg_more:types { ret = "int", abi="stdcall", "pointer" }

zmq_msg_recv = alien.libzmq.zmq_msg_recv
zmq_msg_recv:types { ret = "int", abi="stdcall", "pointer", "pointer", "int" }

zmq_msg_send = alien.libzmq.zmq_msg_send
zmq_msg_send:types { ret = "int", abi="stdcall", "pointer", "pointer", "int" }

zmq_msg_set = alien.libzmq.zmq_msg_set
zmq_msg_set:types { ret = "int", abi="stdcall", "pointer", "int", "int" }

zmq_msg_size = alien.libzmq.zmq_msg_size
zmq_msg_size:types { ret = "int", abi="stdcall", "pointer" }

zmq_poll = alien.libzmq.zmq_poll
zmq_poll:types { ret = "int", abi="stdcall", "pointer", "int", "int" }

zmq_recv = alien.libzmq.zmq_recv
zmq_recv:types { ret = "int", abi="stdcall", "pointer", "pointer", "int", "int"}

zmq_send = alien.libzmq.zmq_send
zmq_send:types { ret = "int", abi="stdcall", "pointer", "pointer", "int", "int"}

zmq_socket = alien.libzmq.zmq_socket
zmq_socket:types { ret = "pointer", abi="stdcall", "pointer", "int"}

zmq_version = alien.libzmq.zmq_version
zmq_version:types { ret = "void", abi="stdcall", "ref int", "ref int", "ref int"}

zmq_unbind = alien.libzmq.zmq_unbind
zmq_unbind:types { ret = "int", abi="stdcall", "pointer", "string"}

function zmq_msg_new()
	local buf = alien.buffer(64)
	return buf
end

function zmq_msg_as_string(msg)
	return alien.tostring(zmq_msg_data(msg), zmq_msg_size(msg))
end

ZMQ_PAIR = 0
ZMQ_PUB = 1
ZMQ_SUB = 2
ZMQ_REQ = 3
ZMQ_REP = 4
ZMQ_DEALER = 5
ZMQ_ROUTER = 6
ZMQ_PULL = 7
ZMQ_PUSH = 8
ZMQ_XPUB = 9
ZMQ_XSUB = 10
ZMQ_STREAM = 11

ZMQ_POLLIN = 1
ZMQ_POLLOUT = 2
ZMQ_POLLERR = 4

ZMQ_SNDMORE = 2
ZMQ_LINGER = 17

running = true

function to_unixtime(year, month, day)
	return date.diff(date(year, month, day), date(1970, 1, 1)):spanseconds()
end

function to_unixtime_fine(year, month, day, hour, min, sec)
	return date.diff(date(year, month, day, hour, min, sec), date(1970, 1, 1)):spanseconds()
end

function serialize_ts(tm)
	return to_unixtime_fine(tm.year, tm.month, tm.day, tm.hour, tm.minute, tm.second)
end

function serialize_bar(ds, i, lot_size)
	return struct.pack("<I8ddddI8", serialize_ts(ds:T(i)), ds:O(i), ds:H(i), ds:L(i), ds:C(i), ds:V(i) * lot_size)
end

function get_data(class, code, t_start, t_end, timeframe)
	local ds, Error = CreateDataSource(class, code, timeframe)
	local timeout = 30
	local lot_size = tonumber(getParamEx(class, code, "LOTSIZE").param_value)
	while (Error == "" or Error == nil) and ds:Size() == 0 do
		sleep(100)
		timeout = timeout - 1
		if timeout <= 0 then
			return nil, "Timeout"
		end
	end
	if Error ~= "" and Error ~= nil then return nil, Error end

	local Size = ds:Size()
	local result = {}
	local cnt = 1
	for i = 1, Size do
		local t = ds:T(i)
		local ts = to_unixtime(t.year, t.month, t.day)
		if ts >= t_start and ts < t_end then
			result[cnt] = serialize_bar(ds, i, lot_size)
			cnt = cnt + 1
		end
	end
	ds:Close()
	--return nil, "foo"
	s = table.sconcat(result)
	return s, nil
end

function to_timeframe(tf)
	if tf == "M1" then return INTERVAL_M1
	elseif tf == "M5" then return INTERVAL_M5
	elseif tf == "M15" then return INTERVAL_M15
	elseif tf == "M30" then return INTERVAL_M30
	elseif tf == "H1" then return INTERVAL_H1
	elseif tf == "H2" then return INTERVAL_H2
	elseif tf == "H4" then return INTERVAL_H4
	elseif tf == "D" then return INTERVAL_D1
	elseif tf == "W" then return INTERVAL_W1
	elseif tf == "MN" then return INTERVAL_MN1
	end
end

function zmq_msg_from_string(str)
	local msg = zmq_msg_new()
	zmq_msg_init_size(msg, str:len())
	buf = zmq_msg_data(msg)
	bsource = alien.buffer(str)
	alien.memcpy(buf, bsource, str:len())
	return msg
end

function zmq_msg_empty()
	local msg = zmq_msg_new()
	zmq_msg_init(msg)
	return msg
end

function handle_message(msg)
	local rq = json:decode(zmq_msg_as_string(msg))
	local class, code = string.match(rq.ticker, "(%w+)#(%w+)")
	local fyear, fmon, fday = string.match(rq.from, "(%d+)-(%d+)-(%d+)")
	local tyear, tmon, tday = string.match(rq.to, "(%d+)-(%d+)-(%d+)")
	local timeframe = to_timeframe(rq.timeframe)
	if timeframe == nil then
		return zmq_msg_from_string("ERROR: invalid timeframe: " .. tostring(timeframe)), zmq_msg_empty()
	end

	local data, err = get_data(class, code, to_unixtime(fyear, fmon, fday), to_unixtime(tyear, tmon, tday), timeframe)
	if data == nil then
		return zmq_msg_from_string("ERROR: can't get data: " .. err .. "(" .. class .. "/" .. code .. ")" ), zmq_msg_empty()
	end

	return zmq_msg_from_string("OK"), zmq_msg_from_string(data)
end

ctx = nil
sock = nil

function OnStop()
	if sock then
		zmq_close(sock)
		sock = nil
	end
	if ctx then
		zmq_ctx_term(ctx)
		ctx = nil
	end
end

function main()
	local status, err = pcall(pmain)
	if not state then
		message("Error: " .. err)
	end
	OnStop()
end

function pmain()
	ctx = zmq_ctx_new()
	sock = zmq_socket(ctx, ZMQ_REP)
	local linger = alien.array("int", {0})
	zmq_setsockopt(sock, ZMQ_LINGER, linger.buffer, 4)
	local rc = zmq_bind(sock, "tcp://*:5521")
	if rc ~= 0 then
		message("Bind error: " .. tostring(rc))
		return
	end

	while running do
		local pi = zmq_pollitem:new()
		pi.socket = sock
		pi.fd = 0
		pi.events = ZMQ_POLLIN
		pi.revents = 0

		rc = zmq_poll(pi(), 1, 60000) 
		if rc > 0 and pi.revents == ZMQ_POLLIN then
			local msg = zmq_msg_new()
			zmq_msg_init(msg)
			rc = zmq_msg_recv(msg, sock, 0) 
			if rc == -1 then
				running = false
				message("rc: " .. tostring(rc) .. "; " .. tostring(zmq_errno()))
			else
				local outmsg_header, outmsg_data = handle_message(msg)
				zmq_msg_send(outmsg_header, sock, ZMQ_SNDMORE)
				zmq_msg_send(outmsg_data, sock, 0)
			end
			zmq_msg_close(msg)
		end
	end
end