--[[
ģƣmqttЭ
ģ鹦ܣʵЭͽĶhttp://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html˽mqttЭ
ģ޸ʱ䣺2017.02.24
]]

--[[
Ŀǰֻ֧QoS=0QoS=1֧QoS=2
]]

module(...,package.seeall)

local lpack = require"pack"
require"common"
require"socket"
require"mqttdup"

local slen,sbyte,ssub,sgsub,schar,srep,smatch,sgmatch = string.len,string.byte,string.sub,string.gsub,string.char,string.rep,string.match,string.gmatch
--
CONNECT,CONNACK,PUBLISH,PUBACK,PUBREC,PUBREL,PUBCOMP,SUBSCRIBE,SUBACK,UNSUBSCRIBE,UNSUBACK,PINGREQ,PINGRSP,DISCONNECT = 1,2,3,4,5,6,7,8,9,10,11,12,13,14
--к
local seq = 1

local function print(...)
	_G.print("mqtt",...)
end

local function encutf8(s)
	if not s then return "" end
	local utf8s = common.gb2312toutf8(s)
	return lpack.pack(">HA",slen(utf8s),utf8s)
end

local function enclen(s)
	if not s or slen(s) == 0 then return schar(0) end
	local ret,len,digit = "",slen(s)
	repeat
		digit = len % 128
		len = (len-digit) / 128
		if len > 0 then
			digit = bit.bor(digit,0x80)
		end
		ret = ret..schar(digit)
	until (len <= 0)
	return ret
end

local function declen(s)
	local i,value,multiplier,digit = 1,0,1
	repeat
		if i > slen(s) then return end
		digit = sbyte(s,i)
		value = value + bit.band(digit,127)*multiplier
		multiplier = multiplier * 128
		i = i + 1
	until (bit.band(digit,128) == 0)
	return true,value,i-1
end

local function getseq()
	local s = seq
	seq = (seq+1)%0xFFFF
	if seq == 0 then seq = 1 end
	return lpack.pack(">H",s)
end

local function iscomplete(s)
	local i,typ,flg,len,cnt
	for i=1,slen(s) do
		typ = bit.band(bit.rshift(sbyte(s,i),4),0x0f)
		--print("typ",typ)
		if typ >= CONNECT and typ <= DISCONNECT then
			flg,len,cnt = declen(ssub(s,i+1,-1))
			--print("f",flg,len,cnt,(slen(ssub(s,i+1,-1))-cnt))
			if flg and cnt <= 4 and len <= (slen(ssub(s,i+1,-1))-cnt) then
				return true,i,i+cnt+len,typ,len
			else
				return
			end
		end
	end
end

--[[
pack
  MQTT
  
		mqttvermqttЭ汾
		typ
		...ɱ
ֵһֵǱݣڶֵÿֱԶĲ
]]
local function pack(mqttver,typ,...)
	local para = {}
	local function connect(alive,id,twill,user,pwd,cleansess)
		local ret = lpack.pack(">bAbbHA",
						CONNECT*16,
						encutf8(mqttver=="3.1.1" and "MQTT" or "MQIsdp"),
						mqttver=="3.1.1" and 4 or 3,
						(user and 1 or 0)*128+(pwd and 1 or 0)*64+twill.retain*32+twill.qos*8+twill.flg*4+(cleansess or 1)*2,
						alive,
						encutf8(id))
		if twill.flg==1 then
			ret = ret..encutf8(twill.topic)..encutf8(twill.payload)
		end
		ret = ret..encutf8(user)..encutf8(pwd)
		return ret
	end

	local function subscribe(p)
		para.dup,para.topic = true,p.topic
		para.seq = p.seq or getseq()
		print("subscribe",p.dup,para.dup,common.binstohexs(para.seq))

		local s = lpack.pack("bA",SUBSCRIBE*16+(p.dup and 1 or 0)*8+2,para.seq)
		for i=1,#p.topic do
			s = s..encutf8(p.topic[i].topic)..schar(p.topic[i].qos or 0)
		end
		return s
	end

	local function publish(p)
		para.dup,para.topic,para.payload,para.qos,para.retain = true,p.topic,p.payload,p.qos,p.retain
		para.seq = p.seq or getseq()
		--print("publish",p.dup,para.dup,common.binstohexs(para.seq))
		local s1 = lpack.pack("bAA",PUBLISH*16+(p.dup and 1 or 0)*8+(p.qos or 0)*2+p.retain or 0,encutf8(p.topic),((p.qos or 0)>0 and para.seq or ""))
		local s2 = s1..p.payload
		return s2
	end

	local function puback(seq)
		return schar(PUBACK*16)..seq
	end

	local function pingreq()
		return schar(PINGREQ*16)
	end

	local function disconnect()
		return schar(DISCONNECT*16)
	end

	local function unsubscribe(p)
		para.dup,para.topic = true,p.topic
		para.seq = p.seq or getseq()
		print("unsubscribe",p.dup,para.dup,common.binstohexs(para.seq))

		local s = lpack.pack("bA",UNSUBSCRIBE*16+(p.dup and 1 or 0)*8+2,para.seq)
		for i=1,#p.topic do
			s = s..encutf8(p.topic[i])
		end
		return s
	end

	local procer =
	{
		[CONNECT] = connect,
		[SUBSCRIBE] = subscribe,
		[PUBLISH] = publish,
		[PUBACK] = puback,
		[PINGREQ] = pingreq,
		[DISCONNECT] = disconnect,
		[UNSUBSCRIBE] = unsubscribe,
	}

	local s = procer[typ](...)
	local s1,s2,s3 = ssub(s,1,1),enclen(ssub(s,2,-1)),ssub(s,2,-1)
	s = s1..s2..s3
	print("pack",typ,(slen(s) > 200) and "" or common.binstohexs(s))
	return s,para
end

local rcvpacket = {}

--[[
unpack
  MQTT
  
		mqttvermqttЭ汾
		sһı
ֵɹһtableݣԪɱ;ʧܣnil
]]
local function unpack(mqttver,s)
	rcvpacket = {}

	local function connack(d)
		print("connack",common.binstohexs(d))
		rcvpacket.suc = (sbyte(d,2)==0)
		rcvpacket.reason = sbyte(d,2)
		return true
	end

	local function suback(d)
		print("suback or unsuback",common.binstohexs(d))
		if slen(d) < 2 then return end
		rcvpacket.seq = ssub(d,1,2)
		return true
	end

	local function puback(d)
		print("puback",common.binstohexs(d))
		if slen(d) < 2 then return end
		rcvpacket.seq = ssub(d,1,2)
		return true
	end

	local function publish(d)
		--print("publish",common.binstohexs(d)) --̫ʱܴ򿪣ڴ治
		if slen(d) < 4 then return end
		local _,tplen = lpack.unpack(ssub(d,1,2),">H")
		local pay = (rcvpacket.qos > 0 and 5 or 3)
		if slen(d) < tplen+pay-1 then return end
		rcvpacket.topic = ssub(d,3,2+tplen)

		if rcvpacket.qos > 0 then
			rcvpacket.seq = ssub(d,tplen+3,tplen+4)
			pay = 5
		end
		rcvpacket.payload = ssub(d,tplen+pay,-1)
		return true
	end

	local function empty()
		return true
	end

	local procer =
	{
		[CONNACK] = connack,
		[SUBACK] = suback,
		[PUBACK] = puback,
		[PUBLISH] = publish,
		[PINGRSP] = empty,
		[UNSUBACK] = suback,
	}
	local d1,d2,d3,typ,len = iscomplete(s)
	if not procer[typ] then print("unpack unknwon typ",typ) return end
	rcvpacket.typ = typ
	rcvpacket.qos = bit.rshift(bit.band(sbyte(s,1),0x06),1)
	rcvpacket.dup = bit.rshift(bit.band(sbyte(s,1),0x08),3)==1
	print("unpack",typ,rcvpacket.qos,(slen(s) > 200) and "" or common.binstohexs(s))
	return procer[typ](ssub(s,slen(s)-len+1,-1)) and rcvpacket or nil
end


--һڵĶӺ̨ʧܣ᳢ΪRECONN_PERIOD룬RECONN_MAX_CNT
--һڶûӳɹȴRECONN_CYCLE_PERIOD·һ
--RECONN_CYCLE_MAX_CNTεڶûӳɹ
local RECONN_MAX_CNT,RECONN_PERIOD,RECONN_CYCLE_MAX_CNT,RECONN_CYCLE_PERIOD = 3,5,3,20

--mqtt clients洢
local tclients = {}

--[[
getclient
  һmqtt clienttclientsе
  
		sckidxmqtt clientӦsocket
ֵsckidxӦmqtt clienttclientsе
]]
local function getclient(sckidx)
	for k,v in pairs(tclients) do
		if v.sckidx==sckidx then return k end
	end
end

--[[
mqttconncb
  MQTT CONNECTĺ첽ص
  
		sckidxsocket idx
		result boolͣͽtrueΪɹΪʧ
		tparatableͣ{key="MQTTCONN",val=CONNECT}
ֵ
]]
function mqttconncb(sckidx,result,tpara)
	--MQTT CONNECTݱʱDUP_TIMEûյCONNACKCONNACKʧܣԶطCONNECT
	--طĴmqttdup.lua
	mqttdup.ins(sckidx,tmqttpack["MQTTCONN"].mqttduptyp,tpara.val)
end

--[[
mqttconndata
  MQTT CONNECT
  
		sckidxsocket idx
ֵCONNECTݺͱĲ
]]
function mqttconndata(sckidx)
	local mqttclientidx = getclient(sckidx)
	return pack(tclients[mqttclientidx].mqttver,
				CONNECT,
				tclients[mqttclientidx].keepalive,
				tclients[mqttclientidx].clientid,
				{
					flg=tclients[mqttclientidx].willflg or 0,
					qos=tclients[mqttclientidx].willqos or 0,
					retain=tclients[mqttclientidx].willretain or 0,
					topic=tclients[mqttclientidx].willtopic or "",
					payload=tclients[mqttclientidx].willpayload or "",
				},
				tclients[mqttclientidx].user,
				tclients[mqttclientidx].password,
				tclients[mqttclientidx].cleansession or 1)
end

--[[
mqttsubcb
  SUBSCRIBEĺ첽ص
  
		sckidxsocket idx
		result boolͣͽtrueΪɹΪʧ
		tparatableͣ{key="MQTTSUB", val=para, usertag=usertag, ackcb=ackcb}
ֵ
]]
local function mqttsubcb(sckidx,result,tpara)
	--·װMQTT SUBSCRIBEģظ־ΪtrueкźtopicԭʼֵݱʱDUP_TIMEûյSUBACKԶطSUBSCRIBE
	--طĴmqttdup.lua
	mqttdup.ins(sckidx,tpara.key,pack(tclients[getclient(sckidx)].mqttver,SUBSCRIBE,tpara.val),tpara.val.seq,tpara.ackcb,tpara.usertag)
end

--[[
mqttunsubcb
  UNSUBSCRIBEĺ첽ص
  
		sckidxsocket idx
		result boolͣͽtrueΪɹΪʧ
		tparatableͣ{key="MQTTUNSUB", val=para, usertag=usertag, ackcb=ackcb}
ֵ
]]
local function mqttunsubcb(sckidx,result,tpara)
	--·װMQTT UNSUBSCRIBEģظ־ΪtrueкźtopicԭʼֵݱʱDUP_TIMEûյUNSUBACKԶطUNSUBSCRIBE
	--طĴmqttdup.lua
	mqttdup.ins(sckidx,tpara.key,pack(tclients[getclient(sckidx)].mqttver,UNSUBSCRIBE,tpara.val),tpara.val.seq,tpara.ackcb,tpara.usertag)
end

--[[
mqttpubcb
  PUBLISHĺ첽ص
  
		sckidxsocket idx
		result boolͣͽtrueΪɹΪʧ
		tparatableͣ{key="MQTTPUB", val=para, qos=qos, usertag=usertag, ackcb=ackcb}
ֵ
]]
local function mqttpubcb(sckidx,result,tpara)
	if tpara.qos==0 then
		if tpara.ackcb then tpara.ackcb(tpara.usertag,result) end
	elseif tpara.qos==1 then
		--·װMQTT PUBLISHģظ־ΪtrueкštopicpayloadԭʼֵݱʱDUP_TIMEûյPUBACKԶطPUBLISH
		--طĴmqttdup.lua
		mqttdup.ins(sckidx,tpara.key,pack(tclients[getclient(sckidx)].mqttver,PUBLISH,tpara.val),tpara.val.seq,tpara.ackcb,tpara.usertag)
	end
end

--[[
mqttdiscb
  MQTT DICONNECTĺ첽ص
  
		sckidxsocket idx
		result boolͣͽtrueΪɹΪʧ
		tparatableͣ{key="MQTTDISC", val=data, usertag=usrtag}
ֵ
]]
function mqttdiscb(sckidx,result,tpara)
	--رsocket
	tclients[getclient(sckidx)].discing = true
	socket.disconnect(sckidx,tpara.usertag)
end

--[[
mqttdiscdata
  MQTT DISCONNECT
  
		sckidxsocket idx
ֵDISCONNECTݺͱĲ
]]
function mqttdiscdata(sckidx)
	return pack(tclients[getclient(sckidx)].mqttver,DISCONNECT)
end

--[[
disconnect
  MQTT DISCONNECT
  
		sckidxsocket idx
		usrtagûԶ
ֵtrueʾ˶nilʾûз
]]
local function disconnect(sckidx,usrtag)
	return mqttsnd(sckidx,"MQTTDISC",usrtag)
end

--[[
mqttpingreqdata
  MQTT PINGREQ
  
		sckidxsocket idx
ֵPINGREQݺͱĲ
]]
function mqttpingreqdata(sckidx)
	return pack(tclients[getclient(sckidx)].mqttver,PINGREQ)
end

--[[
pingreq
  MQTT PINGREQ
  
		sckidxsocket idx
ֵ
]]
local function pingreq(sckidx)
	local mqttclientidx = getclient(sckidx)
	mqttsnd(sckidx,"MQTTPINGREQ")
	if not sys.timer_is_active(disconnect,sckidx) then
		--ʱʱ+30ڣûյpingrspMQTT DISCONNECT
		sys.timer_start(disconnect,(tclients[mqttclientidx].keepalive+30)*1000,sckidx)
	end
end

--[[
snd
  ÷ͽӿڷ
  
		sckidxsocket idx
        data͵ݣڷͽ¼ntfyУḳֵitem.data
		para͵Ĳڷͽ¼ntfyУḳֵitem.para
ֵ÷ͽӿڵĽݷǷɹĽݷǷɹĽntfyеSEND¼֪ͨtrueΪɹΪʧ
]]
function snd(sckidx,data,para)
	return socket.send(sckidx,data,para)
end

--mqttӦñı
tmqttpack =
{
	MQTTCONN = {sndpara="MQTTCONN",mqttyp=CONNECT,mqttduptyp="CONN",mqttdatafnc=mqttconndata,sndcb=mqttconncb},
	MQTTPINGREQ = {sndpara="MQTTPINGREQ",mqttyp=PINGREQ,mqttdatafnc=mqttpingreqdata},
	MQTTDISC = {sndpara="MQTTDISC",mqttyp=DISCONNECT,mqttdatafnc=mqttdiscdata,sndcb=mqttdiscb},
}

local function getidbysndpara(para)
	for k,v in pairs(tmqttpack) do
		if v.sndpara==para then return k end
	end
end

--[[
mqttsnd
  MQTTķܽӿڣݱͣmqttӦñıҵȻ
  
		sckidxsocket idx
        typ
		usrtagûԶ
ֵtrueʾ˶nilʾûз
]]
function mqttsnd(sckidx,typ,usrtag)
	if not tmqttpack[typ] then print("mqttsnd typ error",typ) return end
	local mqttyp = tmqttpack[typ].mqttyp
	local dat,para = tmqttpack[typ].mqttdatafnc(sckidx)

	if mqttyp==CONNECT then
		if tmqttpack[typ].mqttduptyp then mqttdup.rmv(sckidx,tmqttpack[typ].mqttduptyp) end
		if not snd(sckidx,dat,{key=tmqttpack[typ].sndpara,val=dat}) and tmqttpack[typ].sndcb then
			tmqttpack[typ].sndcb(sckidx,false,{key=tmqttpack[typ].sndpara,val=dat})
		end
	elseif mqttyp==PINGREQ then
		snd(sckidx,dat,{key=tmqttpack[typ].sndpara})
	elseif mqttyp==DISCONNECT then
		if not snd(sckidx,dat,{key=tmqttpack[typ].sndpara,usertag=usrtag}) and tmqttpack[typ].sndcb then
			tmqttpack[typ].sndcb(sckidx,false,{key=tmqttpack[typ].sndpara,usertag=usrtag})
		end
	end

	return true
end

--[[
reconn
  socket̨
        һڵĶӺ̨ʧܣ᳢ΪRECONN_PERIOD룬RECONN_MAX_CNT
        һڶûӳɹȴRECONN_CYCLE_PERIOD·һ
        RECONN_CYCLE_MAX_CNTεڶûӳɹ
  
		sckidxsocket idx
ֵ
]]
local function reconn(sckidx)
	local mqttclientidx = getclient(sckidx)
	print("reconn",mqttclientidx,tclients[mqttclientidx].sckreconncnt,tclients[mqttclientidx].sckconning,tclients[mqttclientidx].sckreconncyclecnt)
	--sckconningʾڳӺ̨һҪжϴ˱пܷ𲻱Ҫsckreconncntӣʵʵ
	if tclients[mqttclientidx].sckconning then return end
	--һڵ
	if tclients[mqttclientidx].sckreconncnt < RECONN_MAX_CNT then
		tclients[mqttclientidx].sckreconncnt = tclients[mqttclientidx].sckreconncnt+1
		socket.disconnect(sckidx)
		tclients[mqttclientidx].sckconning = true
	--һڵʧ
	else
		tclients[mqttclientidx].sckreconncnt,tclients[mqttclientidx].sckreconncyclecnt = 0,tclients[mqttclientidx].sckreconncyclecnt+1
		if tclients[mqttclientidx].sckreconncyclecnt >= RECONN_CYCLE_MAX_CNT then
			if tclients[mqttclientidx].sckerrcb then
				tclients[mqttclientidx].sckreconncnt=0
				tclients[mqttclientidx].sckreconncyclecnt=0
				tclients[mqttclientidx].sckerrcb("CONNECT")
			else
				sys.restart("connect fail")
			end
		else
			link.shut()
		end
	end
end

local function connectitem(mqttclientidx)
	connect(tclients[mqttclientidx].sckidx,tclients[mqttclientidx].prot,tclients[mqttclientidx].host,tclients[mqttclientidx].port)
end

--[[
ntfy
  socket״̬Ĵ
  
        idxnumberͣsocketάsocket idxsocket.connectʱĵһͬԺԲ
        evtstringͣϢ¼
		result boolͣϢ¼trueΪɹΪʧ
		itemtableͣ{data=,para=}ϢشĲݣĿǰֻSEND͵¼õ˴˲socket.sendʱĵ2͵3ֱΪdatparitem={data=dat,para=par}
ֵ
]]
function ntfy(idx,evt,result,item)
	local mqttclientidx = getclient(idx)
	print("ntfy",evt,result,item)
	--ӽsocket.connect첽¼
	if evt == "CONNECT" then
		tclients[mqttclientidx].sckconning = false
		--ӳɹ
		if result then
			tclients[mqttclientidx].sckconnected=true
			tclients[mqttclientidx].sckreconncnt=0
			tclients[mqttclientidx].sckreconncyclecnt=0
			tclients[mqttclientidx].sckrcvs=""
			--ֹͣʱ
			sys.timer_stop(reconn,idx)
			--mqtt connect
			mqttsnd(idx,"MQTTCONN")
		--ʧ
		else
			--RECONN_PERIOD
			sys.timer_start(reconn,RECONN_PERIOD*1000,idx)
		end
	--ݷͽsocket.send첽¼
	elseif evt == "SEND" then
		if not result then
			socket.disconnect(idx)
		else
			if item.para then
				if item.para.key=="MQTTPUB" then
					mqttpubcb(idx,result,item.para)
				elseif item.para.key=="MQTTSUB" then
					mqttsubcb(idx,result,item.para)
				elseif item.para.key=="MQTTUNSUB" then
					mqttunsubcb(idx,result,item.para)
				elseif item.para.key=="MQTTDUP" then
					mqttdupcb(idx,result,item.data)
				else
					local id = getidbysndpara(item.para.key)
					print("item.para",type(item.para) == "table",type(item.para) == "table" and item.para.typ or item.para,id)
					if id and tmqttpack[id].sndcb then tmqttpack[id].sndcb(idx,result,item.para) end
				end
			end
		end
	--ӱϿ
	elseif evt == "STATE" and result == "CLOSED" then
		sys.timer_stop(datinactive,idx)
		sys.timer_stop(pingreq,idx)
		mqttdup.rmvall(idx)
		tclients[mqttclientidx].sckconnected=false
		tclients[mqttclientidx].mqttconnected=false
		tclients[mqttclientidx].sckrcvs=""
		if tclients[mqttclientidx].discing then
			if tclients[mqttclientidx].discb then tclients[mqttclientidx].discb() end
			tclients[mqttclientidx].discing = false
		else
			sys.timer_start(connectitem,RECONN_PERIOD*1000,mqttclientidx)
		end
	--Ͽlink.shut첽¼
	elseif evt == "STATE" and result == "SHUTED" then
		sys.timer_stop(datinactive,idx)
		sys.timer_stop(pingreq,idx)
		mqttdup.rmvall(idx)
		tclients[mqttclientidx].sckconnected=false
		tclients[mqttclientidx].mqttconnected=false
		tclients[mqttclientidx].sckrcvs=""
		connectitem(mqttclientidx)
	--Ͽsocket.disconnect첽¼
	elseif evt == "DISCONNECT" then
		sys.timer_stop(datinactive,idx)
		sys.timer_stop(pingreq,idx)
		mqttdup.rmvall(idx)
		tclients[mqttclientidx].sckconnected=false
		tclients[mqttclientidx].mqttconnected=false
		tclients[mqttclientidx].sckrcvs=""
		if item=="USER" then
			if tclients[mqttclientidx].discb then tclients[mqttclientidx].discb() end
			tclients[mqttclientidx].discing = false
		else
			connectitem(mqttclientidx)
		end
	--Ͽ٣socket.close첽¼
	elseif evt == "CLOSE" then
		sys.timer_stop(datinactive,idx)
		sys.timer_stop(pingreq,idx)
		mqttdup.rmvall(idx)
		local cb = tclients[mqttclientidx].destroycb
		table.remove(tclients,mqttclientidx)
		if cb then cb() end
	end
	--Ͽ·
	if smatch((type(result)=="string") and result or "","ERROR") then
		socket.disconnect(idx)
	end
end

--[[
connack
  ·MQTT CONNACK
  
        sckidxsocket idx
		packetıĸʽtable{suc=Ƿӳɹ}
ֵ
]]
local function connack(sckidx,packet)
	local mqttclientidx = getclient(sckidx)
	print("connack",packet.suc)
	if packet.suc then
		tclients[mqttclientidx].mqttconnected = true
		mqttdup.rmv(sckidx,tmqttpack["MQTTCONN"].mqttduptyp)
		if tclients[mqttclientidx].connectedcb then tclients[mqttclientidx].connectedcb() end
	else
		if tclients[mqttclientidx].connecterrcb then tclients[mqttclientidx].connecterrcb(packet.reason) end
	end
end

--[[
suback
  ·MQTT SUBACK
  
        sckidxsocket idx
		packetıĸʽtable{seq=ӦSUBSCRIBEк}
ֵ
]]
local function suback(sckidx,packet)
	local mqttclientidx = getclient(sckidx)
	local typ,cb,cbtag = mqttdup.getyp(sckidx,packet.seq)
	print("suback",common.binstohexs(packet.seq))
	mqttdup.rmv(sckidx,nil,nil,packet.seq)
	if cb then cb(cbtag,true) end
end

--[[
unsuback
  ·MQTT UNSUBACK
  
        sckidxsocket idx
		packetıĸʽtable{seq=ӦUNSUBSCRIBEк}
ֵ
]]
local function unsuback(sckidx,packet)
	local mqttclientidx = getclient(sckidx)
	local typ,cb,cbtag = mqttdup.getyp(sckidx,packet.seq)
	print("unsuback",common.binstohexs(packet.seq))
	mqttdup.rmv(sckidx,nil,nil,packet.seq)
	if cb then cb(cbtag,true) end
end

--[[
puback
  ·MQTT PUBACK
  
        sckidxsocket idx
		packetıĸʽtable{seq=ӦPUBLISHк}
ֵ
]]
local function puback(sckidx,packet)
	local mqttclientidx = getclient(sckidx)
	local typ,cb,cbtag = mqttdup.getyp(sckidx,packet.seq)
	print("puback",common.binstohexs(packet.seq),typ)
	mqttdup.rmv(sckidx,nil,nil,packet.seq)
	if cb then cb(cbtag,true) end
end

--[[
svrpublish
  ·MQTT PUBLISH
  
        sckidxsocket idx
		mqttpacketıĸʽtable{qos=,topic,seq,payload}
ֵ
]]
local function svrpublish(sckidx,mqttpacket)
	local mqttclientidx = getclient(sckidx)
	print("svrpublish",mqttpacket.topic,mqttpacket.seq,slen(mqttpacket.payload)>200 and slen(mqttpacket.payload) or mqttpacket.payload)
	if mqttpacket.qos == 1 then snd(sckidx,pack(tclients[mqttclientidx].mqttver,PUBACK,mqttpacket.seq)) end
	if tclients[mqttclientidx].evtcbs then
		if tclients[mqttclientidx].evtcbs["MESSAGE"] then tclients[mqttclientidx].evtcbs["MESSAGE"](common.utf8togb2312(mqttpacket.topic),mqttpacket.payload,mqttpacket.qos) end
	end
end

--[[
pingrsp
  ·MQTT PINGRSP
  
		sckidxsocket idx
ֵ
]]
local function pingrsp(sckidx)
	sys.timer_stop(disconnect,sckidx)
end

--·Ĵ
mqttcmds = {
	[CONNACK] = connack,
	[SUBACK] = suback,
	[UNSUBACK] = unsuback,
	[PUBACK] = puback,
	[PUBLISH] = svrpublish,
	[PINGRSP] = pingrsp,
}

--[[
datinactive
  ͨ쳣
  
		sckidxsocket idx
ֵ
]]
function datinactive(sckidx)
	local mqttclientidx = getclient(sckidx)
	if tclients[mqttclientidx].sckerrcb then
		socket.disconnect(sckidx)
		tclients[mqttclientidx].sckreconncnt=0
		tclients[mqttclientidx].sckreconncyclecnt=0
		tclients[mqttclientidx].sckerrcb("SVRNODATA")
	else
		sys.restart("SVRNODATA")
	end
end

--[[
checkdatactive
  ¿ʼ⡰ͨǷ쳣
  
		sckidxsocket idx
ֵ
]]
local function checkdatactive(sckidx)
	local mqttclientidx = getclient(sckidx)
	sys.timer_start(datinactive,tclients[mqttclientidx].keepalive*1000*3+30000,sckidx) --3ʱ+
end

--[[
rcv
  socketݵĴ
  
        idx socketάsocket idxsocket.connectʱĵһͬԺԲ
        dataյ
ֵ
]]
function rcv(idx,data)
	local mqttclientidx = getclient(idx)
	if not mqttclientidx or not tclients[mqttclientidx].sckconnected then return end
	print("rcv",slen(data)>200 and slen(data) or common.binstohexs(data))
	sys.timer_start(pingreq,(tclients[mqttclientidx].keepalive*1000-(tclients[mqttclientidx].keepalive*1000%2))/2,idx)
	tclients[mqttclientidx].sckrcvs = tclients[mqttclientidx].sckrcvs..data
	if slen(tclients[mqttclientidx].sckrcvs)>1024*10 then collectgarbage() end

	local f,h,t = iscomplete(tclients[mqttclientidx].sckrcvs)

	while f do
		data = ssub(tclients[mqttclientidx].sckrcvs,h,t)
		tclients[mqttclientidx].sckrcvs = ssub(tclients[mqttclientidx].sckrcvs,t+1,-1)
		local packet = unpack(tclients[mqttclientidx].mqttver,data)
		if packet and packet.typ and mqttcmds[packet.typ] then
			mqttcmds[packet.typ](idx,packet)
			if packet.typ ~= CONNACK and packet.typ ~= SUBACK and packet.typ ~= UNSUBACK then
				checkdatactive(idx)
			end
		end
		f,h,t = iscomplete(tclients[mqttclientidx].sckrcvs)
	end
end


--[[
connect
  ̨socketӣ
        Ѿ׼ãӺ̨ᱻ𣬵׼ԶȥӺ̨
		ntfysocket״̬Ĵ
		rcvsocketݵĴ
  
		sckidxsocket idx
		protstringͣЭ飬֧"TCP""UDP"[ѡ]
		hoststringַ֧ͣIPַ[ѡ]
		portnumberͣ˿[ѡ]
ֵ
]]
function connect(sckidx,prot,host,port)
	socket.connect(sckidx,prot,host,port,ntfy,rcv)
	tclients[getclient(sckidx)].sckconning=true
end

--[[
mqttdupcb
  mqttdupдطķͺ첽ص
  
		sckidxsocket idx
		result boolͣͽtrueΪɹΪʧ
		v
ֵ
]]
function mqttdupcb(sckidx,result,v)
	mqttdup.rsm(sckidx,v)
end

--[[
mqttdupind
  mqttdupдطĴ
  
		sckidxsocket idx
		s
ֵ
]]
local function mqttdupind(sckidx,s)
	if not snd(sckidx,s,{key="MQTTDUP"}) then mqttdupcb(sckidx,false,s) end
end

--[[
mqttdupfail
  mqttdupдطģطڣʧ֪ܵͨϢ
  
		sckidxsocket idx
		tĵûԶ
		s
		cbûص
		cbtagûصĵһ
ֵ
]]
local function mqttdupfail(sckidx,t,s,cb,cbtag)
    print("mqttdupfail",t)
	if cb then cb(cbtag,false) end
end

--mqttdupطϢ
local procer =
{
	MQTT_DUP_IND = mqttdupind,
	MQTT_DUP_FAIL = mqttdupfail,
}
--עϢĴ
sys.regapp(procer)


local tmqtt = {}
tmqtt.__index = tmqtt


--[[
create
  һmqtt client
  
		protstringͣЭ飬֧"TCP""UDP"[ѡ]
		hoststringַ֧ͣIPַ[ѡ]
		portnumberͣ˿[ѡ]
		verstringͣMQTTЭ汾ţ֧"3.1""3.1.1"Ĭ"3.1"
ֵ
]]
function create(prot,host,port,ver)
	if #tclients>=2 then assert(false,"tclients maxcnt error") return end
	local mqtt_client =
	{
		prot=prot,
		host=host,
		port=port,
		sckidx=socket.SCK_MAX_CNT-#tclients,
		sckconning=false,
		sckconnected=false,
		sckreconncnt=0,
		sckreconncyclecnt=0,
		sckrcvs="",
		mqttconnected=false,
		mqttver = ver or "3.1",
	}
	setmetatable(mqtt_client,tmqtt)
	table.insert(tclients,mqtt_client)
	return(mqtt_client)
end

--[[
change
  ıһmqtt clientsocket
  
		protstringͣЭ飬֧"TCP""UDP"[ѡ]
		hoststringַ֧ͣIPַ[ѡ]
		portnumberͣ˿[ѡ]
ֵ
]]
function tmqtt:change(prot,host,port)
	self.prot,self.host,self.port=prot or self.prot,host or self.host,port or self.port
end

--[[
destroy
  һmqtt client
  
		destroycbfunctionͣmqtt clientٺĻص[ѡ]
ֵ
]]
function tmqtt:destroy(destroycb)
	local k,v
	self.destroycb = destroycb
	sys.timer_stop(datinactive,self.sckidx)
	for k,v in pairs(tclients) do
		if v.sckidx==self.sckidx then
			socket.close(v.sckidx)
		end
	end
end

--[[
disconnect
  Ͽһmqtt clientҶϿsocket
  
		discbfunctionͣϿĻص[ѡ]
ֵ
]]
function tmqtt:disconnect(discb)
	print("tmqtt:disconnect",self.discing,self.mqttconnected,self.sckconnected)
	sys.timer_stop(datinactive,self.sckidx)
	if self.discing or not self.mqttconnected or not self.sckconnected then
		if discb then discb() end
		return
	end
	self.discb = discb
	if not disconnect(self.sckidx,"USER") and discb then discb() end
end

--[[
configwill
  
  
		flgnumberͣ־֧01
		qosnumberͣ˷Ϣķȼ֧0,1,2
		retainnumberͣ־֧01
		topicstringͣ˷Ϣ⣬gb2312
		payloadstringͣ˷Ϣغɣgb2312
ֵ
]]
function tmqtt:configwill(flg,qos,retain,topic,payload)
	self.willflg=flg or 0
	self.willqos=qos or 0
	self.willretain=retain or 0
	self.willtopic=topic or ""
	self.willpayload=payload or ""
end

--[[
setcleansession
  clean session־
  
		flgnumberͣclean session־֧01ĬΪ1
ֵ
]]
function tmqtt:setcleansession(flg)
	self.cleansession=flg or 1
end

--[[
connect
  mqtt
  
		clientidstringͣclient identifiergb2312[ѡ]
		keepalivenumberͣʱ䣬λ[ѡĬ600]
		userstringͣûgb2312[ѡĬ""]
		passwordstringͣ룬gb2312[ѡĬ""]
		connectedcbfunctionͣmqttӳɹĻص[ѡ]
		connecterrcbfunctionͣmqttʧܵĻص[ѡ]
		sckerrcbfunctionͣsocketʧܵĻص[ѡ]
ֵ
]]
function tmqtt:connect(clientid,keepalive,user,password,connectedcb,connecterrcb,sckerrcb)
	self.clientid=clientid
	self.keepalive=keepalive or 600
	self.user=user or ""
	self.password=password or ""
	--if autoreconnect==nil then autoreconnect=true end
	--self.autoreconnect=autoreconnect
	self.connectedcb=connectedcb
	self.connecterrcb=connecterrcb
	self.sckerrcb=sckerrcb

	tclients[getclient(self.sckidx)]=self

	if self.mqttconnected then print("tmqtt:connect already connected") return end
	if not self.sckconnected then
		connect(self.sckidx,self.prot,self.host,self.port)
		checkdatactive(self.sckidx)
	elseif not self.mqttconnected then
		mqttsnd(self.sckidx,"MQTTCONN")
	else
		if connectedcb then connectedcb() end
	end
end

--[[
publish
  һϢ
  
		topicstringͣϢ⣬gb2312[ѡ]
		payloadݣϢأûԶ룬ļκαת[ѡ]
		flagsnumberͣqosretain־֧0145[ѡĬ0]
				0ʾqos=0retain=0
				1ʾqos=1retain=0
				4ʾqos=0retain=1
				5ʾqos=1retain=1
		ackcbfunctionͣqosΪ1ʱʾյPUBACKĻص,qosΪ0ʱϢͽĻص[ѡ]
		usertagstringͣûصackcbõĵһ[ѡ]
ֵ
]]
function tmqtt:publish(topic,payload,flags,ackcb,usertag)
	--mqtt״̬
	if not self.mqttconnected then
		print("tmqtt:publish not connected")
		if ackcb then ackcb(usertag,false) end
		return
	end

	if flags and flags~=0 and flags~=1 and flags~=4 and flags~=5 then assert(false,"tmqtt:publish not support flags "..flags) return end
	local qos,retain = flags and (bit.band(flags,0x03)) or 0,flags and (bit.isset(flags,2) and 1 or 0) or 0
	--print("tmqtt:publish",flags,qos,retain)
	--publish
	local dat,para = pack(self.mqttver,PUBLISH,{qos=qos,retain=retain,topic=topic,payload=payload})

	--
	local tpara = {key="MQTTPUB",val=para,qos=qos,retain=retain,usertag=usertag,ackcb=ackcb}
	if not snd(self.sckidx,dat,tpara) then
		mqttpubcb(self.sckidx,false,tpara)
	end
end

--[[
subscribe
  
  
		topicstableͣһ߶⣬gb2312룬ȼ֧01{{topic="/topic1",qos=ȼ}, {topic="/topic2",qos=ȼ}, ...}[ѡ]
		ackcbfunctionͣʾյSUBACKĻص[ѡ]
		usertagstringͣûصackcbõĵһ[ѡ]
ֵ
]]
function tmqtt:subscribe(topics,ackcb,usertag)
	--mqtt״̬
	if not self.mqttconnected then
		print("tmqtt:subscribe not connected")
		if ackcb then ackcb(usertag,false) end
		return
	end

	--֧qos 01
	for k,v in pairs(topics) do
		if v.qos==2 then assert(false,"tmqtt:publish not support qos 2") return end
	end

	--subscribe
	local dat,para = pack(self.mqttver,SUBSCRIBE,{topic=topics})

	--
	local tpara = {key="MQTTSUB", val=para, usertag=usertag, ackcb=ackcb}
	if not snd(self.sckidx,dat,tpara) then
		mqttsubcb(self.sckidx,false,tpara)
	end
end


--[[
unsubscribe
  ȡ
  
		topicstableͣһ߶⣬gb2312룬{"/topic1","/topic2",...}[ѡ]
		ackcbfunctionͣʾյUBSUBACKĻص[ѡ]
		usertagstringͣûصackcbõĵһ[ѡ]
ֵ
]]
function tmqtt:unsubscribe(topics,ackcb,usertag)
	--mqtt״̬
	if not self.mqttconnected then
		print("tmqtt:unsubscribe not connected")
		if ackcb then ackcb(usertag,false) end
		return
	end

	--unsubscribe
	local dat,para = pack(self.mqttver,UNSUBSCRIBE,{topic=topics})

	--
	local tpara = {key="MQTTUNSUB", val=para, usertag=usertag, ackcb=ackcb}
	if not snd(self.sckidx,dat,tpara) then
		mqttunsubcb(self.sckidx,false,tpara)
	end
end

--[[
regevtcb
  ע¼Ļص
  
		evtcbsһԻ߶evtcbʽΪ{evt=cb,...}}evtȡֵ£
				"MESSAGE"ʾӷյϢcbʱʽΪcb(topic,payload,qos)
ֵ
]]
function tmqtt:regevtcb(evtcbs)
	self.evtcbs=evtcbs
end

--[[
getstatus
  ȡMQTT CLIENT״̬
  
ֵMQTT CLIENT״̬stringͣ4״̬
		DISCONNECTEDδ״̬
		CONNECTING״̬
		CONNECTED״̬
		DISCONNECTINGϿ״̬
]]
function tmqtt:getstatus()
	if self.mqttconnected then
		return self.discing and "DISCONNECTING" or "CONNECTED"
	elseif self.sckconnected or self.sckconning then
		return "CONNECTING"
	else
		return "DISCONNECTED"
	end
end
