---------------------------------
--! @file
--! @brief 内部消息API封装
---------------------------------

siganl = require 'posix.signal'
syswait = require 'posix.sys.wait'
unistd = require 'posix.unistd'
mqtt=require "mosquitto"
local ipc_client=require("ipc_client")
nodes={}
internal_api={}

CONFIG_PATH = "/app/config/"
BROKER_CONFIG_FILE = CONFIG_PATH .. "center_broker.json"

mqtt.init()

--! @brief 模块功能：拆分内部topic
--! @param topic: 接收到数据topic
--! @remarks
--!     @code
--!     local result,gw_sn,dev_sn,port,identifier=internal_api.topic_spilt(topic)
--!     @endcode
--! @return
--!     @retval result: true表示成功解析;false表示没有读到符合格式数据
--!     @retval gw_sn: 表示网关SN
--!     @retval dev_sn: 表示设备SN
--!     @retval port: 设备port
--!     @retval identifier: 服务标识
function internal_api.topic_spilt(topic)
    local gw_sn,dev_sn,port,identifier,param,mi,time
    local s = string.split(topic, '/')
    if s[1] == 'ipc' then
        gw_sn = s[2]
        port = s[3]
        if s[4] == "device" then
            dev_sn = s[5]
            identifier = s[8]
            return true,gw_sn,dev_sn,port,identifier
        else
            return nil
        end
    end
end

--! @brief 模块功能：将payload中的实时数据结构提取出来,tag_node,data自动区分
--! @param topic: 接收到数据topic
--! @param payload: 接收到数据payload
--! @remarks
--!     @code
--!     local tags,identifier=internal_api.payload_spilt(topic,payload)
--!     @endcode
--! @return
--!     @retval tags: 捕获到的服务内容
--!     @retval identifier: 服务id
--!     @retval mi: 消息流水号
--!     @retval time: 时间
function internal_api.payload_spilt(topic,payload)
    local tags,identifier
    local obj = cjson.decode(payload)
    local s = string.split(topic, '/')
    if s[1] == 'ipc' and obj then
        mi = obj.mi
        time = obj.time
        identifier = obj.identifier
        if s[6] == "data_filtered" then
            tags = obj["tags"]
        elseif s[6] == "data" and obj["tag_node"]then
            tags =cjson.decode(obj["tag_node"])
        end
    end
    return tags,identifier,mi,time
end


function internal_api.write_response(gw_sn,dev_sn,port,identifier,param,mi,time)
    --ipc/21881A000001/CANBUS_0/device/21881A00000102/data/Rsp_Rglt/RackThresholdQuery
    local topic = string.format( "ipc/%s/%s/device/%s/data/Rsp_Rglt/%s",gw_sn,port,dev_sn,identifier)
    local obj_send={}
    local json_str
    obj_send['mi'] = mi or 0
    -- obj_send['len'] = #data
    obj_send['identifier'] = identifier
    obj_send['sn'] = dev_sn
    obj_send['time'] =  time or os.time()
    obj_send["tags"] = param and type(param) == "table" and param or cjson.decode(param)
    json_str = cjson.encode(obj_send)
    -- log.info("service_response",topic,json_str)
    client:publish(topic, json_str)
end

--! @brief 模块功能：内部服务上报,或者服务请求的应答,用于IOT侧上行数据
--! @param gw_sn: 网关SN
--! @param dev_sn: 表示设备SN
--! @param port: 设备port
--! @param identifier: 服务标识
--! @param param: 数据的具体内容,类型是table
--! @param mi: 消息流水号,默认为0
--! @param time: 采集时间,默认是当前时间
function internal_api.service_response(gw_sn,dev_sn,port,identifier,param,mi,time)

    local topic = string.format( "ipc/%s/%s/device/%s/data_filtered/service/%s",gw_sn,port,dev_sn,identifier)
    local obj_send={}
    local json_str
    obj_send['mi'] = mi or 0
    -- obj_send['len'] = #data
    obj_send['identifier'] = identifier
    obj_send['sn'] = dev_sn
    obj_send['time'] =  time or os.time()
    obj_send["tags"] = param and type(param) == "table" and param or cjson.decode(param)
    json_str = cjson.encode(obj_send)
    -- log.info("service_response",topic,json_str)
    client:publish(topic, json_str)
end

--! @brief 模块功能：内部事件上报,用于IOT侧上行数据
--! @param gw_sn: 网关SN
--! @param dev_sn: 表示设备SN
--! @param port: 设备port
--! @param identifier: 服务标识
--! @param param: 数据的具体内容,类型是table
--! @param mi: 消息流水号,默认为0
--! @param time: 采集时间,默认是当前时间
function internal_api.event(gw_sn,dev_sn,port,identifier,param,mi,time)
    local topic = string.format( "ipc/%s/%s/device/%s/data/event/%s",gw_sn,port,dev_sn,identifier)
    local obj_send={}
    local json_str
    obj_send['mi'] = mi or 0
    -- obj_send['len'] = #data
    obj_send['identifier'] = identifier
    obj_send['sn'] = dev_sn
    obj_send['time'] =  time or os.time()
    obj_send["data_type"] = 1
    obj_send["tag_node"] = type(param) == "string" and param or cjson.encode(param)
    json_str = cjson.encode(obj_send)
    -- log.info("event",topic,json_str)
    client:publish(topic, json_str)
end

--! @brief 模块功能：用于调用访问设备上的服务,一般用于逻辑联动
--! @param gw_sn: 网关SN
--! @param dev_sn: 表示设备SN
--! @param port: 设备port
--! @param identifier: 服务标识
--! @param param: 数据的具体内容,类型是table
--! @param mi: 消息流水号,默认为0
function internal_api.call_dev_service(gw_sn,dev_sn,port,identifier,param,mi)
    local topic = string.format( "ipc/%s/%s/device/%s/data/Set_Rglt",gw_sn,port,dev_sn)
    local obj_send={}
    local json_str
    obj_send['mi'] = mi or 0
    -- obj_send['len'] = #data
    obj_send['identifier'] = identifier
    obj_send['sn'] = dev_sn
    obj_send['time'] =  os.time()
    obj_send['requester'] = 'local'
    if param then
        for k,v in pairs(param)do
            obj_send[k] =  v
        end
    end
    json_str = cjson.encode(obj_send)
    -- log.info("call_dev_service",topic,json_str)
    client:publish(topic, json_str)
end

--! @brief 模块功能：用于解析(decode)后数据推送到property(Modbus-tcp-slave)
--! @param gw_sn: 网关SN
--! @param dev_sn: 表示设备SN
--! @param port: 设备port
--! @param identifier: 服务标识
--! @param param: 数据的具体内容,类型是table
--! @param mi: 消息流水号,默认为0
function internal_api.pp2property(gw_sn,dev_sn,port,identifier,param, mi)
    if not gw_sn or not port or not dev_sn then
        log.error("pp2north", "param error",gw_sn,dev_sn,port,identifier)
        return
    end
    local topic = string.format( "ipc/%s/%s/device/%s/data/property/%s",gw_sn,port,dev_sn,identifier)
    local obj_send={}
    obj_send['mi'] = mi or 0
    -- obj_send['len'] = #data
    obj_send['identifier'] = identifier
    obj_send['sn'] = dev_sn
    obj_send['port'] = port
    obj_send['time'] =  os.time()
    obj_send['disable_report'] =  0
    obj_send['report_period'] =  0
    obj_send['data_type'] =  2
    obj_send['userParam'] = ""
    if param and type(param) == "table" then
        obj_send['tag_node']=cjson.encode(param)
    end
    json_str = cjson.encode(obj_send)
    -- log.info("pp2north",topic,json_str)
    -- mqtt_client:publish(topic, json_str)
    client:publish(topic, json_str)
end


--! @brief 模块功能：用于解析(decode)后数据推送到IOT平台
--! @param gw_sn: 网关SN
--! @param dev_sn: 表示设备SN
--! @param port: 设备port
--! @param identifier: 服务标识
--! @param param: 数据的具体内容,类型是table
--! @param mi: 消息流水号,默认为0
function internal_api.pp2north(gw_sn,dev_sn,port,identifier,param, mi)
    if not gw_sn or not port or not dev_sn then
        log.error("pp2north", "param error",gw_sn,dev_sn,port,identifier)
        return
    end
    local topic = string.format( "ipc/%s/%s/device/%s/data/service/%s",gw_sn,port,dev_sn,identifier)
    local obj_send={}
    obj_send['mi'] = mi or 0
    -- obj_send['len'] = #data
    obj_send['identifier'] = identifier
    obj_send['sn'] = dev_sn
    obj_send['port'] = port
    obj_send['time'] =  os.time()
    obj_send['disable_report'] =  0
    obj_send['report_period'] =  0
    obj_send['data_type'] =  2
    obj_send['userParam'] = ""
    if param and type(param) == "table" then
        obj_send['tag_node']=cjson.encode(param)
    end
    json_str = cjson.encode(obj_send)
    -- log.info("pp2north",topic,json_str)
    -- mqtt_client:publish(topic, json_str)
    client:publish(topic, json_str)
end

--! @brief 模块功能：用于将编码为二进制的数据(encode)之后的数据发送到指定端口
--! @param gw_sn: 网关SN
--! @param dev_sn: 表示设备SN
--! @param port: 设备port
--! @param identifier: 服务标识
--! @param data: hex数据内容
--! @param mi: 消息流水号,默认为0
function internal_api.pp2south(gw_sn,dev_sn,port,identifier,data,period,mi)
    if not gw_sn or not port or not dev_sn then
        log.error("pp2south", "param error",gw_sn,dev_sn,port,identifier,data)
        return
    end
    local topic = string.format( "ipc/%s/%s/device/%s/data/Set_Rglt_Raw",gw_sn,port,dev_sn)
    local obj_send={}
    obj_send['mi'] = mi or 0
    obj_send['len'] = #data
    obj_send['src_identifier'] = identifier
    obj_send['sn'] = dev_sn
    obj_send['port'] = port
    obj_send['protocol'] = 3
    obj_send['communication_timeout'] = 200
    obj_send['period'] = period or 0
    obj_send['term_addr'] = SUB_ADDR
    obj_send['data_b64'] = base64.encode(data)
    json_str = cjson.encode(obj_send)
    -- log.info("pp2south",topic,json_str)
    -- mqtt_client:publish(topic, json_str)
    client:publish(topic, json_str)
end

--! @brief 模块功能：解析nodes配置,转为结构体
--! @param nodes_cfg: 设备链表配置,string json格式
--! @return
--!     @retval nodes: 设备结构体
function internal_api.load_nodes(nodes_cfg)
    local obj = cjson.decode(nodes_cfg)
    -- if obj and obj.nodes_cfg then
    --     nodes = obj.nodes_cfg
    --     if next(nodes) == nil then
    --         return -2,"not any node"
    --     end
    if obj and #obj then
        nodes = obj
    end
    return nodes
end

--! @brief 模块功能：通过SN获取设备地址
--! @param sn: 设备SN
--! @return
--!     @retval term_addr: 设备地址,string类型
function internal_api.search_addr(sn)
    for _,v in pairs(nodes) do
        if v.sn == sn then
            return v.term_addr
        end
    end
end

--! @brief 模块功能：通过SN获取设备port
--! @param sn: 设备SN
--! @return
--!     @retval port: 设备port,string类型
function internal_api.search_port(sn)
    for _,v in pairs(nodes) do
        if v.sn == sn then
            return v.connect_port
        end
    end
end


--! @brief 模块功能：创建MQTT连接句柄,自动维护连接,采用回调函数方式接收消息,相关接口mqtt_send
--! @param addr: broker address
--! @param port: broker port ,默认1883
--! @param usr: broker usr
--! @param pwd: broker pwd
--! @param subscribe: 订阅topic的数组,数组里面是字符串
--! @param cb: 收到的mqtt消息,回调函数
--! @remarks
--!     @code
--!     client = internal_api.mqtt_session("localhost",1883,nil,nil,{"topic1","topic2"})
--!     local function callback(topic,payload) print(topic,payload) end
--!     client = internal_api.mqtt_session("localhost",1883,nil,nil,{"topic1","topic2"},callback)
--!     @endcode
--! @return
--!     @retval client id
function internal_api.mqtt_session(addr, port, usr, pwd, subscribe, cb,connect_cb,disconnect_cb)
    client = mqtt.new()
    others_topic = {}

    for _,v in pairs(subscribe) do
        if type(v) == "string" then
            table.insert( others_topic,v)
        end
    end

    client.ON_CONNECT = function()
        log.warn("mqtt","client connected")
        for _,v in pairs(others_topic) do
            if type(v) == "string" then client:subscribe(v) end
        end
        if connect_cb then 
            connect_cb()
        end
    end
    client.ON_MESSAGE = function(mid, topic, payload)
        cb(topic,payload)
    end

    client.ON_DISCONNECT = function()
        log.warn("mqtt","client disconnected")
        if disconnect_cb then 
            disconnect_cb()
        end
    end
    if usr and pwd then
        client:login_set(usr,pwd)
    end
    client:connect(addr,port)
    table.insert(sys.poll_socket,function()  msg={} msg.id = rtos.MSG_SOCK_RECV_IND client:loop(50,1) return msg end)
    sys.taskInit(function()
        while(true) do
            while client:socket() do sys.wait(2000) end
            sys.wait(1000)
            client:reconnect()
        end
    end)
    return client
end

--! @brief 模块功能：创建MQTT连接句柄,自动维护连接,采用回调函数方式接收消息,可以使用此接口创建多个client
--! @param addr: broker address
--! @param port: broker port ,默认1883
--! @param usr: broker usr
--! @param pwd: broker pwd
--! @param subscribe: 订阅topic的数组,数组里面是字符串
--! @param cb: 收到的mqtt消息,回调函数
--! @remarks
--!     @code
--!     client = internal_api.create("localhost",1883,nil,nil,{"topic1","topic2"})
--!     local function callback(topic,payload) print(topic,payload) end
--!     client = internal_api.create("localhost",1883,nil,nil,{"topic1","topic2"},callback)
--!     @endcode
--! @return
--!     @retval session client id
function internal_api.create(addr, port, usr, pwd, subscribe, cb,connect_cb,disconnect_cb)
    return ipc_client(addr, port, usr, pwd, subscribe, cb,connect_cb,disconnect_cb)
end

--! @brief 模块功能：创建内部MQTT连接句柄,自动维护连接,接收函数自动触发上行回调函数,相关的接口pp2south;pp2north;call_dev_service;service_response;event
--! @param devs: 需要关注的设备SN,table类型
--! @param purpose: 关心消息用途,可选,"PP"or"APP" "pp"or"app".PP标识仅仅做协议解析使用,关注协议解析数据;PP标识仅仅做APP设备编译计算,逻辑处理.
--! @param north_cb: 回调函数.当purpose="PP",当收到iot方向消息时触发,north中处理protocol_encode(); 当purpose="APP",当收到iot方向消息时触发,north中处理平台处理服务中定义的事物;
--! @param south_cb: 回调函数.当purpose="APP",当收到port口模块消息时触发,south中处理protocol_decode(); 当purpose="APP",当收到协议解析后服务/属性/事件,south中做数据加工;
--! @param subscribe: (可选项)需要订阅的其他topic,table类型
--! @param others_cb: (可选项)需要订阅的其他topic,回调函数
--! @remarks
--!     @code
--!     local subscribe={"ipc/+/+/device/+/data/raw_data","ipc/+/+/device/+/data/Set_Rglt"}
--!     local subscribe={"ipc/+/+/device/+/data_filtered/property/post","ipc/+/+/device/+/data/service/+","ipc/+/VIRTUAL_1/device/+/data/Set_Rglt"}
--!     local function iot2pp(topic,payload) print("iot2pp",topic,payload) end
--!     local function port2pp(topic,payload) print("port2pp",topic,payload) end
--!     local client = internal_api.local_mqtt_session({"sn1","sn2"},"PP",iot2pp,port2pp)
--!     local function iot2app(topic,payload) print("iot2app",topic,payload) end
--!     local function pp2app(topic,payload) print("pp2app",topic,payload) end
--!     local function others(topic,payload) print("others",topic,payload) end
--!     local client = internal_api.local_mqtt_session({"sn1","sn2"},"APP",iot2app,pp2app,{"sys/upgrade","sys/status"},others)
--!     @endcode
--! @return
--!     @retval client id
function internal_api.local_mqtt_session(devs,purpose,north_cb,south_cb, subscribe,others_cb,connect_cb,disconnect_cb)
    local addr, port, usr, pwd = "localhost",1883
    client = mqtt.new()
    purpose_g = purpose
    others_topic,north_topic,south_topic={},{},{}
    for _,v in pairs(devs) do
        if type(v) == "string" then
            if purpose_g and string.upper(purpose_g) == "PP" then
                local north = string.format( "ipc/+/+/device/%s/data/raw_data",v)
                table.insert( north_topic,north)
                local south = string.format( "ipc/+/+/device/%s/data/Set_Rglt",v)
                table.insert( south_topic,south)
            elseif purpose_g and string.upper(purpose_g)  == "APP" then
                local north = string.format( "ipc/+/+/device/%s/data/Set_Rglt",v)  --关心虚拟设备的控制消息
                table.insert( north_topic,north)
                -- local south = string.format( "ipc/+/+/device/%s/data_filtered/property/post",v) --汇聚后数据
                -- table.insert( south_topic,south)
                local south = string.format( "ipc/+/+/device/%s/data/service/+",v)  --服务调用数据
                table.insert( south_topic,south)
                local t = string.format( "ipc/+/+/device/%s/data/event/+",v)  --服务调用数据
                table.insert( south_topic,t)
            end
        end
    end
    if subscribe then
        for _,v in pairs(subscribe) do
            if type(v) == "string" then
                table.insert( others_topic,v)
            end
        end
    end
    client.ON_CONNECT = function()
        log.warn("mqtt","client connected")
        for _,v in pairs(others_topic) do
            if type(v) == "string" then
                client:subscribe(v)
            end
        end
        for _,v in pairs(north_topic) do
            if type(v) == "string" then
                client:subscribe(v)
            end
        end
        for _,v in pairs(south_topic) do
            if type(v) == "string" then
                client:subscribe(v)
            end
        end
        if connect_cb then 
            connect_cb()
        end
    end
    client.ON_MESSAGE = function(mid,topic,payload)
        if topic and string.find(topic,"/data/raw_data", 25) or string.find(topic,"/data_filtered/property/post", 25) or string.find(topic,"/data/service/", 25) or string.find(topic,"/data/event/", 25) then   --如果是原始数据一定是南向数据
            if south_cb then south_cb(topic,payload) end
            return
        elseif topic and string.find(topic,"/data/Set_Rglt", 25) then
            if north_cb then north_cb(topic,payload) end
            return
        end
        if others_cb then others_cb(topic,payload) end
    end

    client.ON_DISCONNECT = function()
        log.warn("mqtt","client disconnected")
        if disconnect_cb then 
            disconnect_cb()
        end
    end

    client:connect(addr,port)
    table.insert(sys.poll_socket,function()  msg={} msg.id = rtos.MSG_SOCK_RECV_IND client:loop(50,1) return msg end)
    sys.taskInit(function()
        while(true) do
            while client:socket() do sys.wait(2000) end
            sys.wait(1000)
            client:reconnect()
        end
    end)
    return client
end


--! @brief 模块功能：根据配置文件/app/config/center_broker.json获取相应的host/port/usr/pass,若配置为空，连接localhost，
--！创建MQTT连接句柄,自动维护连接,采用回调函数方式接收消息,相关接口mqtt_send创建内部MQTT连接句柄,自动维护连接,接收函数自动触发上行回调函数,相关的接口pp2south;pp2north;call_dev_service;servic--！e_response;event
--! @param devs: 需要关注的设备SN,table类型
--! @param purpose: 关心消息用途,可选,"PP"or"APP" "pp"or"app".PP标识仅仅做协议解析使用,关注协议解析数据;PP标识仅仅做APP设备编译计算,逻辑处理.
--! @param north_cb: 回调函数.当purpose="PP",当收到iot方向消息时触发,north中处理protocol_encode(); 当purpose="APP",当收到iot方向消息时触发,north中处理平台处理服务中定义的事物;
--! @param south_cb: 回调函数.当purpose="APP",当收到port口模块消息时触发,south中处理protocol_decode(); 当purpose="APP",当收到协议解析后服务/属性/事件,south中做数据加工;
--! @param subscribe: (可选项)需要订阅的其他topic,table类型
--! @param others_cb: (可选项)需要订阅的其他topic,回调函数
--! @remarks
--!     @code
--!     local subscribe={"ipc/+/+/device/+/data/raw_data","ipc/+/+/device/+/data/Set_Rglt"}
--!     local subscribe={"ipc/+/+/device/+/data_filtered/property/post","ipc/+/+/device/+/data/service/+","ipc/+/VIRTUAL_1/device/+/data/Set_Rglt"}
--!     local function iot2pp(topic,payload) print("iot2pp",topic,payload) end
--!     local function port2pp(topic,payload) print("port2pp",topic,payload) end
--!     local client = internal_api.local_mqtt_session({"sn1","sn2"},"PP",iot2pp,port2pp)
--!     local function iot2app(topic,payload) print("iot2app",topic,payload) end
--!     local function pp2app(topic,payload) print("pp2app",topic,payload) end
--!     local function others(topic,payload) print("others",topic,payload) end
--!     local client = internal_api.local_mqtt_session({"sn1","sn2"},"APP",iot2app,pp2app,{"sys/upgrade","sys/status"},others)
--!     @endcode
--! @return
--!     @retval client id
function internal_api.auto_mqtt_session(devs,purpose,north_cb,south_cb, subscribe,others_cb,connect_cb,disconnect_cb)
	-- read broker from /app/config/center_broker.json,若配置文件存在，且配置中addr和port均存在，否则走localhost
	local addr,port,usr,pwd = "localhost",1883
	if io.exists(BROKER_CONFIG_FILE) then
        local str = io.readFile(BROKER_CONFIG_FILE)
        if str and #str > 0 then
            obj = cjson.decode(str or '')
            addr,port,usr,pwd = obj.host, obj.port, obj.user, obj.pass
			if not addr or not port then
				addr, port, usr, pwd = "localhost",1883
			end
		else
			addr, port, usr, pwd = "localhost",1883
        end
	else
		addr, port, usr, pwd = "localhost",1883
    end
	gw_sn = io.getGatewayID()
    
    --log.info("auto_mqtt_session",  string.format("\naddr='%s'\nport='%s'\nusr='%s'\npwd='%s'",addr or "",port or "",usr or "",pwd or ""))
	
    client = mqtt.new()
    purpose_g = purpose
    others_topic,north_topic,south_topic={},{},{}
    for _,v in pairs(devs) do
        if type(v) == "string" then
            if purpose_g and string.upper(purpose_g) == "PP" then
                local north = string.format( "ipc/+/+/device/%s/data/raw_data",v)
                table.insert( north_topic,north)
                local south = string.format( "ipc/+/+/device/%s/data/Set_Rglt",v)
                table.insert( south_topic,south)
            elseif purpose_g and string.upper(purpose_g)  == "APP" then
                local north = string.format( "ipc/+/+/device/%s/data/Set_Rglt",v)  --关心虚拟设备的控制消息
                table.insert( north_topic,north)
                -- local south = string.format( "ipc/+/+/device/%s/data_filtered/property/post",v) --汇聚后数据
                -- table.insert( south_topic,south)
                local south = string.format( "ipc/+/+/device/%s/data/service/+",v)  --服务调用数据
                table.insert( south_topic,south)
                local t = string.format( "ipc/+/+/device/%s/data/event/+",v)  --服务调用数据
                table.insert( south_topic,t)
            end
        end
    end
    if subscribe then
        for _,v in pairs(subscribe) do
            if type(v) == "string" then
                table.insert( others_topic,v)
            end
        end
    end
    client.ON_CONNECT = function()
        log.warn("mqtt","client connected")
        for _,v in pairs(others_topic) do
            if type(v) == "string" then
                client:subscribe(v)
            end
        end
        for _,v in pairs(north_topic) do
            if type(v) == "string" then
                client:subscribe(v)
            end
        end
        for _,v in pairs(south_topic) do
            if type(v) == "string" then
                client:subscribe(v)
            end
        end
        if connect_cb then 
            connect_cb()
        end
    end
    client.ON_MESSAGE = function(mid,topic,payload)
        if topic and string.find(topic,"/data/raw_data", 25) or string.find(topic,"/data_filtered/property/post", 25) or string.find(topic,"/data/service/", 25) or string.find(topic,"/data/event/", 25) then   --如果是原始数据一定是南向数据
            if south_cb then south_cb(topic,payload) end
            return
        elseif topic and string.find(topic,"/data/Set_Rglt", 25) then
            if north_cb then north_cb(topic,payload) end
            return
        end
        if others_cb then others_cb(topic,payload) end
    end

    client.ON_DISCONNECT = function()
        log.warn("mqtt","client disconnected")
        if disconnect_cb then 
            disconnect_cb()
        end
    end

    if usr and pwd then
        client:login_set(usr,pwd)
    end
	
    client:connect(addr,port)
    table.insert(sys.poll_socket,function()  msg={} msg.id = rtos.MSG_SOCK_RECV_IND client:loop(50,1) return msg end)
    sys.taskInit(function()
        while(true) do
            while client:socket() do sys.wait(2000) end
            sys.wait(1000)
            client:reconnect()
        end
    end)
    return client
end

--! @brief 模块功能：发送自定义MQTT消息,不局限内部/外部broker
--! @param client: mqtt句柄
--! @param topic: 接收到数据topic
--! @param payload: 接收到数据payload
function internal_api.mqtt_send(client,topic,payload)
    if not client then return end
    local mid,errid,errmsg = client:publish(topic, payload)
    if not mid then log.error("mqtt.send","error:", errid,errmsg) end
end

--! @brief 模块功能：更新订阅的topic
--! @param gw_sn: 网关SN
--! @param sub_nodes: 订阅的子设备数组，格式如下
--!     local sub_nodes = {
--!         {
--!             sn_suffix = "_bmsmeter",-- SN 尾缀
--!             grp_idx = nil,--nil:不分组总数据,1:第一组,2:第二组,
--!             ident_list = {"yougongzongdianliang","sanxiangcv"}, --关心的采集服务
--!         },
--!         {
--!             sn_suffix = "_DI",
--!             grp_idx = 1,--nil:不分组总数据,1:第一组,2:第二组,
--!             ident_list = {"DIRead"}, --关心的采集服务
--!         }
--!     } 
--! @param nodes_cfg: 全脚本入参，关联脚本的设备SN 格式如下
--!     local nodes_cfg{
--!         {
--!     	"channelId": "",
--!     	"connect_port": "VIRTUAL_1",
--!     	"depth": 0,
--!     	"product_key": "5043658676",
--!     	"sn": "21881FFF0005_GUARD",
--!     	"template_id": "5043658676"
--!         }
--!     }
--! @return
--!     @retval subscribe:需要订阅的topic集合
function internal_api.get_subscribes(gw_sn,sub_nodes,nodes_cfg)
    local subscribe = {}
    for i,v in ipairs(nodes_cfg) do
        local topic =  string.format( "ipc/+/+/device/%s/data/Set_Rglt",v.sn)
        table.insert(subscribe,topic)
        log.info("sub topic",topic)
    end
    for i,v in ipairs(sub_nodes) do
        assert(v.sn_suffix)
        assert(v.ident_list)
        for _,ident in ipairs(v.ident_list) do
            local topic =  string.format( "ipc/+/+/device/%s%s/data/service/%s",gw_sn,v.sn_suffix,ident)
            table.insert(subscribe, topic)  --服务调用数据)
            log.info("sub topic",topic)
        end
    end
    return subscribe
end

--! @brief 模块功能：将sub node 格式转为,已SN为下标的索引
--! @param gw_sn: 网关SN
--! @param sub_nodes: 订阅的子设备数组，格式如下
--!     local sub_nodes = {
--!         {
--!             sn_suffix = "_bmsmeter",-- SN 尾缀
--!             grp_idx = nil,--nil:不分组总数据,1:第一组,2:第二组,
--!             ident_list = {"yougongzongdianliang","sanxiangcv"}, --关心的采集服务
--!         },
--!         {
--!             sn_suffix = "_DI",
--!             grp_idx = 1,--nil:不分组总数据,1:第一组,2:第二组,
--!             ident_list = {"DIRead"}, --关心的采集服务
--!         }
--!     } 
--! @return
--!     @retval subscribe:需要订阅的topic集合
function internal_api.update_nodes_info(gw_sn,sub_nodes)
    assert(gw_sn)
    assert(sub_nodes)
    local nodes_info = {}
    for i,v in ipairs(sub_nodes) do
        if v.sn_suffix then
            local dev_sn = string.format("%s%s",gw_sn,v.sn_suffix)
            nodes_info[dev_sn] = v
        end
    end
    return nodes_info
end




return internal_api