mqtt

Documentation

Index

• Constants
• Variables
• func OnSubMessageReceived(_ MQTT.Client, msg MQTT.Message)
• func RegisterMsgHandler()
• type AccessInfo
• type Client
  • func (mq *Client) InitPubClient()
  • func (mq *Client) InitSubClient()
• type HandlerFunc
• type MessageExpression
  • func NewExpression() *MessageExpression
  • func (exp *MessageExpression) GetExpression(topic string) *MessageExpression
• type MessageMux
  • func NewMessageMux() *MessageMux
  • func (mux *MessageMux) Dispatch(topic string, payload []byte)
  • func (mux *MessageMux) Entry(pattern *MessagePattern, handle func(topic string, payload []byte)) *MessageMux
• type MessageMuxEntry
  • func NewEntry(pattern *MessagePattern, handle func(topic string, payload []byte)) *MessageMuxEntry
• type MessagePattern
  • func NewPattern(resource string) *MessagePattern
  • func (pattern *MessagePattern) Match(topic string) bool
• type Server
  • func NewMqttServer(sqz int, url string, retain bool, qos int) *Server
  • func (m *Server) InitInternalTopics()
  • func (m *Server) Publish(topic string, payload []byte)
  • func (m *Server) RemoveTopic(topic string)
  • func (m *Server) Run() error
  • func (m *Server) SetTopic(topic string)

Constants

const UploadTopic = "SYS/dis/upload_records"

Variables

var (
	// MQTTHub client
	MQTTHub *Client
	// GroupID stands for group id
	GroupID string
	// ConnectedTopic to send connect event
	ConnectedTopic = "$hw/events/connected/%s"
	// DisconnectedTopic to send disconnect event
	DisconnectedTopic = "$hw/events/disconnected/%s"
	// MemberGet to get membership device
	MemberGet = "$hw/events/edgeGroup/%s/membership/get"
	// MemberGetRes to get membership device
	MemberGetRes = "$hw/events/edgeGroup/%s/membership/get/result"
	// MemberDetail which edge-client should be pub when service start
	MemberDetail = "$hw/events/edgeGroup/%s/membership/detail"
	// MemberDetailRes MemberDetail topic resp
	MemberDetailRes = "$hw/events/edgeGroup/%s/membership/detail/result"
	// MemberUpdate updating of the twin
	MemberUpdate = "$hw/events/edgeGroup/%s/membership/updated"
	// GroupUpdate updates a edgegroup
	GroupUpdate = "$hw/events/edgeGroup/%s/updated"
	// GroupAuthGet get temperary aksk from cloudhub
	GroupAuthGet = "$hw/events/edgeGroup/%s/authInfo/get"
	// GroupAuthGetRes temperary aksk from cloudhub
	GroupAuthGetRes = "$hw/events/edgeGroup/%s/authInfo/get/result"
	// SubTopics which edge-client should be sub
	SubTopics = []string{
		"$hw/events/upload/#",
		"$hw/events/device/+/+/state/update",
		"$hw/events/device/+/+/twin/+",
		"$hw/events/node/+/membership/get",
		UploadTopic,
		"+/user/#",
	}
)

Functions

func OnSubMessageReceived

func OnSubMessageReceived(_ MQTT.Client, msg MQTT.Message)

OnSubMessageReceived msg received callback

func RegisterMsgHandler

func RegisterMsgHandler()

RegisterMsgHandler register handler for message if topic is matched in pattern for "$hw/events/device/+/twin/+", "$hw/events/node/+/membership/get", send to twin for other, send to hub for "SYS/dis/upload_records", no need to base64 topic

Types

type AccessInfo

type AccessInfo struct {
	Name    string `json:"name"`
	Type    string `json:"type"`
	Topic   string `json:"topic"`
	Content []byte `json:"content"`
}

AccessInfo that deliver between edge-hub and cloud-hub

type Client

type Client struct {
	MQTTUrl     string
	PubClientID string
	SubClientID string
	Username    string
	Password    string
	PubCli      MQTT.Client
	SubCli      MQTT.Client
}

Client struct

func (*Client) InitPubClient

func (mq *Client) InitPubClient()

InitPubClient init pub client

func (*Client) InitSubClient

func (mq *Client) InitSubClient()

InitSubClient init sub client

type HandlerFunc

type HandlerFunc func(topic string, payload []byte)

type MessageExpression

type MessageExpression struct {
	Matcher  *regexp.Regexp
	VarNames []string
	VarCount int
}

MessageExpression message expression

func NewExpression

func NewExpression() *MessageExpression

NewExpression new expression

func (*MessageExpression) GetExpression

func (exp *MessageExpression) GetExpression(topic string) *MessageExpression

GetExpression get expression

type MessageMux

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

MessageMux message mux

func NewMessageMux

func NewMessageMux() *MessageMux

NewMessageMux new message mux

func (*MessageMux) Dispatch

func (mux *MessageMux) Dispatch(topic string, payload []byte)

func (*MessageMux) Entry

func (mux *MessageMux) Entry(pattern *MessagePattern, handle func(topic string, payload []byte)) *MessageMux

Entry mux := NewMessageMux(ctx, module) mux.Entry(NewPattern(res).Op(opr), handle))

type MessageMuxEntry

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

MessageMuxEntry message mux entry

func NewEntry

func NewEntry(pattern *MessagePattern, handle func(topic string, payload []byte)) *MessageMuxEntry

NewEntry new entry

type MessagePattern

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

MessagePattern message pattern

func NewPattern

func NewPattern(resource string) *MessagePattern

NewPattern new pattern

func (*MessagePattern) Match

func (pattern *MessagePattern) Match(topic string) bool

Match /path/{param}/sub

type Server

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

Server serve as an internal mqtt broker.

func NewMqttServer

func NewMqttServer(sqz int, url string, retain bool, qos int) *Server

NewMqttServer create an internal mqtt server.

func (*Server) InitInternalTopics

func (m *Server) InitInternalTopics()

InitInternalTopics sets internal topics to server by default.

func (*Server) Publish

func (m *Server) Publish(topic string, payload []byte)

Publish will dispatch topic msg to its subscribers directly.

func (*Server) RemoveTopic

func (m *Server) RemoveTopic(topic string)

RemoveTopic remove the topic from internal mqtt broker.

func (*Server) Run

func (m *Server) Run() error

Run launch a server and accept connections.

func (*Server) SetTopic

func (m *Server) SetTopic(topic string)

SetTopic set the topic to internal mqtt broker.