local ss = analiz.libs.serv

local base64 = require("ee5_base64")

local pwd = ss.CreatePwdServ()
local base64 = require("ee5_base64")
local perf = CreatePerfomaceObj()
local log = analiz.libs.logserv
log.init()
SetSecLevel()

local finn_serv = analiz.mix.finnova.serv

-- patch_validator:ignore:ru

function go_pskill()
    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 function IsUseNewVariant(fa_params)
    ss.l("IsUseNewVariant.start")
    local val = ss.def(fa_params.SCRIPT_VERSION, "ver2")
    local is_support_FAU_Proxy1_5 = val > "ver2"

    local isQueueDelta = fa_params.KUNDESTAMMDELTA ==
                             "queue:FAU_kundenstammDelta5"

    return is_support_FAU_Proxy1_5 and isQueueDelta
end

local function IsUseNewVariantFactiva(fa_params)
    ss.l("IsUseNewVariantFactiva.start")
    local val = ss.def(fa_params.SCRIPT_VERSION, "ver2")
    local is_support_FAU_Proxy1_5 = val > "ver2"

    local is_ver32 = fa_params.SCRIPT_VERSION_FULL_0032 == "yes"

    return is_ver32
end

-- move to failedServ.lua
-- last_corrid=nil
-- last_bid = nil
-- last_cache=nil

internal_state = nil

local queues = {}
-- сразу после записи во внутреннюю очередь отсыласть commit или нет
-- не используется local is_immediately_execute_commit=true

-- игнорировать транзакции release
-- читает но не куда дальше не отсылает
-- чисто в отладочных целях
-- нормальное состояние - false
local disable_release_transaction = false

local unique_identifier = ss.g("UNIQUE_IDENTIFIER")
local servicename = "proxy_" .. unique_identifier
local profilename_default = "proxy_std"
if unique_identifier:find("batch") == 1 then profilename_default = "proxy_batch" end
local profilename_for_proxy = finn_serv.GetNodeProfileName(servicename,
                                                           "Profile",
                                                           profilename_default,
                                                           "1")

local org_engine = CreateOrgClientEngine()

local fa_params = Get_FA_SCHEMA_PARAMS()
local is_kundenstamDelta = fa_params.KUNDESTAMMDELTA ==
                               "queue:FAU_kundenstammDelta5"
local is_support_FAU_Proxy1_5 = ss.def(fa_params.SCRIPT_VERSION, "ver2") >
                                    "ver2"
local isNewVariant = IsUseNewVariant(fa_params)
local isNewVariantFactiva = IsUseNewVariantFactiva(fa_params)

local proxy_profile   = org_engine:profile(profilename_for_proxy, {advanced_ = {}}).advanced_
local system_profile_ = org_engine:profile("SYSTEM",{advanced_={}}).advanced_

ss.l("fa_params:")
ss.tprint(fa_params)
ss.l("proxy_profile:")
ss.tprint(proxy_profile)

if is_kundenstamDelta then ss.l("is_kundenstamDelta:true") end
if is_support_FAU_Proxy1_5 then ss.l("11.is_support_FAU_Proxy1_5:true") end
if isNewVariant then ss.l("isNewVariant:true") end
if isNewVariantFactiva then ss.l("isNewVariantFactiva(ver3.22):true") end

local input_queue_name = global_oracle_setting_local.finnova.from
local script_ver = ss.def(fa_params.SCRIPT_VERSION, "ver2")
if ss.def(proxy_profile.INPUT_QUEUE, ""):find("custom:") == 1 and script_ver >
    "ver2" then
    local l = proxy_profile.INPUT_QUEUE
    input_queue_name = finn_serv.GetFullQName(l:sub(8, l:len()))
end
ss.l("queues.input.name:" .. input_queue_name)

if proxy_write_xml_config~=nil then 
  proxy_write_xml_config({org_engine=org_engine})
end

local qselector = CreateQueueSelector({org_engine_ = org_engine})
if OrgClientEngine_errors() ~= "" then
    ss.l(" failed load orgclients:" .. OrgClientEngine_errors())
    ss.l("sleep&restart proxy")
    ss.e("SLEEP 3 min")
    ss.a("1")
end

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

local tick = CreateTimeServ()

local dtserv = CreateDateTime()

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

seclevel = CreateSecLevel("proxy")

-- сколько отсылать в очередь сообщений о смене версий
local restart_message_count = 10
list_ver_id = GetCurrentListID({pwd_plib = "PLIB"})
ss.l("start.list_ver_id:" .. list_ver_id)

local wait_count = 0
local no_wait_count = 0
local count_ok = 0
local count_bid = 0
local bid_pref = ss.b("$(GUID)_")
local xml_io = CreateXmlIO()

local isdirty_ = false

local is_restart_new_version = false

local mem = ss.CreateMemItem(10 * 1024)

local parse_serv = CreateParseServ()
parse_serv:init({
    deadline_ = deadline,
    cache_dir = ss.g("LOG_FORMAT_dir") .. "\\cache"
})
--[[
local cache_dir = ss.g("LOG_FORMAT_dir").."\\cache"
ss.e("FORCEDIRECTORIES "..cache_dir)


-- -------------------------------------------------------------------------
deadline:DeadLine("LoadPrenames",deadline.time.LOAD_PRENAMES)

local prenames      = LoadPrenames({cacheio=cache_dir.."\\prename."..list_ver_id})
local family_parser = CreateEntitiesNamesParser_ver3(prenames)
--]]
local isDebug = false
if isDebug then
    ss.e("i $(lua_analiz_libs)\\init_socket.rc")
    ss.e("i $(lua_analiz_libs)\\moddebug.rc")
    debug = require("mobdebug")
    ss.l("start debug..")
    debug.start() -- same as start("localhost", 8172)
    ss.l("start debug..ok")
end

global_failed_counters = {null_roolback = 0, null_roolback_limit = 3}

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

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

local function qflush(op)
    for k, v in pairs(queues) do if v:isDirty() then v:flush(op) end end
end

-- ---------------------------------------------------
local function num_fix(n)
    local s = "" .. n
    local l = s:len()
    while l < 7 do
        s = "0" .. s
        l = l + 1
    end
    return s
end

-- ---------------------------------------------------
local function kundenstamm_transform_(val1, bank_id)
    AddInternalState("kundenstamm_transform.Extract.Remittance")
    local parse_op = {
        type_ = "kundenstamm",
        bank_id_ = bank_id,
        org_engine_ = org_engine
    }

    val1.lastname = parse_serv:ExtractNamesFromLastName(val1, parse_op)
    val1.remittance = parse_serv:ExtractNamesFromRemittance(val1, parse_op)

    AddInternalState("kundenstamm_transform.Extract.Addrline")
    val1.address = parse_serv:ExtractNamesFromAddrline(val1, parse_op)

    val1.lastname = parse_serv:ExtractNamesFromLastNameFinal(val1.lastname,
                                                             val1, parse_op)

    local tmp = {}

    AddInternalState("kundenstamm_transform.Extract.2")
    local kk = global_oracle_setting.fields.prop_fields_Kundenstamm
    for i = 1, #kk do
        k = kk[i]
        tmp[k] = val1[k]
    end

    AddInternalState("kundenstamm_transform.Extract.3")
    tmp.remittance = val1.remittance
    tmp.address = val1.address
    tmp.remittance_raw = val1.remittance_raw
    tmp.address_raw = val1.address_raw
    tmp.FIRST_NAME = val1.F_CUSTOMER_NAME
    tmp.LAST_NAME = val1.F_CUSTOMER_SURNAME
    tmp.ADD_NAME1 = val1.F_CUSTOMER_OTHER_NAMES
    tmp.BIRTHDAY = val1.F_BIRTHDAY
    tmp.BIRTHYEAR = val1.F_BIRTHYEAR
    tmp.GENDER = val1.F_GENDER
    tmp.NATIONALITY = val1.F_NATIONALITY
    tmp.ADDRESS_STREET = val1.F_ADDRESS_STREET
    tmp.ADDRESS_ZIP = val1.F_ADDRESS_ZIP
    tmp.ADDRESS_CITY = val1.F_ADDRESS_CITY
    tmp.ADDRESS_COUNTRY = val1.F_ADDRESS_COUNTRY
    AddInternalState("kundenstamm_transform.Extract.end")

    return tmp

end
-- ---------------------------------------------------
local function std_transform_(val1, bank_id,record)
    local function conv_load_(key, val) return seclevel:get("load", key, val) end

    AddInternalState("parse xml")
    ss.l("load.rec:" .. ss.toJson(val1, conv_load_))
    ss.l("parse:")

    val1.remittance_raw = val1.remittance

    val1.address_raw = val1.address
    local address_merger = finasServ.CreateMerger(
                               {
            name = "address",
            separator = "\n\n",
            ignore = {[""] = 1},
            uniq = true,
            trim = true
        })
    local remittance_merger = finasServ.CreateMerger(
                                  {
            name = "remittance",
            separator = "\n\n",
            ignore = {[""] = 1}
        })

    address_merger:add(val1.address)
    remittance_merger:add(val1.remittance)

    ss.l("val1.remittance_raw:" ..
             seclevel:get("prepare", "remittance_raw",
                          ss.def(val1.remittance_raw, "nil")))
    ss.l("val1.address_raw:" ..
             seclevel:get("prepare", "address_raw",
                          ss.def(val1.address_raw, "nil")))
    local tmp = {}
    AddInternalState("parse xml.Remittance")

    local parse_op = {
        type_ = "checkpayment",
        bank_id_ = bank_id,
        org_engine_ = org_engine
    }

    val1.lastname = parse_serv:ExtractNamesFromLastName(val1, parse_op)

    val1.remittance = parse_serv:ExtractNamesFromRemittance(val1, parse_op)
    ss.l("extract remittance:" .. ss.def(val1.remittance, "nil"))
    AddInternalState("parse xml.Addrline")
    val1.address = parse_serv:ExtractNamesFromAddrline(val1, parse_op)
    AddInternalState("parse xml.3")

    val1.lastname = parse_serv:ExtractNamesFromLastNameFinal(val1.lastname,
                                                             val1, parse_op)

    tmp.remittance = val1.remittance
    tmp.address = val1.address
    tmp.remittance_raw = val1.remittance_raw
    remittance_merger:add(val1.remittance2)

    val1.remittance_raw = remittance_merger:get()

   -- fastt fix for wuro
   do 
    local direct =  finn_serv.GetDirectType(record)-- incoming/outgoing
    local use_all = false -- not support: ss.def(curr_profile.ADDRESS_PARSING_MODE,"DIRECT"):upper()=="BOTH_DIRECTIONS"

    if direct=="incoming" or use_all==true then 
      if val1.dbtr~=nil then       address_merger:add(val1.dbtr.address) end
--1588         if val1.debtor_agent_addrline~=nil then   address_merger:add(val1.debtor_agent_addrline) end --in
         if val1.debitor_addrline~=nil then        address_merger:add(val1.debitor_addrline) end
  
      end
      if direct=="outgoing" or use_all==true then 
      if val1.cdtr~=nil then       address_merger:add(val1.cdtr.address) end
--1588      if val1.creditor_agent_addrline~=nil then       address_merger:add(val1.creditor_agent_addrline) end
      if val1.creditor_addrline~=nil then       address_merger:add(val1.creditor_addrline) end
      if val1.ultimate_agent_addrline ~= nil then         address_merger:add(val1.ultimate_agent_addrline)   end --UltmtCdtr\\PstlAdr
      end







     do
       local option_value = ss.def(system_profile_.ENABLED_FPATCH1688,"YES"):upper() --nornal.state = YES or  not define
        if option_value~="NO" then  --Any value is acceptable except NO.                           
          ss.l("name52D and name57D skip to address_merger.ENABLED_FPATCH1688:"..option_value)
       else
          ss.l("name52D and name57D add to address_merger.ENABLED_FPATCH1688:"..option_value)
          if val1.name52D~=nil then                 address_merger:add(val1.name52D.address) end --???
          if val1.name57D~=nil then                 address_merger:add(val1.name57D.address) end --???
       end
     end



    if val1.name53D ~= nil then address_merger:add(val1.name53D.address) end
    if val1.name54D ~= nil then address_merger:add(val1.name54D.address) end
    if val1.name55D ~= nil then address_merger:add(val1.name55D.address) end
    if val1.name56D ~= nil then address_merger:add(val1.name56D.address) end
   end --fast fix

    val1.address_raw = address_merger:get()

    ss.l("tmp.remittance_raw:" ..
             seclevel:get("prepare", "remittance_raw",
                          ss.def(tmp.remittance_raw, "nil")))
    ss.l("tmp.address_raw:" ..
             seclevel:get("prepare", "address_raw",
                          ss.def(tmp.address_raw, "nil")))

    log.l("data2:", log.level.INFO)
    log.tprint(tmp, log.level.INFO, {transform_function = conv_load_})
    log.l("data2:end", log.level.INFO)
    return tmp
end
-- ---------------------------------------------------
local function add(a, b, sep)
    if a == nil or a == "" then return b end
    return a .. sep .. b
end
-- ---------------------------------------------------
local function IsKundenstamm(record) return finn_serv.IsKundenstamm(record) end

-- ---------------------------------------------------
local function IsBot(record)
    local is_bot_std = finn_serv.IsBot(record)
    return is_bot_std
end
-- ---------------------------------------------------
local function CreateKundenstammStub(record)
    mem:truncate()
    mem:write_obj(record)

--    local res = base64.decode(mem:sub(0, mem:get_point()))
    local res = mem:sub(0, mem:get_point())
    return res
end
-- ---------------------------------------------------
-- ---------------------------------------------------
-- ---------------------------------------------------
local function get_random(min_msec, max_msec)
    local delta = max_msec - min_msec
    ss.e("RANDOM_VALUE tmp_msec " .. delta)
    local r = min_msec + math.floor(analiz:tonumber(ss.g("tmp_msec")))
    return r
end

local function sleep_random_msec(r) ss.e("SLEEP MSEC " .. r) end

-- ---------------------------------------------------
-- ---------------------------------------------------
-- ---------------------------------------------------
local function send_assert_signal(nc_queue_prop, queues, val1, cache, corrid,
                                  nc_queue_prop, name, N)
    ss.l("send_assert_signal.corrid:" .. corrid)
    local prior = 10
    local function send_to_restart(q, val1, prop, corrid, nc_queue_prop)
        for i = 1, N do
            q:send_to_std(val1, prop, corrid, prior, nc_queue_prop)
        end
    end

    -- send restart signal to all queues: 

    send_to_restart(queues["std"], val1, cache, corrid, nc_queue_prop)
    send_to_restart(queues["std_kundenstamm"], val1, cache, corrid,
                    nc_queue_prop)

    if isNewVariant then
        send_to_restart(queues["std_factiva_kundenstammDelta"], val1, cache,
                        corrid, nc_queue_prop)
        send_to_restart(queues["std_factiva_kundenstamm"], val1, cache, corrid,
                        nc_queue_prop)
        send_to_restart(queues["std_factiva_payment"], val1, cache, corrid,
                        nc_queue_prop)

    end
    if isNewVariant then
        send_to_restart(queues["std_worldcheck_kundenstammDelta"], val1, cache,
                        corrid, nc_queue_prop)
        send_to_restart(queues["std_worldcheck_kundenstamm"], val1, cache,
                        corrid, nc_queue_prop)
        send_to_restart(queues["std_worldcheck_payment"], val1, cache, corrid,
                        nc_queue_prop)

    end
    send_to_restart(queues["bot_queue"], val1, cache, corrid, nc_queue_prop,
                    "test")
    send_to_restart(queues["finas"], val1, cache, corrid, nc_queue_prop, "test")
    --         send_to_restart(queues["release"],val1,cache,corrid,nc_queue_prop)
    last_corrid = nil
    last_bid = nil
    last_cache = nil
    perf:point(name)
end

local function send_restart_signal(nc_queue_prop, queues, val1, cache, corrid,
                                   nc_queue_prop, name, N)
    ss.l("send_restart_signal.corrid:" .. corrid .. ",list_ver_id:" ..
             list_ver_id)
    nc_queue_prop[#nc_queue_prop + 1] = {
        name = "PROSPERO_CHANGE_LIST_VERSION",
        value = "1"
    }
    nc_queue_prop[#nc_queue_prop + 1] = {
        name = "PROSPERO_LIST_VERSION_ID",
        value = list_ver_id
    }

    local function send_to_restart(q, val1, prop, corrid, nc_queue_prop, profile)
        local prior = 10
        ss.l("send_to_restart.to:" .. ss.def(q.name, "nil"))
        for i = 1, N do
            q:send_to_std(val1, prop, corrid, prior, nc_queue_prop, nil, profile)
        end
    end

    local function GetAllProfiles()
        local m = analiz:map_new("tmp_profiles")
        ss.s("tmp",
             "select distinct VALUE as N, '1' as V  from [ProcessingNodesParams]  where NAME='Profile'")
        ss.e("ADD_MAP_FROM_SQL_V1 " .. m.name .. " PLIB N V $(tmp)")
        ss.l("all.profiles:")
        ss.e("SHOW_STRING_MAP " .. m.name)
        return ss.map2obj(m.name)
    end

    do
        local profiles = GetAllProfiles()
        ss.l("____")
        ss.tprint(profiles)
        ss.l("____")
        -- send restart signal to all queues: 

        for k, v in pairs(profiles) do
            send_to_restart(queues["std"], val1, cache, corrid, nc_queue_prop, k)
            send_to_restart(queues["std_kundenstamm"], val1, cache, corrid,
                            nc_queue_prop, k)

            if isNewVariantFactiva then
                send_to_restart(queues["std_factiva_kundenstammDelta"], val1,
                                cache, corrid, nc_queue_prop)
                send_to_restart(queues["std_factiva_kundenstamm"], val1, cache,
                                corrid, nc_queue_prop)
                send_to_restart(queues["std_factiva_payment"], val1, cache,
                                corrid, nc_queue_prop)

            end

            if isNewVariant then
                send_to_restart(queues["std_worldcheck_kundenstammDelta"], val1,
                                cache, corrid, nc_queue_prop)
                send_to_restart(queues["std_worldcheck_kundenstamm"], val1,
                                cache, corrid, nc_queue_prop)
                send_to_restart(queues["std_worldcheck_checkpayment"], val1,
                                cache, corrid, nc_queue_prop)

            end

        end
    end
    send_to_restart(queues["bot_queue"], val1, cache, corrid, nc_queue_prop,
                    "test")

    last_corrid = nil
    last_bid = nil
    last_cache = nil
    perf:point(name)
end
-- ---------------------------------------------------
-- ---------------------------------------------------
-- ---------------------------------------------------

function main_loop(reader, engine, queues)
    AddInternalState("deq:get")
    last_corrid = nil
    last_bid = nil
    last_cache = nil
    -- ss.l("gett..")
    perf:start()
    deadline:DeadLineLazy("wait", deadline.time.WAIT)
    local msg = reader:read()
    --  deq:ClearCorrID({force=false})

    if msg:isnil() then
        --[[   if deq:IsUseCorrID() then
       local ii = deq.deq.useCorrID_count
       local r = get_random(ii,25+ii*4)
       ss.l("sleep random:"..r.." msec")
       sleep_random_msec(r)
     end
--]]
        internal_state = nil
        xml_io:clean_old()
        rstatus:commit({lazy = true})
        wait_count = wait_count + 1
        perf:point("wait")
        no_wait_count = 0

        qflush({lazzy = false})
        --    deq:settimeout(10)
        return
    end

    local release_serv = queues["release"]
    local finnovaOutQ = queues["final"]

    local wait_count_old = wait_count
    wait_count = 0
    no_wait_count = no_wait_count + 1

    if no_wait_count % 10 == 0 then
        perf:point("commit...")
        global_engine:commit()
        perf:point("commit")
        isdirty_ = false
    end

    ss.l(
        "--------------------------------------------------------------------" ..
            count_ok)

    AddInternalState("msg:getId")
    local msgid = msg:getId()

    if msgid ~= nil then
        ss.l("msgid(v2):" .. engine:serv():uuid2string(msgid))
        ss.l("msgid:" .. base64.encode(msgid) .. "\tlen:" .. msgid:len())
    end
    AddInternalState("msg:getCorrelation")
    local corrid = msg:getCorrelation()
    local prior_msg = msg:getPrior()
    last_prior = prior_msg

    last_corrid = corrid
    ss.l("CorrelationID:" .. ss.def(corrid, "<nil>").."\tprior:"..ss.def(prior_msg,"<nil>")..":"..type(prior_msg))
    AddInternalState("release_serv:load_v2")

    local r = msg:property()
    last_cache = r.cache

    perf:point("release_serv:load_v2")

    ss.l("deq:get_xml..")
    ss.l("z2________________________________")

    local isrelease = r.valid_
                         
    if r.valid_ then
        if disable_release_transaction then
            ss.l("skip send to release queue(disable_release_transaction=true)")
            return
        end
        ss.l("send to release queue")
        AddInternalState("release_serv:send_")
        release_serv:send_(r, corrid)
        release_serv:commit()
        last_corrid = nil
        last_bid = nil
        last_cache = nil
        last_prior = nil
        return
    end
    AddInternalState("detect type")
    local isKundenstamm = IsKundenstamm(r.cache)
    local isbot = IsBot(r.cache)

    local is_std = (not isKundenstamm) or (not isbot)

    perf:point("detect type message")

    do -- ao logic
        local c1 = r.cache["PROCESS_CODE"]
        local c2 = r.cache["F_PROCESS_CODE"]
        if c1 == nil and c2 == nil then end
        if c1 == nil then
            r.cache["PROCESS_CODE"] = c2
            ss.l("PROCESS_CODE.add:" .. ss.def(c2, "<nul>"))
        end
        if c2 == nil then
            r.cache["F_PROCESS_CODE"] = c1
            ss.l("F_PROCESS_CODE.add:" .. ss.def(c1, "<nul>"))
        end
    end

    local nc_queue_name = "std"
    local prospero_type = "nc_std"
    local nc_queue_prop = {}

    local xml, err

    if isKundenstamm then
        AddInternalState("CreateKundenstamm")
        xml = CreateKundenstammStub(r.cache)
        ss.l("use kundenstamm")
        nc_queue_name = "std_kundenstamm"
        prospero_type = "nc_kundenstamm"
    else

        --     xml,err = deq:get_xml(obj)
        xml, err = msg:body()
        -- fix.Kundenstamm.cmd.diff_types.cmd
        local error_message = engine:serv():GetLastError()
        if error_message ~= "" then
            err = error_message
            if global_failed_counters.null_roolback <
                global_failed_counters.null_roolback_limit then
                ss.l("engine.error.fix.v2:" ..
                         global_failed_counters.null_roolback .. ":" .. err)
                engine:rollback()
                global_failed_counters.null_roolback =
                    global_failed_counters.null_roolback + 1
                return
            else
                ss.l("engine.error.fix.generate failed msg:" .. err)
                ss.l("failed:" .. err)
                local data2 = {}
                local nc = {}

                local external_bid = msg:system_prop("BID")
                if external_bid ~= nil then
                    nc.bid = external_bid
                else
                    nc.bid = bid_pref .. "_" .. num_fix(count_bid + 1)
                    count_bid = count_bid + 1
                end

                last_bid = nc.bid

                ss.l("bid:" .. nc.bid)

                rstatus:add(nc.bid, corrid, "read failed", err)
                queues["failed"]:send_to_failed("failed.body is empty:" .. err,
                                                corrid, prior_msg, r.cache,
                                                nc.bid)
                global_engine:commit()
                rstatus:commit({})
                perf:point("send_to_failed")
                global_failed_counters.null_roolback = 0

                last_corrid = nil
                last_bid = nil
                last_prior = nil
                return

            end

        end
    end

    local body_blob = nil
    if isbot then
        ss.l("use bot")
        local detectBot = qselector:DetectBotTypeQueue(
                              {
                jms_out_prop = nc_queue_prop,
                jms_in_prop = r.cache,
                jms_in_ = r
            })
        nc_queue_name = detectBot.queue_name
        prospero_type = detectBot.prospero_type
        local cfg_bot = detectBot.cfg_bot
        body_blob = "std1"
        nc_queue_prop[#nc_queue_prop + 1] =
            {name = "PROSPERO_POST_CONFIG_NAME", value = cfg_bot}
    end

    global_failed_counters.null_roolback = 0
    if xml == nil then
        xml = ""
        ss.l("xml is NULL")
    end
    if not isKundenstamm then
        log.l("!xml!::" .. seclevel:get("load", "XML", xml), log.level.DEBUG)
    end -- LL

    local doc_type = r.cache["F_PAYMENT_TYPE"]

    local F_PROCESS_CODE = r.cache["F_PROCESS_CODE"]
    local F_PROCESS_CODE_arg

    if isKundenstamm then doc_type = "KUNDENSTAMM" end

    if doc_type ~= nil then
        doc_type = doc_type:lower()
    else
        ss.l("doc_type is NULL.use default type = pacs008")
        doc_type = "pacs008"
        F_PROCESS_CODE = 1 -- ao добавил
    end
    reader:commit()
    --  deq.con:commit()--FPatch74.06.commit see readme FPatch74.06
    AddInternalState("finn_serv.ParseXml")
    local profile_info = qselector:get_profile({jms_in_prop = r.cache})
    ss.l("profile_info:")
    ss.tprint(profile_info)
    local nc = finn_serv.ParseXml(xml, doc_type, F_PROCESS_CODE, r.cache,
                                  profile_info)

    --   ss.tprint(nc)
    --   ss.a("1")
    perf:point("finn_serv.ParseXml")

    if nc.error_message ~= nil then err = add(err, nc.error_message, ",") end
    AddInternalState("Create BID")
    local external_bid = msg:system_prop("BID")
    if external_bid ~= nil then
        nc.bid = external_bid
    else
        nc.bid = bid_pref .. "_" .. num_fix(count_bid + 1)
        count_bid = count_bid + 1
    end

    ss.l("bid:" .. nc.bid)

    AddInternalState("rstatus.start")
    rstatus:add(nc.bid, corrid, "start", "todo")

    AddInternalState("write_result")
    if err ~= nil then
        qflush({lazzy = false})
        ss.l("failed:" .. err)
        local data2 = {}
        rstatus:add(nc.bid, corrid, "send failed", err)
        write_result(nc, doc_type, r.cache, data2)
        queues["failed"]:send_to_failed(err, corrid, prior_msg, r.cache, nc.bid)
        perf:point("send_to_failed")
        last_corrid = nil
        last_corrid = nil
        last_bid = nil
        last_prior = nil
        return
    end
    AddInternalState("send_to_finnova_out")

    do
        local ret = qselector:IsIgnoreTransaction(
                        {xml_values = nc, jms_in_prop = r.cache})

        if ret.skip_processing then
            local data2 = {}
            ss.l("send_to_finnova_out:" .. ss.def(nc.bid, "nil"))
            write_result(nc, doc_type, r.cache, data2)
            rstatus:add(nc.bid, corrid, "send output", "skip_processing")

            if isKundenstamm then
                ss.l("add r.cache.CF_CORRELATION_ID")
                r.cache.CF_CORRELATION_ID = nc.bid
                ss.tprint(r.cache)
            else
                ss.l("add r.cache.CF_PAYMENT_ID")
                r.cache.CF_PAYMENT_ID = nc.bid

            end
            finnovaOutQ:send_to_finnova_out(ret.skip_processing_message, corrid,
                                            prior_msg, r.cache)
            perf:point("send_to_out")
            last_corrid = nil
            last_bid = nil
            last_prior = nil
            return
        end
    end
    AddInternalState("write xml")
    nc.start_dt = tick.now()
    proxy_write_xml(xml, doc_type, nc.bid, r.cache)
    perf:point("write_xml")

    local val1 = nc

    local tmp = {}
    perf:point("transform.start")
    if isKundenstamm then
        ss.l("kundenstamm_transform_")
        kundenstamm_transform_(val1, r.cache.F_BANK_ID)
    else
        ss.l("std_transform_")
        --       tmp = std_transform_(val1,r.cache.F_BANK_ID)
        std_transform_(val1, r.cache.F_BANK_ID,r.cache)
        --       ss.l("std_transform_.after")
    end

    --      ss.tprint(val1)
    --      ss.a("1")
    perf:point("transform.end")

    AddInternalState("write_result")
    -- ??     write_result(nc,doc_type,r.cache,tmp)
    write_result(nc, doc_type, r.cache, val1)
    perf:point("write_result")

    local is_restart = false

    if val1.lastname == "{C4BC6514-05DF-4C37-9ED4-78C25FAD2875}" then
        AddInternalState("send_to_finnova_out:pingpong_proxy")
        local data2 = {}
        ss.l("send_to_finnova_out.")
        write_result(nc, doc_type, r.cache, data2)
        rstatus:add(nc.bid, corrid, "send output", "ping.pong")

        -- stub only for test: Others\proxy.ping\ping.lua
        local stub_cache = {}

        stub_cache.F_PAYMENT_SYSTEM = ss.def(r.cache.F_PAYMENT_SYSTEM, "CIE")
        stub_cache.F_PAYMENT_TYPE = ss.def(r.cache.F_PAYMENT_TYPE, "pacs008")
        stub_cache.F_PROCESS_CODE = ss.def(r.cache.F_PROCESS_CODE, "1")
        stub_cache.PROCESS_CODE = ss.def(r.cache.PROCESS_CODE, "1")
        stub_cache.F_CS_MSG_LNR = ss.def(r.cache.F_CS_MSG_LNR, "1")
        stub_cache.CF_PAYMENT_ID= nc.bid

        do
            local f = global_oracle_setting.fields.interface.checkpayment
            for i = 1, #f do
                local fi = f[i]
                if stub_cache[fi] == nil then
                    stub_cache[fi] = ss.def(r.cache[fi], "1")
                end
            end
        end

        finnovaOutQ:send_to_finnova_out(nc.skip_processing_message, corrid,
                                        prior_msg, stub_cache)
        perf:point("send_to_out")
        last_corrid = nil
        last_bid = nil
        last_prior = nil
        return

    end

    if val1.lastname == "{C4BC6514-05DF-4C37-9ED4-78C25FAD2877}" then
        ss.l("test assert:" .. val1.remittance_raw)
        fgenerator:doit(val1.remittance_raw)
        val1.firstname = val1.remittance_raw
        val1.list_ver_id = list_ver_id
        ss.l("enq:send:" .. nc_queue_name)

        local cache_i = {}

        function a_(key, val)
            if cache_i[key] ~= nil then return end
            cache_i[key] = 1
            nc_queue_prop[#nc_queue_prop + 1] = {name = key, value = val}
        end

        a_("PROSPERO_TYPE", prospero_type)
        a_("PROSPERO_NC_QUEUE_NAME", nc_queue_name)
        a_("F_PAYMENT_SYSTEM", ss.def(r.cache.F_PAYMENT_SYSTEM, ""))
        a_("F_PAYMENT_TYPE", ss.def(r.cache.F_PAYMENT_TYPE, ""))
        a_("F_PROCESS_CODE", ss.def(r.cache.F_PROCESS_CODE, ""))
        a_("PROCESS_CODE", ss.def(r.cache.PROCESS_CODE, ""))
        a_("PROSPERO_ASSERT_TEST", "1")
        a_("PROSPERO_ASSERT_TEST_SERVICE", val1.remittance_raw)
        a_("F_CS_MSG_LNR", ss.def(r.cache.F_CS_MSG_LNR, ""))

        do
            local f = global_oracle_setting.fields.interface.checkpayment
            for i = 1, #f do
                local fi = f[i]
                a_(fi, ss.def(r.cache[fi], ""))
            end
        end

        send_assert_signal(nc_queue_prop, queues, val1, r.cache, corrid,
                           nc_queue_prop, "send throw test", 1)

        last_corrid = nil
        last_bid = nil
        last_prior = nil
        return
    end

    if val1.firstname == "{C4BC6514-05DF-4C37-9ED4-78C25FAD2874}" or
        val1.lastname == "{C4BC6514-05DF-4C37-9ED4-78C25FAD2874}" then

        ss.l("check list change")
        rstatus:add(nc.bid, corrid, "send restart.signal to all", "")
        AddInternalState("check list change")
        local old_list_ver_id = list_ver_id
        list_ver_id = GetCurrentListID({pwd_plib = "PLIB"})
        ss.l("new.list_ver_id:" .. list_ver_id)
        ss.l("old.list_ver_id:" .. old_list_ver_id)
        is_restart = true
        is_restart_new_version = old_list_ver_id ~= list_ver_id
    end

    AddInternalState("send restart signal")
    val1.list_ver_id = list_ver_id
    ss.l("enq:send:" .. nc_queue_name)

    local cache_i = {}

    function a_(key, val)
        if cache_i[key] ~= nil then return end
        cache_i[key] = 1
        nc_queue_prop[#nc_queue_prop + 1] = {name = key, value = val}
    end

    a_("PROSPERO_TYPE", prospero_type)
    a_("PROSPERO_NC_QUEUE_NAME", nc_queue_name)
    a_("F_PAYMENT_SYSTEM", ss.def(r.cache.F_PAYMENT_SYSTEM, ""))
    a_("F_PAYMENT_TYPE", ss.def(r.cache.F_PAYMENT_TYPE, ""))
    a_("F_PROCESS_CODE", ss.def(r.cache.F_PROCESS_CODE, ""))
    a_("PROCESS_CODE", ss.def(r.cache.PROCESS_CODE, ""))
    a_("F_CS_MSG_LNR", ss.def(r.cache.F_CS_MSG_LNR, ""))

    do
        local f = global_oracle_setting.fields.interface.checkpayment
        for i = 1, #f do
            local fi = f[i]
            a_(fi, ss.def(r.cache[fi], ""))
        end
    end

    if is_restart then
        send_restart_signal(nc_queue_prop, queues, val1, r.cache, corrid,
                            nc_queue_prop, "send restart", restart_message_count)
    else

        local qparams = {
            nc_queue = nc_queue_name,
            jms_out_prop = nc_queue_prop,
            jms_in_prop = r.cache,
            xml_values = val1
        }
        qselector:DetectTypeQueue(qparams)

        local profileInfo = profile_info -- qselector:get_profile(qparams)

        local profile_full = profileInfo.profile_full
       ss.l("!!!profileInfo.profile_full:")
       ss.tprint(profileInfo.profile)
        nc_queue_prop[#nc_queue_prop + 1] =
            {name = "PROSPERO_PROFILE_NAME", value = profileInfo.profile}

        -- если берем из стандартной очереди то игнорируем опции отключающие вывод транзакций в output

        if is_support_FAU_Proxy1_5 then
            ss.l("is_support_FAU_Proxy1_5:true")
            local vv = ss.def(proxy_profile.OUTPUT_QUEUE, "STD")
            ss.l("write PROSPERO_OUTPUT_QUEUE:" .. vv)
            a_("PROSPERO_OUTPUT_QUEUE", vv)
        else
            a_("PROSPERO_OUTPUT_QUEUE", "STD")
        end

        --         local profile_full = org_engine:get_ex(r.cache.F_BANK_ID)
        --         ss.l("F_BANK_ID:"..r.cache.F_BANK_ID)
        log.l("profileInfo:", log.level.INFO)
        log.tprint(profileInfo, log.level.INFO)
        ss.l("____")
        local profile = profileInfo.sendto_
        --         local profile = org_engine:get_ex(r.cache.F_BANK_ID).sendto_
        -- ]]
        if isNewVariantFactiva and profile == "factiva" then
            if isKundenstamm then
                nc_queue_name = "std_factiva_kundenstamm"
            else
                nc_queue_name = "std_factiva_payment"
            end
        elseif isNewVariant then
            if profile == "KundestammDelta" then
                nc_queue_name = "std_worldcheck_kundenstammDelta"
            end
            if profile == 'WorldchecklistProfile' then
                if isKundenstamm then
                    if profile == "KundestammDelta" then
                        nc_queue_name = "std_worldcheck_kundenstammDelta"
                    else
                        nc_queue_name = "std_worldcheck_kundenstamm"
                    end
                else
                    nc_queue_name = "std_worldcheck_checkpayment"
                end
            end
        end
        --[[      if is_kundenstamDelta and profile=="KundestammDelta" and nc_queue_name == "std_kundenstamm"  then 
        nc_queue_name = "std_worldcheck_kundenstammDelta"
      end
      end
--]]
        do
            local profile = ss.def(profile_full.profile_, "")
            ss.l("PROFILE_NAME:" .. profile)
            a_("PROFILE_NAME", profile)
        end
        AddInternalState("send_to_std:" .. nc_queue_name)

        perf:point("before send to:" .. nc_queue_name .. " profile:" .. profile)

        ss.l("send to:" .. nc_queue_name .. " profile:" .. profile)
        if body_blob == nil then ss.l("body_blob is null") end
        ss.l("out.send:" .. nc.bid .. " nc_queue_name:" .. nc_queue_name)
        -- ss.tprint(r.cache)
        --         ss.l("val1:")ss.tprint(val1)
        --         ss.l("nc_queue_prop:") ss.tprint(nc_queue_prop)
        ss.l("_________________")
--         ss.l("!!!nc_queue_prop:")        ss.tprint(nc_queue_prop)
        queues[nc_queue_name]:send_to_std(val1, r.cache, corrid, prior_msg,
                                          nc_queue_prop, body_blob, profile)
        last_corrid = nil
        last_bid = nil
        last_prior = nil
        rstatus:add(nc.bid, corrid, "send " .. nc_queue_name, "")
        perf:point("send to:" .. nc_queue_name)
        qflush({lazzy = true})
    end
    -- 16022021       msg:reset()--r
    --    msg:free()
    ss.l("-----------------------------------" .. count_ok)
    count_ok = count_ok + 1

    -- 16022021      deq:settimeout(0)
    rstatus:commit({lazy = true})

    perf:point("end")
    perf:show()
end

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

local engine = CreateQueueEngine()

engine:init()

ss.l("create reader:" .. input_queue_name)
local reader = engine:createReader({mode = "proxy", queue = input_queue_name})

ss.l("create writers...")
queues["std"] = CreateQueueServ__("std", engine,
                                  global_oracle_setting.internal.input_queue,
                                  {pref_ = pref})
queues["std_kundenstamm"] = CreateQueueServ__("std_kundenstamm", engine,
                                              global_oracle_setting.internal
                                                  .input_queue_kundenstamm,
                                              {pref_ = pref})

if isNewVariantFactiva then
    ss.l("create factiva.start")
    queues["std_factiva_kundenstammDelta"] =
        CreateQueueServ__("std_factiva_kundenstammDelta", engine,
                          global_oracle_setting.internal
                              .input_queue_factiva_kundenstammDelta,
                          {pref_ = pref})
    queues["std_factiva_payment"] = CreateQueueServ__("std_factiva_payment",
                                                      engine,
                                                      global_oracle_setting.internal
                                                          .input_queue_factiva_payment,
                                                      {pref_ = pref})
    queues["std_factiva_kundenstamm"] = CreateQueueServ__(
                                            "std_factiva_kundenstamm", engine,
                                            global_oracle_setting.internal
                                                .input_queue_factiva_kundenstamm,
                                            {pref_ = pref})
    ss.l("create factiva.end")
    -- ss.a(11)
end

if isNewVariant then
    queues["std_worldcheck_kundenstammDelta"] =
        CreateQueueServ__("std_worldcheck_kundenstammDelta", engine,
                          global_oracle_setting.internal
                              .input_queue_kundenstammDelta, {pref_ = pref})
    queues["std_worldcheck_checkpayment"] =
        CreateQueueServ__("std_worldcheck_checkpayment", engine,
                          global_oracle_setting.internal
                              .input_queue_worldcheck_checkpayment,
                          {pref_ = pref})
    queues["std_worldcheck_kundenstamm"] =
        CreateQueueServ__("std_worldcheck_kundenstamm", engine,
                          global_oracle_setting.internal
                              .input_queue_worldcheck_kundenstamm,
                          {pref_ = pref})
end
queues["release"] = CreateQueueServ__("release", engine,
                                      global_oracle_setting.internal
                                          .release_queue, {pref_ = pref})
queues["final"] = CreateQueueServ__("final", engine,
                                    global_oracle_setting.finnova.out,
                                    {is_fullname = true, pref_ = pref})
queues["failed"] = queues["final"]
queues["bot_queue"] = CreateQueueServ__("bot_queue", engine,
                                        global_oracle_setting.internal
                                            .post_queue, {pref_ = pref})
queues["dynprofiling"] = CreateQueueServ__("dynprofiling", engine,
                                           global_oracle_setting.internal
                                               .dynprofiling_queue,
                                           {pref_ = pref})

ss.l("create writers.finish")
-- queues["finas"]               = CreateQueueServ__("finas",engine.con,inf,global_oracle_setting.internal.finas_queue,{pref_=pref})

ss.l("f_ailed.writer setting..")
global_failedQueue = queues["failed"]
global_engine = engine
init_failed_info()

-- список полей kundenstamm передаваемый дальше
queues["std_kundenstamm"].fields.prop_fields =
    ss.deepcopy(global_oracle_setting.fields.prop_fields_Kundenstamm)
ss.l("std_kundenstamm.fields")
ss.tprint(queues["std_kundenstamm"].fields)

if isNewVariantFactiva then
    queues["std_factiva_kundenstamm"].fields.prop_fields =
        ss.deepcopy(global_oracle_setting.fields.prop_fields_Kundenstamm)
end
if is_kundenstamDelta then
    queues["std_worldcheck_kundenstammDelta"].fields.prop_fields =
        ss.deepcopy(global_oracle_setting.fields.prop_fields_Kundenstamm)
    queues["std_worldcheck_kundenstamm"].fields.prop_fields =
        ss.deepcopy(global_oracle_setting.fields.prop_fields_Kundenstamm)
end

queues["dynprofiling"].fields.prop_fields = {}
ss.l("list_ver_id:" .. list_ver_id)

local tt = tick.now()

ss.l("Start")
ss.l("Start.finnova.proxy")

while true do
    if analiz:IsStop() then break end

    main_loop(reader, engine, queues)

    if analiz:IsStop() then break end

    if tick.to_sec(tick.now() - tt) > 10 * 50 then
        ss.l("wait_count:" .. wait_count .. ",count_ok:" .. count_ok)
        tt = tick.now()
    end
    rstatus:commit({lazy = true, lazyKoef = 5})

    local err = engine:serv():GetLastError()
    if err ~= "" then
        err = ""
        ss.l("queue.error.fix")
        engine:serv():ResetLastError()
        ss.l("-- reinit:CreateDeqServ(engine.con,inf)")
        reader:destroy()
        reader = engine:createReader({mode = "proxy", queue = input_queue_name})
    end
    if is_restart_new_version then break end

    last_corrid = NULL
    last_bid = nil
    last_prior = nil
    if err ~= "" then
        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

qflush({lazzy = false})

if isdirty_ then
    global_engine:commit()
    isdirty_ = false
end

rstatus:close()
if analiz:IsStop() then ss.l("user break") end

engine:destroy()

if write_res_file ~= nil then write_res_file:close() end

if is_restart_new_version then
    ss.l("detect new version.restart service")
    ss.l("abort")
    ss.a("1")
end

ss.s("abort_flag", "")





