local ss = analiz.libs.serv

local ss = analiz.libs.serv
local finn_serv = analiz.mix.finnova.serv
local tick = CreateTimeServ()
local log = analiz.libs.logserv
local mem = ss.CreateMemItem(100000)

local THRESHOLD = 150
--local MAX_PRIOR_COUNT=2
--local MAX_PRIOR_COUNT=2

local RECONNECT_COUNT_=10
--local RECONNECT_COUNT_=2

local function GetMaxPriorCount()
 local gps = global_prospero_setting
 if gps.options.queue_engine~=nil then 
    return ss.def(gps.options.queue_engine.max_prior,1)
 end

 return 1
end

local locID_=-1
local function GetObjID_()
  locID_=locID_+1
--  ss.l("GetObjID_.id:"..locID_) ss.l(debug.traceback())
  return locID_
end

local emulate_oracle_null_cache={}
local emulate_oracle_null_count=1


function CreateIOv2()
    local io_ = {}

    function io_:load(mem)
        local corrid = mem:read_string(THRESHOLD)
        return mem:read_obj()
    end

    function io_:save(mem, obj)

        local corrid = "xxx"
        local s = obj.SYSTEM
        if s ~= nil then
            if s.corrid ~= nil then
                corrid = s.corrid
            elseif s.CORRELATION ~= nil then
                corrid = s.CORRELATION
            end
        end
        local len = corrid:len()
        if corrid == "xxx" then
            ss.l("failed corrid")
            ss.l(debug.traceback())
            ss.l("_____________________________")
            ss.tprint(obj)
            ss.l("_____________________________")
        end
        if len > THRESHOLD then
            corrid = corrid:sub(1, THRESHOLD)
        elseif len < THRESHOLD then
            corrid = corrid .. string.rep(" ", THRESHOLD - len)
        end
        ss.l("io_:save:" .. #corrid)
        mem:truncate()
        mem:write_string(corrid)
        mem:write_obj(obj)
    end

    return io_
end

local function GetListName_(profilename)
    ss.l("GetListName_...:" .. ss.def(profilename, "<nil>"))
    if profilename == "default_profile" or profilename == nil then
        ss.l("profilename load from SYSTEM ")
        local prop = LoadProfilePropertyFor("SYSTEM")
        profilename = ss.def(prop["NC_type_of_list"], "info4c")
    end

    if profilename == "WorldchecklistProfile" or profilename == "worldcheck" then
        return "worldcheck"
    elseif profilename == "factiva" then
        return "factiva"
    elseif profilename == "info4c" then
        return "info4c"
    end
    ss.l("unknow map profile->listname:" .. profilename)
    ss.l(debug.traceback())
end

local function queryname2brokername(reader_writer_mode, engine_options, options)

    ss.l("queryname2brokername:" .. reader_writer_mode .. ":" ..
             ss.toJson(options))

    local queryname = options.queue

    if queryname == global_oracle_setting.internal.post_queue then
        local internal = global_prospero_setting.options.brokers.nc_internal
        local broker_op = ss.deepcopy(global_prospero_setting.options.brokers
                                          .nc_internal)
        broker_op.name = "post_queue"
        broker_op.max_prior = GetMaxPriorCount()
        return broker_op
    end


    if queryname == "CSUFAFOUT" or queryname==global_oracle_setting_local.finnova.from then

        return global_prospero_setting.options.brokers.nc_input
    end
    if queryname == "CSUFAFIN" or queryname==global_oracle_setting_local.finnova.out then
        local broker_op = global_prospero_setting.options.brokers.nc_output
        if engine_options.system == "testsystem" then
            broker_op = global_prospero_setting.options.brokers.nc_test
        end
        return broker_op
    end

    local internal = global_prospero_setting.options.brokers.nc_internal
    local broker_op = ss.deepcopy(global_prospero_setting.options.brokers
                                      .nc_internal)
    -- broker_op.name= options.name

    broker_op.name = finn_serv.RemoveSchemaName(options.queue)

    if broker_op.name == "FAU_kundenstamm5" and options.corrid ==
        "KundestammDelta" then
        broker_op.name = "worldcheck.kundenstamm.delta"
        broker_op.max_prior = GetMaxPriorCount()
        broker_op.rollback=true
    elseif broker_op.name == "FAU_kundenstamm5" then
        broker_op.name = GetListName_(options.corrid) .. ".kundenstamm"
        broker_op.max_prior = GetMaxPriorCount()
        broker_op.rollback=true
    elseif broker_op.name == "FAU_queue_i_input5" then
        broker_op.name = GetListName_(options.corrid) .. ".payment"
        broker_op.max_prior = GetMaxPriorCount()
        broker_op.rollback=true
    else            
        broker_op.name = broker_op.name
        broker_op.rollback=true
    end

    ss.l("broker_op:______________________________")
    ss.tprint(options)
    ss.l("______________________________")
    ss.l("options.queue:" .. options.queue)
    ss.l("broker_op.name:" .. broker_op.name)

    ss.l("nc_schema:" .. ss.def(global_oracle_setting_local.nc_schema, "nil"))


    if broker_op.name == "FAU_finas_input5" then
        broker_op.max_prior = GetMaxPriorCount()
    end

    if not (broker_op.name == "FAUs_VMT_oof1_m_0" or broker_op.name ==
        "FAU_finas_input5") then

    end
    return broker_op
end
local function remove_(objs, obj)
    for i = 1, #objs do
        if objs[i] == obj then
            table.remove(objs, i)
            break
        end
    end
end

local function is_array(table)
    if type(table) ~= 'table' then return false end

    -- objects always return empty size
    if #table > 0 then return true end

    -- only object can have empty length with elements inside
    for k, v in pairs(table) do return false end

    -- if no elements it can be array and not at same time
    return true
end
-- ---------------------------------------------------------------------------------
local function createServEngine_(ora_engine)
    local serv = {engine_ = ora_engine}

    function serv:CreateCorrID()
        local op = self.engine_.options
        if op.system == "testsystem" then
--            local pref = "nc_restapi:"
            local pref = "test:"
--            local pref = "nc_rest_and_test:"
            local corrid =
                pref .. op.sender_type .. ":" .. op.sender_handle .. ":" ..
                    ss.b("$(GUID)")
            return corrid
        end
        return ss.b("$(GUID)")
    end

    function serv:GetLastError() return ss.def(self.engine_.emessage, "") end
    function serv:ResetLastError() self.engine_.emessage = nil end
    function serv:uuid2string(msgid) return msgid end
    function serv:IsExpired(state)
        return false -- state==oMQ.types.message_state.OCI_AMS_EXPIRED
    end
    function serv:Convert_DT_to_mssqlDT(val) return DateOracle2ms(val) end

    return serv

end

-- ---------------------------------------------------------------------------------
local function createReaderMsg_(reader)
    ss.l("create createReaderMsg_")
    local msg = {reader = reader}

    function msg:clear()
        self.task = nil
        self.objs = nil

    end
    function msg:isnil() return self.task == nil end

    function msg:body()
        if self.task == nil then return "" end
        if self.task.BODY == nil then return "" end
        return self.task.BODY
    end

    function msg:getCorrelation()
        ss.l("msg:getCorrelation")
--         ss.tprint(self.task)
        if self.task == nil then return "-1" end
        if self.task.SYSTEM == nil then return "-1" end

        local c = self.task.SYSTEM.corrid
        if c ~= nil then return c end
        return ss.def(self.task.SYSTEM.CORRELATION, "-1")
    end
    function msg:getPrior()
        if self.task == nil then return "1" end
        if self.task.SYSTEM == nil then return "1" end
        if self.prior~=nil then return ""..self.prior end
        return ss.def(self.task.SYSTEM.PRIOR, "1")
    end

    function msg:system_prop(name)
        local s = self.task.SYSTEM
        if s == nil then s = {} end
        local val = s[name]
        if val ~= nil then return val end
        if name == "corrid" then return self:getCorrelation() end
        if name == "prior" then return self:getPrior() end
        return nil
    end

    function msg:property()
        -- is_array
        local function arr2cache_(cache, vals)

            for i = 1, #vals do
                local ii = vals[i]
                local value = "" .. ss.def(ii.val, ii.value)
                cache[ii.name] = value
                --       ss.l("add:"..i..":"..ss.def(ii.name,"nil")..":"..ss.def(value,"nil"))
            end

        end

        ss.l("msg:property")
        if self.task == nil then
            ss.l("task:null")
            return {}
        end
        if self.task.PROPERTIES == nil then
            ss.l("task.task.PROPERTIES null")
            return {}
        end
        local t = {} -- ss.deepcopy(self.task.PROPERTIES)
        if self.task.PROPERTIES.arr ~= nil then
            t.arr = ss.deepcopy(self.task.PROPERTIES.arr)
            t.cache = {}
            arr2cache_(t.cache, self.task.PROPERTIES.arr)
        else
            ss.l("ss.deepcopy(self.task.PROPERTIES)...")
            if is_array(self.task.PROPERTIES) then
                t.cache = {}
                arr2cache_(t.cache, self.task.PROPERTIES)
                t.arr = ss.deepcopy(self.task.PROPERTIES)
            else
                t.cache = ss.deepcopy(self.task.PROPERTIES)
            end

        end

        local emulate_oracle_null = ""
        for k, v in pairs(t.cache) do
            if type(v) ~= "string" then
                --  ss.l("msg:property.convert to string:"..k..":"..v)

                t.cache[k] = "" .. v
            elseif v == "" then
                t.cache[k] = nil
                --           ss.l("emulate oracle null for "..k) 
                if emulate_oracle_null ~= "" then
                    emulate_oracle_null = emulate_oracle_null .. ";"
                end
                emulate_oracle_null = emulate_oracle_null .. k
            end
        end
        if emulate_oracle_null ~= "" then
            local v = emulate_oracle_null_cache[emulate_oracle_null]
            if v==nil then
              v = ""..emulate_oracle_null_count
              emulate_oracle_null_cache[emulate_oracle_null] = v
              emulate_oracle_null_count=emulate_oracle_null_count+1
            else
              emulate_oracle_null=""
            end
            
            ss.l("emulate oracle null for:N=" ..v..":".. emulate_oracle_null)
        end

        --    ss.l("self.task.PROPERTIES:")    ss.tprint(t)    ss.l("self.task.PROPERTIES.end")

        t.arr = self.task.PROPERTIES
        ss.l("11.size:" .. #t.arr)
        t.valid_ = finn_serv.IsRelease(t.cache)
        return t
    end

    function msg:getId() return "msg:getId todo" end

    function msg:getExpiration() return "msg:getExpiration() todo" end
    function msg:getState()
        return 0 -- todo
    end
    function msg:read_lex()
        ss.l("msg:read_lex.start")
        local prop = self:property()
        -- ss.l("prop val:")ss.tprint(prop)
        local ret = {}

        ret.bid = prop.cache.bid
        ret.score = prop.cache.score
        ret.worker = prop.cache.worker
        ret.start_dt = prop.cache.start_dt -- ss.def(obj:getd("start_dt",0))
        return ret
    end

    return msg
end
local function CreateSimpleJHeader_()
    local h = {}

    h.properties = {}
    h.prop_index_cache = {}
    h.data = {}

    function h:clear()
        ss.l("jms.clear")
        self.properties = {}
        self.prop_index_cache = {}
    end

    function h:set_string_property(key, val)
        local index = self.prop_index_cache[key]
        local is_new = "no"
        if index == nil then
            index = #self.properties + 1
            is_new = "yes"
        end
        if key=="PROSPERO_PROFILE_NAME" then 
            ss.l("!!!PROSPERO_PROFILE_NAME:"..val)
        end
        self.properties[key] = val

        self.prop_index_cache[key] = index
    end

    function h:show()
        if not (log.IsPrint(log.level.DEBUG)) then return end
        ss.l("show prop.start")
        for i = 1, #self.properties do
            local p = self.properties[i]
            if p == nil then
                ss.l("self.properties[i]:" .. i .. " is NULL")
            else
                ss.l(ss.def(p.NAME.data, "<!nil!>") .. ":" ..
                         ss.def(p.STR_VALUE.data, "<!nil!>"))
            end
        end
        ss.l("show prop.end")
    end

    function h:set_type(val) self.data.TYPE = val end
    function h:set_appid(val) self.data.APPID = val end
    function h:set_protocol(val) self.data.PROTOCOL = val end

    function h:set_groupid(val) self.data.GROUPID = val end

    function h:set_groupseq(val) self.data.GROUPSEQ = val end

    function h:write_header(obj)
        ss.l("h:write_header")
        if obj.PROPERTIES == nil then obj.PROPERTIES = {} end
        self:write_prop(obj.PROPERTIES)

    end

    function h:write_prop(prop)
        ss.l("jmsprop.write_prop.start")
       if self.properties==nil then
             ss.l("jmsprop.write_prop. self.properties is null")
             ss.l(debug.traceback())
       end
       if prop==nil then
             ss.l("jmsprop.write_prop. prop is null")
             ss.l(debug.traceback())
       end

        for k, v in pairs(self.properties) do
            if prop[k]~=nil then 
             ss.l("rewrite h:write_prop.key="..k.." old:"..prop[k].." new:"..v)
            end
            prop[k] = v
            --     ss.l(k..":"..v)
        end
        ss.l("jmsprop.write_prop.end")
    end

    return h
end

-- ---------------------------------------------------------------------------------
local function createWriter_(engine, options)
    ss.l("createWriter_")

    if options.queue == 'FAUs_VMT_oof1_m_0' or options.queue ==
        'PROSPERO_V263_FAU.FAUs_VMT_oof1_m_0' then
        ss.tprint(options)
        ss.l(debug.traceback())

        -- ss.a(1)
    end

    local writer = {engine_ = engine, options = options, ex_engines_ = {}}
    writer.name = "msbroker.writer"
    writer.mode = "simple"
    writer.jheader = CreateSimpleJHeader_()

    writer.fields = ss.deepcopy(global_oracle_setting.fields)

    local broker_option =
        queryname2brokername("writer", engine.options, options)
    broker_option.io_obj = CreateIOv2()
    writer.id=GetObjID_()
  ss.l("__________________________________________________________________________.start")

    local ii=0
    while true do 

        writer.impl_ = ss.factory.msbroker.CreateSender(broker_option)
        local isfailed = writer.impl_:IsInitFailed()
        if isfailed==false then 
           break
        end
       ss.l("writer init false.ii:"..ii)
       if analiz:IsStop() then 
           ss.l("user break")

       end
       ss.l("sleep 5 sec")
       ss.e("sleep sec 5")
       ii=ii+1
       if ii>120 then break end --10 min
    end

  ss.l("__________________________________________________________________________.end")
    writer.broker_option = broker_option
    writer.impl_:login()

    function writer:commit() self.engine_:commit() end

-- --------------------------------------------
    function writer:send_(prop, corrid, body)
  ss.l("writer:send_")
        local obj = {
            corrid = corrid,
            PROPERTIES=ss.deepcopy(prop.cache),
            body=body
}
            self:write(obj, {})

    end
-- --------------------------------------------
    function writer:send_to_std(val1, prop, corrid, prior, nc_queue_prop,
                                body_blob, profile)

        --    local obj={prir=prior,corrid=corrid,body=val1,PROPERIES={}}
        local obj = {
            prir = prior,
            corrid = corrid,
            xml_prop = val1,
            PROPERIES = {},
            body = "",
            profile = profile
        }

        local function add_(cache, arr)
            ss.l("add_:"..#arr)
            for i = 1, #arr do
                local ii = arr[i]
                cache[ii.name] = ii.value
            end
        end

        obj.PROPERTIES = ss.deepcopy(prop)
        add_(obj.PROPERTIES, nc_queue_prop)

        ss.l("writer:send_to_std:" .. ss.def(profile, "nil")..":prior:"..prior)
--        ss.tprint(val1)        ss.l("____________________")        ss.tprint(prop)        ss.l("____________________")        ss.tprint(nc_queue_prop)        ss.l("____________________")

        local function get_engine_(name, normal_queue_name)
            local engine_name = self.options.name .. "_" .. name
            local e = self.ex_engines_[engine_name]
            if e ~= nil then return e end
            local e = CreateQueueServ__(self.options.name, self.engine_,
                                        normal_queue_name, {pref_ = nil})
            self.ex_engines_[engine_name] = e
            return e
        end

        if profile == "WorldchecklistProfile" and self.options.name ==
            "std_kundenstamm" then
            get_engine_("WorldchecklistProfile", "worldcheck.kundenstamm"):write(
                obj, {})
        elseif profile == "WorldchecklistProfile" and self.options.name == "std" then
            get_engine_("WorldchecklistProfile", "worldcheck.payment"):write(
                obj, {})
        elseif profile == "factiva" and self.options.name == "std_kundenstamm" then
            get_engine_("factiva", "factiva.kundenstamm"):write(obj, {})
        elseif profile == "factiva" and self.options.name == "std" then
            get_engine_("factiva", "factiva.payment"):write(obj, {})
        elseif profile == "KundestammDelta" then
            get_engine_("KundestammDelta", "worldcheck.kundenstamm.delta"):write(
                obj, {})
        else
            self:write(obj, {prior=prior})
        end
    end

    function writer:write(obj, options)

        if options==nil then options={} end
        local prior = ss.def(options.prior,"1")
        ss.l("writer:write:" .. ss.toJson(self.broker_option)..":prior:"..prior)
--        ss.l(debug.traceback())
        -- start.only for test system
        if obj.prop ~= nil and obj.PROPERTIES == nil then
            obj.PROPERTIES = obj.prop
            obj.prop = nil
        end
--      ss.l("_______________________") ss.tprint(obj.PROPERTIES)      ss.l("_______________________")
        if obj.body ~= nil and obj.BODY == nil then
            obj.BODY = obj.body
            obj.body = nil
        end
        if obj.SYSTEM == nil then obj.SYSTEM = {} end

        if obj.corrid ~= nil then
            obj.SYSTEM.CORRELATION = obj.corrid
            obj.corrid = nil
        end

        ss.l("write corrid:" .. ss.def(obj.SYSTEM.CORRELATION, "<nil>"))
        -- end.only for test system
        self.jheader:clear()
        finn_serv.ConvertXmlObjToQueueObj_v2(self.jheader, obj.xml_prop,
                                             obj.prop1, self.fields)
 local function week_add_profile_(name)
    if obj.PROPERTIES[name]~=nil then 
         local val = obj.PROPERTIES[name]
         ss.l("week_add_profile_.PROPERTIES:"..name.."="..val) 
         self.jheader:set_string_property("PROFILE_NAME",val)
         return
    end
    if obj.profile==nil then ss.l("week_add_profile_:obj.profile is NULL")     ss.l(debug.traceback()) return end
    ss.l("week_add_profile_:"..name..":"..obj.profile)
--    ss.l(debug.traceback())
    self.jheader:set_string_property(name,obj.profile)
 end
   week_add_profile_("PROSPERO_PROFILE_NAME")
   week_add_profile_("PROFILE_NAME")
--[[        if obj.profile ~= nil then
            self.jheader:set_string_property("PROSPERO_PROFILE_NAME",
                                             obj.profile)
            self.jheader:set_string_property("PROFILE_NAME", obj.profile)
        end
--]]
        ss.l("writer:write:" .. ss.toJson(self.broker_option))
        self.jheader:write_prop(obj.PROPERTIES)
        -- ss.l("obj.PROPERTIES:")ss.tprint(obj.PROPERTIES)ss.l("_______________")ss.l(debug.traceback())
        -- ss.a(1)

        --   ss.tprint(obj)
        if self.options.mode == "writer2" then
            writePerf("msbroker.sendobj.start")
            self.impl_:sendobj(obj,{prior=prior})
            writePerf("msbroker.sendobj.end")
            return
        end

        local body = nil

        -- nil {A4E5E851-C5FA-493C-A2C9-A5EB5940E3B4}
        local body_val = nil
        local is_transform = false
        if obj.BODY ~= nil and obj.BODY:find("\n") ~= nil then

            body_val = obj.BODY
            obj.BODY = "{A4E5E851-C5FA-493C-A2C9-A5EB5940E3B4}"
            is_transform = true
        end
        local body = table.show(obj, "")

        if body:find(" = ", 1, true) == 1 then
            ss.l("remove =")
            body = body:sub(4)
        end

        if is_transform == true then
            ss.l("replace body")
            body = ss.ReplaceValue(body,
                                   '"{A4E5E851-C5FA-493C-A2C9-A5EB5940E3B4}"',
                                   "[===[" .. body_val .. "]===]")
        end
--         ss.l("sendval2")ss.l(body)


--        self.impl_:sendval2(body)
        writePerf("msbroker.sendval.start")

        local emessage = self.impl_:sendval(body,{prior=prior})
        if ss.def(emessage,"")~="" then 
          ss.l("failed.sendval:"..emessage)
          for i=1,RECONNECT_COUNT_ do 
              ss.l("failed:"..i.."sendval:"..emessage)
              ss.l("sleep "..i.." sec")
              ss.l(debug.traceback())
              ss.e("sleep sec "..i)
              self.impl_:reconnect({force=i>3})
              emessage = self.impl_:sendval(body,{prior=prior})
              if ss.def(emessage,"")=="" then 
                ss.l("send.i:"..i..":ok")
                break
              end
          end
        end

        writePerf("msbroker.sendobj.start")
       return ss.def(emessage,"") 
    end

    function writer:get_obj_(message, corrid, prior_arg, cache, body_message,
                             nc_params)
   ss.l("writer:get_obj_")
        if nc_params == nil then nc_params = {} end
        local obj = {}
        obj.PROPERTIES = {}
        obj.corrid = corrid
        obj.BODY = body_message

        local function add_(ii)
            if ii == nil then
                ss.l("failed. ii is null")
                ss.l(debug.traceback())
            end
            if ii.val == nil then
                ss.l("failed. ii.val is null:name:" .. ii.name)
                ss.l(debug.traceback())
                ss.l("----------")
                ss.tprint(ii)
                ss.l("----------")
            end

            if ii.val ~= nil then
--                       ss.l("writer:get_obj_.add:"..ii.name..":"..ii.val)
                obj.PROPERTIES[ii.name] = ii.val
            end

        end

        local function add2_(cache, fields)
            ss.l("add2_.size:" .. #fields)
            local count = 0
            for i = 1, #fields do
                local ff = fields[i]
                local vv = cache[ff]
                --     ss.l("add2:"..ff)
                if vv ~= nil then
                    count = count + 1
                    add_({name = ff, val = vv})
                end
            end
            ss.l("add2_.added:" .. count)
        end

 local function  convert_nc_param_to_cache_(params)
    local c={}
    for i=1,#params do
      local p=params[i]
      c[p.name]=p.val
    end
  return c
 end

      ss.l("calc_bid...")
        local iskundestamm = ss.g("TYPE2") == "Kundenstamm" or
                                 finn_serv.IsKundenstamm(cache)
        local bid = "writer:get_obj_.bid"
        local score = nil -- todo
        local nc_param_cache = convert_nc_param_to_cache_(nc_params)
        if finn_serv.IsKundenstamm(cache) then
           ss.l("kundenstamm:cache.CF_CORRELATION_ID")
            bid = cache.CF_CORRELATION_ID
        else
          if cache.CF_PAYMENT_ID~=nil then
           ss.l("payment:cache.CF_PAYMENT_ID")
             bid= cache.CF_PAYMENT_ID
          elseif nc_param_cache.CF_PAYMENT_ID~=nil then 
           ss.l("payment:nc_param_cache.CF_PAYMENT_ID")
             bid= nc_param_cache.CF_PAYMENT_ID
          elseif cache.BID~=nil then 
           ss.l("payment:cache.BID")
             bid= cache.BID
           else
           ss.l("payment:bid:not found")
           ss.tprint(cache)
           ss.l("___________________")
           ss.tprint(nc_param_cache)
           ss.l("___________________")

          end
        end



        if iskundestamm then

            add_({name = 'CF_CORRELATION_ID', val = bid});
            add_({name = 'F_RECORD_TYPE', val = "KD02"});
            add_({name = 'F_BATCH_NR', val = ss.def(cache.F_BATCH_NR,"batchNR.e1")});
--ss.l("-----------------------------------------")
--ss.l(debug.traceback())            ss.l("--batch_NR:")            ss.tprint(cache)
--          ss.l("--")           ss.tprint(nc_params)           ss.l("--batch_NR.end")
  
        else
            ss.l("finnovaOut.impl_.send.add CF_PAYMENT_ID")
            add_({name = 'CF_PAYMENT_ID', val = bid});
        end

        add_({name = "MSG_UUID", val = corrid});

        add_({name = "CF_HIT_FLAG", val = ss.def(cache.CF_HIT_FLAG, "0")})
        add_({name = "CF_HIT_DESCRIPTION", val = "todo"})
        if emessage ~= nil then
            add_({name = "CF_RESP_MESSAGE", val = emessage})
        end
        do
            local score_str = "" .. ss.def(score, "")
            if score_str ~= "" then
                add_({name = "CF_RESP_LEX_SCORE", val = score_str})
            end
        end

        ss.l("add transit from cache...")
        add2_(cache, global_oracle_setting.fields.namechecking_transit)
        if iskundestamm then
            add2_(cache, global_oracle_setting.fields.prop_fields_Kundenstamm)
        end
        ss.l("add_nc_params:" .. (#nc_params))
        for i = 1, #nc_params do
            local ii = nc_params[i]
  --          ss.l("add_nc:"..i)
            add_(ii)
        end
--        ss.l("__________after add_nc_params:")        ss.tprint(obj.PROPERTIES)        ss.l("__________")
        if iskundestamm then
            -- todo
        end

        if obj.PROPERTIES["BID"] == nil then
            obj.PROPERTIES["BID"] = cache.BID
        end
        if obj.PROPERTIES.CF_PAYMENT_ID~=nil then 
            obj.PROPERTIES["BID"] = obj.PROPERTIES.CF_PAYMENT_ID
        end

        if obj.PROPERTIES["BID"] == nil then
            ss.l("not found bid:")
            ss.l(debug.traceback())
            ss.tprint(cache)
        end
        if obj.PROPERTIES["BID"] == nil then
            obj.PROPERTIES["BID"] = "test_bid118"
        end

        -- ss.l("_CF_HIT_DESCRIPTION:"..ss.def(message,"nil"))  add_({name="CF_HIT_DESCRIPTION",val=message})

        return obj
    end

    -- simple_send(r,{correlation=task.corrid, expiration=180*20*20, prior=task.prior}
    function writer:simple_send(val, prop)
        ss.l("writer:simple_send")
        local obj = {}
        obj.SYSTEM = {}
        obj.PROPERTIES = val
        ss.l("send_simple.lua:send_simple")
        --  ss.l(debug.traceback())
        --  ss.l("send_simple.lua.prop:")
        --  ss.tprint(prop)

        if prop == nil then prop = {} end

        if prop.correlation ~= nil then
            obj.SYSTEM.corrid = prop.correlation
        end

        if prop.expiration ~= nil then
            -- todo         msg:setExpiration(prop.expiration)
        end

        if prop.header ~= nil then
            ss.l("write header")
            prop.header:write_header(obj)
        end

        ss.l("send_simple.prior.before(11)")
        if prop.prior ~= nil then
            ss.l("send_simple.prior.val:" .. prop.prior .. ":" ..
                     type(prop.prior))
            -- todo        msg:setPrior(prop.prior)

        end
        ss.l(
            "----------------------------------------------------------------------")
        -- ss.tprint(obj)
        -- ss.a(1)

        self:write(obj, {})

    end

    function writer:send_to_failed(err, corrid, prior_arg, cache, bid)
        ss.l("writer:send_to_failed.start")
        local error_message = ss.def(err, "")
        local error_message_short = error_message
        if error_message_short:len() > 1000 then
            error_message_short = error_message_short:sub(1, 1000)
        end

        local send_type = "checkPayments"

        if cache == nil then ss.l("serv:send_to_failed:cache is nil") end

        if finn_serv.IsKundenstamm(cache) then send_type = "kundenstamm" end

        local params = {}
        params.jmsheader = self.jheader
        params.cache = cache
        params.bid = bid -- ??
        params.score = nil
        params.hits_flag = "2"
        params.emessage = error_message_short
        params.corrid = corrid
        params.prior = prior_arg
        params.out_xml = nil
        params.queue = self

        --[[
    transit_prop_(jmsheader,cache,global_oracle_setting.fields.namechecking_transit)
    if params.jms_ex_~=nil then
      for i=1,#params.jms_ex_ do
         local v =params.jms_ex_[i]
         jmsheader:set_string_property(v.name,v.val)
      end
    end
--]]

        ss.l("writer:send_to_failed.type:" .. ss.def(send_type, "nil"))
        if send_type == "checkPayments" then
            finnovaOut.checkPayments.sendError(params)
        elseif send_type == "kundenstamm" then
            finnovaOut.kundenstamm.sendError(params)
        end
        ss.l("writer:send_to_failed.commit")
        self:commit()
    end

    function writer:send_to_finnova_out(message, corrid, prior_arg, cache,
                                        body_message, nc_params)
        ss.l("send_to_finnova_out")
        ss.l("message:" .. ss.def(message, "nil"))
--        ss.l("body_message:" .. ss.def(body_message, "nil"))
--        ss.l("cache:" .. ss.toJson(cache))
--        ss.l("nc_params:" .. ss.toJson(nc_params))

        local obj = self:get_obj_(message, corrid, prior_arg, cache,
                                  body_message, nc_params)
        
        self:write(obj, {prior=prior_arg})
    end

    function writer:destroy()
        remove_(self.engine_.writers, self)
        self.impl_:logout()
        self.impl_:close()
    end
    function writer:isDirty() return false end -- stub
    function writer:flush() end -- stub

    return writer
end

-- ---------------------------------------------------------------------------------
local function createReader_proxy_(engine, options)
ss.l("createReader_proxy_.start:"..ss.def(options.name,"<nil>"))

    local reader = {
        engine_ = engine, -- proploader_=finn_serv.CreatePropLoader()
        options = options
        
    }
    reader.msg = createReaderMsg_(reader)

    local broker_option =
        queryname2brokername("reader", engine.options, options)
    broker_option.max_prior = GetMaxPriorCount()

    broker_option.io_obj = CreateIOv2()
    reader.id = GetObjID_()
  ss.l("__________________________________________________________________________.start")

    local ii=0
    while true do 

        reader.impl_ = ss.factory.msbroker.CreateReciver(broker_option)
        local isfailed = reader.impl_:IsInitFailed()
        if isfailed==false then 
           break
        end
       ss.l("reader init false.ii:"..ii)
       if analiz:IsStop() then 
           ss.l("user break")
       end
       ss.l("sleep 5 sec")
       ss.e("sleep sec 5")

      ii=ii+1
      if ii>120 then break end
    end

  ss.l("__________________________________________________________________________.end")
    if reader.impl_.session.islock then
        ss.l("______________________________________1")
        ss.l("lock mode")
        local sesson = reader.impl_.session

        engine.options.sender_type = sesson.sender_type
        engine.options.sender_handle = sesson.sender_handle
  ss.l("__________________________________________________________________________2")
    end

    reader.lock_update = tick.now() + tick.second * 60
    reader.broker_option = broker_option
  ss.l("__________________________________________________________________________3")
    reader.impl_:login()
  ss.l("__________________________________________________________________________.end")
    function reader:setting(cmd, options)
        local l = tick.now()
        if l < self.lock_update then return end
        self.impl_:updateLockDT({}, 10)
        self.lock_update = l + tick.second * 60
    end
    function reader:update_lock() self.impl_:updateLockDT({}, 10) end
    function reader:finish_lock() self.impl_:Lock_finallize({}) end

    function reader:read(options)
        if options == nil then options = {} end

        self:update_lock()
        --   ss.l("reader:read.proxy.start...")
        --   ss.l("reader:name:"..self.options.queue)

        local msg = self.msg
        msg.task = nil

        local objs = nil

        objs, self.engine_.emessage = reader.impl_:recivevals(
                                          {limit = 1, timeout = 10000})
        self:update_lock()

        if objs == nil then return msg end
        if #objs == 0 then return msg end

        ss.l("read.read.proxy:")
        ss.l("task.len:" .. (#objs))

        local code_string = "return " .. objs[1].msg
        local prior = ""..objs[1].prior

        local task, emessage = load(code_string)
        emessage = ss.def(emessage, "")
        if emessage ~= "" then
            ss.l("failed task:" .. code_string)
            ss.l("failed parse task:" .. emessage)
        end

        msg.task = task()
        msg.prior = prior
        self.engine_.emessage = emessage

        ss.l("reader:read.parse:" .. ss.def(self.engine_.emessage, "ok"))
        --  ss.l("msg.task:") ss.tprint(msg.task)
        --  ss.l("-------------------")

        return msg
    end

    function reader:commit() self.engine_:commit() end

    function reader:destroy()
        self:finish_lock()
        remove_(self.engine_.readers, self)
        self.impl_:logout()
        self.impl_:close()
    end

    ss.l("createReader.end")
    return reader
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()

    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

local function createReader_std_(engine, options)
    ss.l("createReader_std_.create")
    ss.tprint(options)

    local reader = createReader_proxy_(engine, options)
    reader.msg_cache = {}

    function reader:read_v2(options) return obj_get_(self:read(options)) end

    function reader:readImpl_old(options)
ss.l("reader:readImpl")
        local msg = self.msg
        msg:clear()
        -- ss.tprint(options) ss.a(1)
        local objs = nil
        --   ss.l("reader:read.impl(obj)...")
        self:update_lock()
        objs, self.engine_.emessage = reader.impl_:reciveobj(
                                          {
                limit = ss.def(options.limit, 1),
                timeout = ss.def(options.timeout, 10000)
            })
        self:update_lock()
 

        --   ss.l("reader:read.impl.end")
        if objs == nil then ss.l("objs:nil") return msg end
        if #objs == 0 then    return msg end
ss.l("reader:readImpl:read(ok)")
        msg.task  = objs[1].msg
        msg.prior = objs[1].prior
        msg.objs = objs
        --   ss.l("reader:read.task:")   ss.tprint(msg.task)   ss.l("reader:read.end...")
        return msg

    end


    function reader:readImpl__(options)

        local msg = self.msg
        msg:clear()
        -- ss.tprint(options) ss.a(1)
        local objs = nil
        --   ss.l("reader:read.impl(obj)...")
        self:update_lock()
        objs, self.engine_.emessage = reader.impl_:reciveobj(
                                          {
                limit = ss.def(options.limit, 1),
                timeout = ss.def(options.timeout, 10000)
            })
        self:update_lock()

        return objs,msg,self.engine_.emessage

    end


    function reader:readImpl(options)

        local objs,msg = self:readImpl__(options)

       if ss.def(self.engine_.emessage,"")~="" then 
        ss.l("detect problems")
          for i=1,RECONNECT_COUNT_ do 
              ss.l("failed:"..i.."reader:readImpl:"..self.engine_.emessage)
              ss.l("sleep "..i.." sec")
              ss.l(debug.traceback())
              ss.e("sleep sec "..i)
              reader.impl_:reconnect({force=i>3})
              objs,msg = self:readImpl__(options)
              if ss.def(self.engine_.emessage,"")=="" then break end
          end
        end

        if ss.def(self.engine_.emessage,"") ~="" then return msg,ss.def(self.engine_.emessage,"") end
        --   ss.l("reader:read.impl.end")
        if objs == nil then ss.l("objs=NUL") return msg,ss.def(self.engine_.emessage,"")  end
        if #objs == 0 then  return msg,"" end
ss.l("reader:readImpl:read(ok)")
        msg.task  = objs[1].msg
        msg.prior = objs[1].prior
        msg.objs = objs
        --   ss.l("reader:read.task:")   ss.tprint(msg.task)   ss.l("reader:read.end...")
        return msg,""

    end

    function reader:readImpl_corrid(options)
        local msg = self.msg

        local function load_from_cache_(op)
            msg:clear()
            local corrid = op.corrid
            ss.l("wait corrid:" .. corrid)
            local msg_c = self.msg_cache[corrid]
            if msg_c ~= nil then
                self.msg_cache[corrid] = nil
                msg.task = msg_c.msg
                msg.task.prior = msg_c.prior
                return msg
            end
            return msg
        end

        local function load_from_queue_(options)
            local m,emessage = self:readImpl({
                limit = 30,
                timeout = ss.def(options.timeout, 5000)
            })

--           ss.l("z1.emessage:"..ss.def(emessage,"<nil>"))
            if ss.def(emessage,"")~="" then return nil, emessage end
            if m:isnil() then return end
            ss.l("read.add to msg.cache:" .. (#m.objs))

            for i = 1, #m.objs do
                m.task = m.objs[i].msg
                local corrid = m:getCorrelation()
                ss.l("add to cache corrid:" .. corrid)
                self.msg_cache[corrid] = m.objs[i]
            end
            m:clear()
        end

        local m = load_from_cache_(options)
        if m:isnil() == false then return m end

        local xxx,emessage =load_from_queue_(options)  

        if ss.def(emessage,"")~="" then
             return nil,emessage
         end

        return load_from_cache_(options)
    end

    function reader:read(options)
        --   ss.l("reader:read(obj).start...")
        --   ss.l("reader:name:"..self.options.queue)
        if options == nil then options = {} end
        if options.timeout ~= nil then
            options.timeout = options.timeout * 1000
        end -- sec->mssec
        if options.corrid ~= nil then
            return self:readImpl_corrid(options)
        end
        return self:readImpl(options)
    end
    return reader
end

-- ----------------------------------------------------------
-- ----------------------------------------------------------
-- ----------------------------------------------------------
local is_init_msbroker_ = false
function CreateMsBrokerEngine(options)
    if options == nil then options = {} end
    local engine = {
        readers = {},
        writers = {},
        options = options,
        ttype = "msbroker"
    }
    if not (is_init_msbroker_) then
        --     oracle_engine:Initialize()
        ss.l("init msbroker")
        is_init_msbroker_ = true
    end

    engine.serv_ = createServEngine_(engine)

    function engine:serv() return self.serv_ end

    function engine:init() end

    function engine:rollback()
--[[
      if self.rollback==nil then 
      ss.l("engine:rollback.skip") 
      end
      ss.l("engine:rollback:"..self.rollback.info) 
      self.rollback.writer:write()  a
--]]
    end

    function engine:commit() end

    function engine:createReader(options)
       ss.l("engine:createReader.start:"..ss.def(options.queue,"<nil>"))
        local reader = nil
        if options.mode == "proxy" then
            reader = createReader_proxy_(self, options)
        else
            reader = createReader_std_(self, options)
        end
        self.readers[#self.readers + 1] = reader
        return reader
    end

    function engine:createWriter(options)
       ss.l("engine:createWriter")
--        ss.l(debug.traceback())
        ss.tprint(options)
        if options.queue ~= "nc_input" then
            --    ss.a(1)
        end
        local writer = createWriter_(self, options)
        self.writers[#self.writers + 1] = writer
        return writer
    end

    function engine:createHeaderForJmsQueue()
        ss.l("msbroker.engine:createHeaderForJmsQueue")
        --    return finn_serv.CreateHeaderForJmsQueue(self.con)
        return CreateSimpleJHeader_()
    end

    function engine:destroy()
        ss.l("msbroker destroy.start")

        local function destroy_(lists)
           ss.l("item.count:"..#lists)
            local v = ss.deepcopy(lists)
            for i = 1, #v do
                local vv = v[i]
                if vv ~= nil and vv.destroy ~= nil then
                   ss.l("delete id:"..ss.def(vv.id,"<null>"))
                    vv:destroy()
                else
                    ss.l(
                        "oracle.engine:destroy.failed.not found destroy for obj[" ..
                            i .. "]")
                    ss.tprint(vv)

                end
            end
        end
        ss.l("msbroker destroy.readers...")
        destroy_(self.readers)
        ss.l("msbroker destroy.writers...")
        destroy_(self.writers)
        ss.l("broker  destroy.engine:free...")
        --    self.engine:free()
    end

    return engine
end



