antlinker / go-mqtt

MQTT 3.1.1 golang客户端，支持同步模式，事件模式，自动重连，报文统计,tcp/tls连接等功能

MQTT　golang 客户端

支持

• 支持mqtt 3.1.1协议
• 支持同步模式
• 支持事件模式
• 支持自动重连
• 支持报文数量统计，支持tcp tls连接
• 目前仅支持内存数据存储，可以扩展其他存储

获取

$ go get github.com/antlinker/go-mqtt/client

例子

• 参考例子项目mqttgroup的go-mqtt分支
• 参考例子项目mqttpersonal的go-mqtt分支

API使用

客户端创建

	import (
		fmt
		 "github.com/antlinker/go-mqtt/client"
	)
	.
	.
	.
	//开始日志　false则关闭日志显示
	client.Mlog.SetEnabled(true)
	client, err := client.CreateClient(client.MqttOption{
			Addr:  "tcp://localhost:8883",
			//断开连接１秒后自动连接，０不自动重连
			ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	.
	.
	.
	//断开连接
	client.Disconnect()

连接事件

	//连接事件接口
	type MqttDisConnListener interface {
		event.Listener
		OnDisconning(event *MqttEvent)
		OnDisconned(event *MqttEvent)
		OnLostConn(event *MqttEvent, err error)
	}
	//默认连接事件实现，可以打印事件信息
	type DefaultPrintDisConnListen struct {
	}
	
	func (l *DefaultPrintDisConnListen) OnLostConn(event *MqttEvent, err error) {
		Mlog.Debugf("OnLostConn :%v", err)
	}
    
	func (l *DefaultPrintDisConnListen) OnDisconning(event *MqttEvent) {
		Mlog.Debugf("OnDisconning")
	}
	func (l *DefaultPrintDisConnListen) OnDisconned(event *MqttEvent) {
	 	Mlog.Debugf("OnDisconned")
	}
	//默认事件实现，不做任何事情
	type DefaultDisConnListen struct {
	}

	func (l *DefaultDisConnListen) OnDisconning(event *MqttEvent) {

	}
	func (l *DefaultDisConnListen) OnDisconned(event *MqttEvent) {

	}
	func (l *DefaultDisConnListen) OnLostConn(event *MqttEvent, err error) {

	}
    

例子

    .
    .
    .
	 client, err := client.CreateClient(client.MqttOption{
		Addr:  "tcp://localhost:1883",
		//断开连接１秒后自动连接，０不自动重连
		ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	listener:=&DefaultPrintDisConnListen{}
	//注册事件
	client.AddConnListener(listener)
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	//移除事件
	client.RemoveConnListener(listener)   
	.
	.
	.

断开连接事件

	//事件接口
	type MqttDisConnListener interface {
		event.Listener
		OnDisconning(event *MqttEvent)
		OnDisconned(event *MqttEvent)
		OnLostConn(event *MqttEvent, err error)
	}
	//默认事件实现，可以打印事件信息
	type DefaultPrintDisConnListen struct {
	}
	func (l *DefaultPrintDisConnListen) OnLostConn(event *MqttEvent, err error) {
		Mlog.Debugf("OnLostConn :%v", err)
	}
	func (l *DefaultPrintDisConnListen) OnDisconning(event *MqttEvent) {
		Mlog.Debugf("OnDisconning")
	}
	func (l *DefaultPrintDisConnListen) OnDisconned(event *MqttEvent) {
		Mlog.Debugf("OnDisconned")
	}
	//默认事件实现，不做任何事情
	type DefaultDisConnListen struct {
	}
	func (l *DefaultDisConnListen) OnDisconning(event *MqttEvent) {
	}
	func (l *DefaultDisConnListen) OnDisconned(event *MqttEvent) {
	}
	func (l *DefaultDisConnListen) OnLostConn(event *MqttEvent, err error) {
	}
    

例子

	.
	.
	.
	 client, err := client.CreateClient(client.MqttOption{
		Addr:  "tcp://localhost:1883",
		//断开连接１秒后自动连接，０不自动重连
		ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	listener:=&DefaultPrintDisConnListen{}
	//注册断开连接事件
	client.AddDisConnListener(listener)
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	client.Disconnect()
	//移除事件
	client.RemoveDisConnListener(listener)   
	.
	.
	.

订阅事件

	//事件接口
	type MqttSubListener interface {
		event.Listener
		OnSubStart(event *MqttEvent, sub []SubFilter)
		OnSubSuccess(event *MqttEvent, sub []SubFilter, result []QoS)
	}
    	//默认事件实现，可以打印事件信息
	type DefaultPrintSubscribeListen struct {
	}
	func (*DefaultPrintSubscribeListen) OnSubStart(event *MqttEvent, sub []SubFilter) {
		Mlog.Debugf("OnSubStart:%v", sub)
	}
	func (*DefaultPrintSubscribeListen) OnSubSuccess(event *MqttEvent, sub []SubFilter, result []QoS) {
		Mlog.Debugf("OnSubSuccess:%v:%v", sub, result)
	}
	//默认事件实现，不做任何事情
	type DefaultSubscribeListen struct {
	}
	func (*DefaultSubscribeListen) OnSubStart(event *MqttEvent, sub []SubFilter) {
	}
	func (*DefaultSubscribeListen) OnSubSuccess(event *MqttEvent, sub []SubFilter, result []QoS) {
	}
    

例子

	.
	.
	.
	 client, err := client.CreateClient(client.MqttOption{
			Addr:  "tcp://localhost:1883",
			//断开连接１秒后自动连接，０不自动重连
			ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	listener:=&DefaultPrintSubscribeListen{}
	//注册订阅事件
	client.AddSubListener(listener)
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	mq,err:= client.Subscribe("Test/1", client.QoS1)
	if err!=nil {
		//订阅失败
		panic(fmt.Errorf("订阅失败:%v",err))
	}
	
	//等待订阅成功
	mq.Wait()
	if mq.Err()!=nil {
		//订阅失败
		panic(fmt.Errorf("订阅失败:%v",mqt.Err()))
	}
	mq,err= client.Subscribes(client.CreateSubFilter("test/1", QoS1),
		client.CreateSubFilter("test/2", QoS2),
		client.CreateSubFilter("test/0", QoS0))
	if err!=nil {
		//订阅失败
		panic(fmt.Errorf("订阅失败:%v",err))
	}
	//等待订阅成功
	mq.Wait()
	if mq.Err()!=nil {
		//订阅失败
		panic(fmt.Errorf("订阅失败:%v",mqt.Err()))
	}
	.
	.
	.
	//移除事件
	client.RemoveSubListener(listener)   
	.
	.
	.

发布消息事件

	//事件接口
	//发布消息事件监听
	type MqttPublishListener interface {
		event.Listener
		//准备发送消息
		OnPubReady(event *MqttPublishEvent, packet *MqttPacket)
		//消息发布成功
		OnPubSuccess(event *MqttPublishEvent, packet *MqttPacket)
		//消息发布完成
		OnPubFinal(event *MqttPublishEvent, packet *MqttPacket)
	}
    	//默认事件实现，可以打印事件信息
	type DefaultPrintPublishListen struct {
	}
	func (*DefaultPrintPublishListen) OnPubReady(event *MqttPublishEvent, mp *MqttPacket) {
		Mlog.Debugf("OnPubReady:%v", event.GetSendCnt(PubCnt_TOTAL))
	}
	func (*DefaultPrintPublishListen) OnPubSuccess(event *MqttPublishEvent, mp *MqttPacket) {
		Mlog.Debugf("OnPubSuccess:%v", mp.Packet)

	}
	func (*DefaultPrintPublishListen) OnPubFinal(event *MqttPublishEvent, mp *MqttPacket) {
		Mlog.Debugf("OnPubFinal:%v", mp.Packet)
	}

	//默认事件实现，不做任何事情
	type DefaultPublishListen struct {
	}

	func (*DefaultPublishListen) OnPubReady(event *MqttPublishEvent, mp *MqttPacket) {

	}
	func (*DefaultPublishListen) OnPubSuccess(event *MqttPublishEvent, mp *MqttPacket) {

	}
	func (*DefaultPublishListen) OnPubFinal(event *MqttPublishEvent, mp *MqttPacket) {

	}
    

例子

	.
	.
	.
	 client, err := client.CreateClient(client.MqttOption{
			Addr:  "tcp://localhost:1883",
			//断开连接１秒后自动连接，０不自动重连
			ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	listener:=&DefaultPrintPublishListen{}
	//注册订阅事件
	client.AddPubListener(listener)
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	mq, err := client.Publish("test0", QoS0, false, []byte("测试"))
	if err!=nil {
		//发布消息失败
		panic(fmt.Errorf("发布消息失败:%v",err))
	}
	
	//等待发布消息完成
	mq.Wait()
	if mq.Err()!=nil {
		//发布消息失败
		panic(fmt.Errorf("发布消息失败:%v",mqt.Err()))
	}
	
	.
	.
	.
	//移除事件
	client.RemoveSubListener(listener)   
	.
	.
	.

接收消息事件

	//事件接口
	//接收消息事件监听
	type MqttRecvPubListener interface {
		event.Listener
		OnRecvPublish(event *MqttRecvPubEvent, topic string, payload []byte, qos QoS)
	}
    	//默认事件实现，可以打印事件信息
	type DefaultPrintRecvPubListen struct {
	}
	func (l *DefaultPrintRecvPubListen) OnRecvPublish(event *MqttRecvPubEvent, topic string, payload []byte, qos QoS) {
		Mlog.Debugf("OnRecvPublish:%s(%d) :%s", topic, qos, string(payload))
	}
	//默认事件实现，不做任何事情
	type DefaultRecvPubListen struct {
	}

	func (l *DefaultRecvPubListen) OnRecvPublish(event *MqttRecvPubEvent, topic string, payload []byte, qos QoS) {

	}
    

例子

	.
	.
	.
	 client, err := client.CreateClient(client.MqttOption{
			Addr:  "tcp://localhost:1883",
			//断开连接１秒后自动连接，０不自动重连
			ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	listener:=&MqttRecvPubListener{}
	//注册订阅事件
	client.AddRecvPubListener(listener)
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	mq,err= client.Subscribe("test",QoS2)
	if err!=nil {
		//订阅失败
		panic(fmt.Errorf("订阅失败:%v",err))
	}
	//等待订阅成功
	mq.Wait()
	if mq.Err()!=nil {
		//订阅失败
		panic(fmt.Errorf("订阅失败:%v",mqt.Err()))
	}
	mq, err := client.Publish("test", QoS1, false, []byte("测试"))
	if err!=nil {
		//发布消息失败
		panic(fmt.Errorf("发布消息失败:%v",err))
	}
	
	//等待发布消息完成
	mq.Wait()
	if mq.Err()!=nil {
		//发布消息失败
		panic(fmt.Errorf("发布消息失败:%v",mqt.Err()))
	}
	
	.
	.
	.
	//移除事件
	client.RemoveRecvPubListener(listener)   
	.
	.
	.

取消订阅事件

	//事件接口
	type MqttUnSubListener interface {
		event.Listener
		OnUnSubStart(event *MqttEvent, filter []string)
		OnUnSubSuccess(event *MqttEvent, filter []string)
	}
    	//默认事件实现，可以打印事件信息
	type DefaultPrintUnSubListen struct {
	}

	func (*DefaultPrintUnSubListen) OnUnSubStart(event *MqttEvent, filter []string) {
		Mlog.Debugf("OnUnSubStart:%v", filter)
	}
	func (*DefaultPrintUnSubListen) OnUnSubSuccess(event *MqttEvent, filter []string) {
		Mlog.Debugf("OnUnSubSuccess:%v", filter)
	}

	//默认事件实现，不做任何事情
	type DefaultUnSubListen struct {
	}
	func (*DefaultUnSubListen) OnUnSubStart(event *MqttEvent, filter []string) {
	}
	func (*DefaultUnSubListen) OnUnSubSuccess(event *MqttEvent, filter []string) {
	}
    

例子

	.
	.
	.
	 client, err := client.CreateClient(client.MqttOption{
			Addr:  "tcp://localhost:1883",
			//断开连接１秒后自动连接，０不自动重连
			ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	listener:=&DefaultPrintUnSubListen{}
	//注册订阅事件
	client.AddUnSubListener(listener)
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	mq,err:= client.UnSubscribe("Test/1")
	if err!=nil {
		//取消订阅失败
		panic(fmt.Errorf("取消订阅失败:%v",err))
	}
	
	//等待取消订阅成功
	mq.Wait()
	if mq.Err()!=nil {
		//取消订阅失败
		panic(fmt.Errorf("取消订阅失败:%v",mqt.Err()))
	}
	mq,err= client.UnSubscribes("test1","test2")
	if err!=nil {
		//订阅失败
		panic(fmt.Errorf("取消订阅失败:%v",err))
	}
	//等待取消订阅成功
	mq.Wait()
	if mq.Err()!=nil {
		//订阅失败
		panic(fmt.Errorf("订阅失败:%v",mqt.Err()))
	}
	.
	.
	.
	//移除事件
	client.RemoveUnSubListener(listener)   
	.
	.
	.

报文发送接收事件

	//发送接收报文接口
	type MqttPacketListener interface {
		event.Listener
		//接收到报文
		OnRecvPacket(event *MqttEvent, msg packet.MessagePacket, recvPacketCnt int64)
		//发送报文
		OnSendPacket(event *MqttEvent, msg packet.MessagePacket, sendPacketCnt int64, err error)
	}
    	//默认事件实现，可以打印事件信息
	type DefaultPrintPacketListen struct {
	}
	func (l *DefaultPrintPacketListen) OnRecvPacket(event *MqttEvent, msg packet.MessagePacket, recvPacketCnt int64) {
		Mlog.Debugf("OnRecvPacket:(%d) :%v", recvPacketCnt, msg)
	}
	func (l *DefaultPrintPacketListen) OnSendPacket(event *MqttEvent, msg packet.MessagePacket, sendPacketCnt int64, err error) {
		Mlog.Debugf("OnSendPacket:(%d) :%v\n %v", sendPacketCnt, msg, err)
	}

	//默认事件实现，不做任何事情
	type DefaultPacketListen struct {
	}
	func (l *DefaultPacketListen) OnRecvPacket(event *MqttEvent, packet packet.MessagePacket, recvPacketCnt int64) {
	}
	func (l *DefaultPacketListen) OnSendPacket(event *MqttEvent, packet packet.MessagePacket, sendPacketCnt int64) {
	}
    

例子

	.
	.
	.
	 client, err := client.CreateClient(client.MqttOption{
			Addr:  "tcp://localhost:1883",
			//断开连接１秒后自动连接，０不自动重连
			ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	listener:=&DefaultPrintPacketListen{}
	//注册订阅事件
	client.AddPacketListener(listener)
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	mq,err:= client.Subscribe("Test/1",client.QoS1)
	if err!=nil {
		//取消订阅失败
		panic(fmt.Errorf("取消订阅失败:%v",err))
	}
	
	//等待取消订阅成功
	mq.Wait()
	if mq.Err()!=nil {
		//取消订阅失败
		panic(fmt.Errorf("取消订阅失败:%v",mqt.Err()))
	}
	mq,err= client.Publish("Test/1", QoS1, false, []byte("测试"))
	if err!=nil {
		//订阅失败
		panic(fmt.Errorf("取消订阅失败:%v",err))
	}
	mq,err= client.Publish("Test/2", QoS2, false, []byte("测试"))
	if err!=nil {
		//订阅失败
		panic(fmt.Errorf("取消订阅失败:%v",err))
	}
	.
	.
	.
	//移除事件
	client.RemovePacketListener(listener)   
	.
	.
	.

go-mqtt's Issues

关闭网络后能重连，运行一段时间后连接自动丢失，再也无法连接上

OnLostConn: EOF
OnLostConn: read tcp 192.168.94.180:59636->59.110.94.190:10001: i/o timeout
OnLostConn: read tcp 192.168.94.180:59637->59.110.94.190:10001: i/o timeout
OnLostConn: read tcp 192.168.94.180:59638->59.110.94.190:10001: i/o timeout
OnLostConn: read tcp 192.168.94.180:59640->59.110.94.190:10001: i/o timeout
OnLostConn: read tcp 192.168.94.180:59641->59.110.94.190:10001: i/o timeout
OnLostConn: read tcp 192.168.94.180:59644->59.110.94.190:10001: i/o timeout
OnLostConn: read tcp 192.168.94.180:59649->59.110.94.190:10001: i/o timeout
OnLostConn: read tcp 192.168.94.180:59653->59.110.94.190:10001: i/o timeout

测试列子变量client与包名client同名，导致 client.QoS1， client.QoS2显示未定义编译无法通过

func TestCli(t *testing.T) {
client, err := client.CreateClient(client.MqttOption{
Addr: "tcp://192.168.175.6:1883",
})
listener := &TestMqttListener{
subFilter: "test",
subQos: client.QoS2,
sendTopic: "test",
sendPayload: []byte("我的测试"),
sendQos: client.QoS1,
sendRetain: false,
}
client.AddConnListener(listener)
client.AddPubListener(listener)
client.AddRecvPubListener(listener)
client.AddSubListener(listener)
client.AddDisConnListener(listener)
listener.Add(1)
Expect(err).NotTo(HaveOccurred())
By("进行连接")
err = client.Connect()
Expect(err).NotTo(HaveOccurred())
listener.Wait()
By("断开连接")

client.Disconnect()

}

测试列子变量client与包名client同名，导致 client.QoS1， client.QoS2显示未定义编译无法通过

undefined: MqttEvent

package main

import (
  "fmt"
 "github.com/antlinker/go-mqtt/client"
 "github.com/antlinker/go-mqtt/event"
)


  type MqttSubListener interface {
        event.Listener
        OnSubStart(event *MqttEvent, sub []SubFilter)
        OnSubSuccess(event *MqttEvent, sub []SubFilter, result []QoS)
    }
        //默认事件实现，可以打印事件信息
    type DefaultPrintSubscribeListen struct {
    }
    func (*DefaultPrintSubscribeListen) OnSubStart(event *MqttEvent, sub []SubFilter) {
        Mlog.Debugf("OnSubStart:%v", sub)
    }
    func (*DefaultPrintSubscribeListen) OnSubSuccess(event *MqttEvent, sub []SubFilter, result []QoS) {
        Mlog.Debugf("OnSubSuccess:%v:%v", sub, result)
    }
    //默认事件实现，不做任何事情
    type DefaultSubscribeListen struct {
    }
    func (*DefaultSubscribeListen) OnSubStart(event *MqttEvent, sub []SubFilter) {
    }
    func (*DefaultSubscribeListen) OnSubSuccess(event *MqttEvent, sub []SubFilter, result []QoS) {
    }


func main(){

    //开始日志　false则关闭日志显示
    client.Mlog.SetEnabled(true)
    client, err := client.CreateClient(client.MqttOption{
            Addr:  "tcp://127.0.0.1:1883",
            //断开连接１秒后自动连接，０不自动重连
            ReconnTimeInterval: 1,
    })
    if err!=nil {
        //配置文件解析失败
        panic("配置文件解析失败")
    }
    //建立连接
    err1:=client.Connect()
    if err1!=nil {
        //连接失败，不会进入自动重连状态
        panic(fmt.Errorf("连接失败:%v",err))
    }


  mq,err:= client.Subscribe("Test/1", client.QoS1)
    if err!=nil {
        //订阅失败
        panic(fmt.Errorf("订阅失败:%v",err))
    }

    //等待订阅成功
    mq.Wait()
    if mq.Err()!=nil {
        //订阅失败
        panic(fmt.Errorf("订阅失败:%v",mqt.Err()))
    }
    mq,err= client.Subscribes(client.CreateSubFilter("test/1", QoS1),
        client.CreateSubFilter("test/2", QoS2),
        client.CreateSubFilter("test/0", QoS0))
    if err!=nil {
        //订阅失败
        panic(fmt.Errorf("订阅失败:%v",err))
    }
    //等待订阅成功
    mq.Wait()
    if mq.Err()!=nil {
        //订阅失败
        panic(fmt.Errorf("订阅失败:%v",mqt.Err()))
    }

    //移除事件
    client.RemoveSubListener(listener)   

}

.dd.go:12: undefined: MqttEvent
.dd.go:12: undefined: SubFilter
.dd.go:13: undefined: MqttEvent
.dd.go:13: undefined: SubFilter
.dd.go:13: undefined: QoS
.dd.go:18: undefined: MqttEvent
.dd.go:18: undefined: SubFilter
.dd.go:21: undefined: MqttEvent
.dd.go:21: undefined: SubFilter
.dd.go:21: undefined: QoS
.dd.go:21: too many errors