mqtt

Documentation

Index

• Variables
• func NewMQ() mq.IReactMQ
• type ReactMQTT
  • func (this *ReactMQTT) Close()
  • func (this *ReactMQTT) Connect(mode mq.Mode, strURL string) (err error)
  • func (this *ReactMQTT) Consume(strBindingKey, strQueueName string, handler mq.ReactHandler) (err error)
  • func (this *ReactMQTT) Debug(enable bool)
  • func (this *ReactMQTT) GetAdapter() (adapter mq.Adapter)
  • func (this *ReactMQTT) IsClosed() bool
  • func (this *ReactMQTT) NewTLSConfig(ca, key, crt string) *tls.Config
  • func (this *ReactMQTT) OnConnect(Client MQTT.Client)
  • func (this *ReactMQTT) OnDisconnect(Client MQTT.Client, err error)
  • func (this *ReactMQTT) OnReceive(client MQTT.Client, msg MQTT.Message)
  • func (this *ReactMQTT) Publish(strBindingKey, strQueueName, key string, value string) (err error)
  • func (this *ReactMQTT) Reconnect() (err error)

Variables

var PARAM_CLIENT_ID = "client-id"

var PARAM_TLS = "tls"

var PARAM_TLS_CA = "ca"

var PARAM_TLS_CER = "cer"

var PARAM_TLS_KEY = "key"

var TLS_IS_ENABLE = "true"

Functions

func NewMQ

func NewMQ() mq.IReactMQ

Types

type ReactMQTT

type ReactMQTT struct {
	// contains filtered or unexported fields
}

func (*ReactMQTT) Close

func (this *ReactMQTT) Close()

关闭MQ

func (*ReactMQTT) Connect

func (this *ReactMQTT) Connect(mode mq.Mode, strURL string) (err error)

* @brief MQ服务器连接接口定义 * @param strUrl 连接服务器URL( 格式规范 mqtt://username:password@host:port[/config?tls=[true|false]&&ca=ca.crt&key=client.key&cer=client.crt&client-id=MyClientID] ) * @return err 连接失败返回具体错误信息

func (*ReactMQTT) Consume

func (this *ReactMQTT) Consume(strBindingKey, strQueueName string, handler mq.ReactHandler) (err error)

* @brief 消息消费接口定义 * @param strBindingKey 队列绑定Key * @param strQueueName 队列名称 * @param handler 消费回调处理对象 * @return err 成功返回nil，失败返回返回具体错误信息 * @remark 服务器异常或重启时内部会自动重连服务器

func (*ReactMQTT) Debug

func (this *ReactMQTT) Debug(enable bool)

* @brief 开启或关闭调式模式 * @param enable true开启/false关闭

func (*ReactMQTT) GetAdapter

func (this *ReactMQTT) GetAdapter() (adapter mq.Adapter)

* @brief 获取当前MQ类型 * @param adapter MQ类型

func (*ReactMQTT) IsClosed

func (this *ReactMQTT) IsClosed() bool

* @brief 判定是否MQ服务器断开连接（异常宕机或重启） * @param * @return 远程服务器连接断开返回true，否则返回false

func (*ReactMQTT) NewTLSConfig

func (this *ReactMQTT) NewTLSConfig(ca, key, crt string) *tls.Config

func (*ReactMQTT) OnConnect

func (this *ReactMQTT) OnConnect(Client MQTT.Client)

MQTT 服务器连接成功回调

func (*ReactMQTT) OnDisconnect

func (this *ReactMQTT) OnDisconnect(Client MQTT.Client, err error)

MQTT 服务器断开连接回调

func (*ReactMQTT) OnReceive

func (this *ReactMQTT) OnReceive(client MQTT.Client, msg MQTT.Message)

MQTT 服务器接收数据回调

func (*ReactMQTT) Publish

func (this *ReactMQTT) Publish(strBindingKey, strQueueName, key string, value string) (err error)

* @brief 消息发布接口定义(仅支持字符串类型消息) * @param strBindingKey 队列绑定Key(topic) * @param strQueueName 队列名称(group) * @param key 消息KEY(仅kafka必填，其他MQ类型默认填PRODUCER_KEY_NULL) * @param value 消息数据 * @return err 发布失败返回具体错误信息

func (*ReactMQTT) Reconnect

func (this *ReactMQTT) Reconnect() (err error)

* @brief MQ服务器重新连接接口定义 * @param * @return err 连接失败返回具体错误信息 * @remark 当Publish返回错误且IsClosed()方法亦返回true时调用此方法重连MQ服务器 * 如果已使用Consume订阅过，内部会监听连接断开事件并自动重连。