local ss = analiz.libs.serv
local log = analiz.libs.logserv
log.init()
SetSecLevel()
ss.s("abort_flag", "1")

local pwd = ss.CreatePwdServ()
local tick = CreateTimeServ()

local ll = analiz.libs.logs
local finn_serv = analiz.mix.finnova.serv
local xml_io = CreateXmlIO()
local mem = ss.CreateMemItem(8000)

local orgclient_default = pwd:get("PLIB").org

local org_engine = CreateOrgClientEngine()
if OrgClientEngine_errors() ~= "" then
    ss.l(" failed load orgclients:" .. OrgClientEngine_errors())
    ss.l("sleep&restart finas")
    ss.e("SLEEP 3 min")
    ss.a("1")
end

function go_pskill()
ss.l(debug.traceback())
    ss.l("go_pskill")
    ss.e("calc_param pskill_exe $(_ANALIZ_BIN)\\pskill.exe")
    ss.e("GET_SHORT_PATH pskill_exe $(pskill_exe)")
    ss.e("GET_CURRENT_PROCESS_ID txm_pid")
    ss.e("echo execute $(pskill_exe) $(txm_pid)")
    ss.e("execute $(pskill_exe) $(txm_pid)")
    ss.e("_abort 1")
end

local fgenerator = CreateGenerateThrow()
fgenerator:init("endpoint")

list_ver_id = GetCurrentListID({pwd_plib = "PLIB"})
ss.l("list_ver_id:" .. list_ver_id)

-- local clog =  ll.CreateCompactLog2({limit=80,limit_dt=600,separator=""})

local uniq_id = ss.b("$(service_family):$(UNIQUE_IDENTIFIER)")
rstatus = CreateRecordStatusObject({name = uniq_id, lazy = true})

internal_state = nil

function AddInternalState(msg)
    -- ss.l("InternalState:"..msg)
    internal_state = msg
end
-- -------------------------------------------------------------------------

-- -------------------------------------------------------------------------

-- -------------------------------------------------------------------------

-- -----------------------------------------------------------
-- -----------------------------------------------------------
-- -----------------------------------------------------------
-- -----------------------------------------------------------
local function obj_get_(queue_msg)
    local msg = queue_msg;
    if msg:isnil() then return nil, nil, nil end
    --   local obj = msg:getObj();
    local msgid = nil
    ss.l("serv_oracle.lua:msg:getCorrelation")
    local corrid = msg:getCorrelation()
    local prior = msg:getPrior()
    ss.l("serv_oracle.lua:msg:getCorrelation:" .. ss.def(corrid, "<nil>") ..
             ".prior:" .. ss.def(prior, "<nil>"))

    msgid = msg:getId()

    local all_prop = msg:property()
    -- ss.l("________________________________________________")
    --   ss.tprint(all_prop)
    -- ss.a(1)
    local ret = {}

    ret.bid = ss.def(all_prop.cache.BID, "")
    ret.lastname = ss.def(all_prop.cache.LASTNAME, "")
    ret.firstname = ss.def(all_prop.cache.FIRSTNAME, "")
    ret.countries_filter = ss.def(all_prop.cache.COUNTRIES_FILTER, "<nil>")
    ret.dob = ss.def(all_prop.cache.DOB, "<nil>")
    ret.start_dt = analiz:tonumber(ss.def(all_prop.cache.START_DT, 0))
    ret.list_ver_id = ss.def(all_prop.cache.LIST_VER_ID, "")
    ret.add_name1 = ss.def(all_prop.cache.ADD_NAME1, "")
    ret.add_name2 = ss.def(all_prop.cache.ADD_NAME2, "")
    ret.address = ss.def(all_prop.cache.ADDRESS, "Address:nul:2")
    ret.remmitance = ss.def(all_prop.cache.REMITTANCE, "REMITTANCE:nul:2")
    ret.reserve2 = ss.def(all_prop.cache.reserve2, "reserve2:nul:2")
    ret.USERNAME = ss.def(all_prop.cache.USERNAME, "username:nul:2") -- todo
    ret.raw = all_prop
    ss.l("debug point:1729")
    -- LL   ss.tprint(ret)
    --    ret.IDNAME = ret.bid
    --    ret.DOB = ret.dob
    return ret, msgid, corrid, prior
end

-- ----------

-- ----------main loop
function main_(ctx)
    if ctx.engine.ttype ~= "msbroker" then
        ss.e("sleep sec 10")
        if tick.to_sec(tick.now() - ctx.time_) > 10 * 60 then
            ss.l("main.skip:wait_count:" .. ctx.wait_count)
            ctx.time_ = tick.now()
        end
        return
    end
    last_corrid = nil
    last_bid = nil
    last_cache = nil
    deadline:DeadLineLazy("wait", deadline.time.WAIT)
    --   local msg,msgid,corrid,prior = deq:get();
    rstatus:commit({lazy = true,fast=1})
    local queue_msg = ctx.reader:read()

    local msg, msgid, corrid, prior = obj_get_(queue_msg)

    if msg == nil then
        if tick.to_sec(tick.now() - ctx.time_) > 10 * 60 then
            ss.l("wait_count:" .. ctx.wait_count)
            ctx.time_ = tick.now()
        end
        ctx.wait_count = ctx.wait_count + 1
        rstatus:commit({lazy = true})
        return
    end
    ctx.time_ = tick.now()
    ss.l("message.prior:" .. prior)
    AddInternalState("load msg")
    local val = msg

    last_corrid = corrid
    last_bid = val.bid
    last_cache = val.raw.cache
    rstatus:add(val.bid, corrid, "endpoint.start", "")
    ctx.engine:commit()

    deadline:DeadLine("processing", deadline.time.PROCESSING)
    ctx.endpoint:DoTask(ctx, queue_msg)

end

ss.l("Start")
-- -------------------------init

local ctx = {}
ss.l("queue.engine init...")

ctx.engine = CreateQueueEngine()
ctx.engine:init()
ctx.time_ = tick.now()

ss.l("ReaderCreate..")

local input_queue_name = global_oracle_setting.finnova.out
ss.l("create reader:" .. input_queue_name)
ctx.reader = ctx.engine:createReader({queue = input_queue_name})

ctx.endpoint = CreateEndPointEngine()
ctx.endpoint:init()
ctx.wait_count = 0

ctx.writer = nil -- todo
ss.l("f_ailed.writer setting..")
global_failedQueue = ctx.writer
global_engine = ctx.engine
init_failed_info()

-- ----------main loop
while true do
    if analiz:IsStop() then
        ss.l("user break")
        break
    end
    main_(ctx)

    local err = ctx.engine:serv():GetLastError()
    if err ~= "" then
        rstatus:close()
        ss.l("queue.error:" .. err)
        ss.l("restart after cleep")
        ss.e("sleep sec 15")
        ss.l("abort")
        go_pskill()
        ss.a("1")
    end
end

rstatus:close()
rstatus:close()
ctx.engine:destroy()

ss.s("abort_flag", "")


