开源包mqtt源码_订阅和发布
重头戏： 订阅
官方订阅代码
// Subscribe starts a new subscription. Provide a MessageHandler to be executed when
// a message is published on the topic provided.
//
// If options.OrderMatters is true (the default) then callback must not block or
// call functions within this package that may block (e.g. Publish) other than in
// a new go routine.
// callback must be safe for concurrent use by multiple goroutines.
func (c *client) Subscribe(topic string, qos byte, callback MessageHandler) Token {
token := newToken(packets.Subscribe).(*SubscribeToken)
DEBUG.Println(CLI, "enter Subscribe")
// 连接判断代码 与主线无关
if !c.IsConnected() {
token.setError(ErrNotConnected)
return token
}
if !c.IsConnectionOpen() {
switch {
case !c.options.ResumeSubs:
// if not connected and resumesubs not set this sub will be thrown away
token.setError(fmt.Errorf("not currently connected and ResumeSubs not set"))
return token
case c.options.CleanSession && c.connectionStatus() == reconnecting:
// if reconnecting and cleansession is true this sub will be thrown away
token.setError(fmt.Errorf("reconnecting state and cleansession is true"))
return token
}
}
sub := packets.NewControlPacket(packets.Subscribe).(*packets.SubscribePacket)
if err := validateTopicAndQos(topic, qos); err != nil { // topic 格式校验
token.setError(err)
return token
}
sub.Topics = append(sub.Topics, topic)
sub.Qoss = append(sub.Qoss, qos)
if strings.HasPrefix(topic, "$share/") {
topic = strings.Join(strings.Split(topic, "/")[2:], "/")
}
if strings.HasPrefix(topic, "$queue/") {
topic = strings.TrimPrefix(topic, "$queue/")
}
if callback != nil { // 添加路由， 这个需要深入看
c.msgRouter.addRoute(topic, callback)
}
token.subs = append(token.subs, topic)
if sub.MessageID == 0 {
mID := c.getID(token)
if mID == 0 {
token.setError(fmt.Errorf("no message IDs available"))
return token
}
sub.MessageID = mID
token.messageID = mID
}
DEBUG.Println(CLI, sub.String())
persistOutbound(c.persist, sub) // 存储订阅
switch c.connectionStatus() {
case connecting:
DEBUG.Println(CLI, "storing subscribe message (connecting), topic:", topic)
case reconnecting:
DEBUG.Println(CLI, "storing subscribe message (reconnecting), topic:", topic)
default:
DEBUG.Println(CLI, "sending subscribe message, topic:", topic)
subscribeWaitTimeout := c.options.WriteTimeout
if subscribeWaitTimeout == 0 {
subscribeWaitTimeout = time.Second * 30
}
select {
case c.oboundP <- &PacketAndToken{p: sub, t: token}:
// 这个写法下面没东西，容易看眼睛花掉
// 那这段代码的意义是啥呢? 回调函数只是在 route 用了
case <-time.After(subscribeWaitTimeout):
token.setError(errors.New("subscribe was broken by timeout"))
}
}
DEBUG.Println(CLI, "exit Subscribe")
return token
}
router
链表加上管道，但是可以深入看一下 matchAndDispatch() 函数， 主要是做一些ack 管道开始关闭操作
type router struct {
sync.RWMutex
routes *list.List
defaultHandler MessageHandler
messages chan *packets.PublishPacket
}
// newRouter returns a new instance of a Router and channel which can be used to tell the Router
// to stop
func newRouter() *router {
router := &router{routes: list.New(), messages: make(chan *packets.PublishPacket)}
return router
}
// addRoute takes a topic string and MessageHandler callback. It looks in the current list of
// routes to see if there is already a matching Route. If there is it replaces the current
// callback with the new one. If not it add a new entry to the list of Routes.
func (r *router) addRoute(topic string, callback MessageHandler) {
r.Lock()
defer r.Unlock()
for e := r.routes.Front(); e != nil; e = e.Next() {
if e.Value.(*route).topic == topic {
r := e.Value.(*route)
r.callback = callback
return
}
}
r.routes.PushBack(&route{topic: topic, callback: callback})
}
// deleteRoute takes a route string, looks for a matching Route in the list of Routes. If
// found it removes the Route from the list.
func (r *router) deleteRoute(topic string) {
r.Lock()
defer r.Unlock()
for e := r.routes.Front(); e != nil; e = e.Next() {
if e.Value.(*route).topic == topic {
r.routes.Remove(e)
return
}
}
}
// setDefaultHandler assigns a default callback that will be called if no matching Route
// is found for an incoming Publish.
func (r *router) setDefaultHandler(handler MessageHandler) {
r.Lock()
defer r.Unlock()
r.defaultHandler = handler
}
// matchAndDispatch takes a channel of Message pointers as input and starts a go routine that
// takes messages off the channel, matches them against the internal route list and calls the
// associated callback (or the defaultHandler, if one exists and no other route matched). If
// anything is sent down the stop channel the function will end.
func (r *router) matchAndDispatch(messages <-chan *packets.PublishPacket, order bool, client *client) <-chan *PacketAndToken {
var wg sync.WaitGroup
ackOutChan := make(chan *PacketAndToken) // Channel returned to caller; closed when messages channel closed
var ackInChan chan *PacketAndToken // ACKs generated by ackFunc get put onto this channel
stopAckCopy := make(chan struct{}) // Closure requests stop of go routine copying ackInChan to ackOutChan
ackCopyStopped := make(chan struct{}) // Closure indicates that it is safe to close ackOutChan
goRoutinesDone := make(chan struct{}) // closed on wg.Done()
if order {
ackInChan = ackOutChan // When order = true no go routines are used so safe to use one channel and close when done
} else {
// When order = false ACK messages are sent in go routines so ackInChan cannot be closed until all goroutines done
ackInChan = make(chan *PacketAndToken)
go func() { // go routine to copy from ackInChan to ackOutChan until stopped
for {
select {
case a := <-ackInChan:
ackOutChan <- a
case <-stopAckCopy:
close(ackCopyStopped) // Signal main go routine that it is safe to close ackOutChan
for {
select {
case <-ackInChan: // drain ackInChan to ensure all goRoutines can complete cleanly (ACK dropped)
DEBUG.Println(ROU, "matchAndDispatch received acknowledgment after processing stopped (ACK dropped).")
case <-goRoutinesDone:
close(ackInChan) // Nothing further should be sent (a panic is probably better than silent failure)
DEBUG.Println(ROU, "matchAndDispatch order=false copy goroutine exiting.")
return
}
}
}
}
}()
}
go func() { // Main go routine handling inbound messages
for message := range messages {
// DEBUG.Println(ROU, "matchAndDispatch received message")
sent := false
r.RLock()
m := messageFromPublish(message, ackFunc(ackInChan, client.persist, message))
var handlers []MessageHandler
for e := r.routes.Front(); e != nil; e = e.Next() {
if e.Value.(*route).match(message.TopicName) {
if order {
handlers = append(handlers, e.Value.(*route).callback)
} else {
hd := e.Value.(*route).callback
wg.Add(1)
go func() {
hd(client, m)
m.Ack()
wg.Done()
}()
}
sent = true
}
}
if !sent {
if r.defaultHandler != nil {
if order {
handlers = append(handlers, r.defaultHandler)
} else {
wg.Add(1)
go func() {
r.defaultHandler(client, m)
m.Ack()
wg.Done()
}()
}
} else {
DEBUG.Println(ROU, "matchAndDispatch received message and no handler was available. Message will NOT be acknowledged.")
}
}
r.RUnlock()
for _, handler := range handlers {
handler(client, m)
m.Ack()
}
// DEBUG.Println(ROU, "matchAndDispatch handled message")
}
if order {
close(ackOutChan)
} else { // Ensure that nothing further will be written to ackOutChan before closing it
close(stopAckCopy)
<-ackCopyStopped
close(ackOutChan)
go func() {
wg.Wait() // Note: If this remains running then the user has handlers that are not returning
close(goRoutinesDone)
}()
}
DEBUG.Println(ROU, "matchAndDispatch exiting")
}()
return ackOutChan
}
那么问题来了，订阅的核心代码在哪？
在上一次连接代码分析时候，在连接下面有创建各种管道代码
// 下面是连接 ok 代码
inboundFromStore := make(chan packets.ControlPacket) // there may be some inbound comms packets in the store that are awaiting processing
if c.startCommsWorkers(conn, inboundFromStore) { // 开启各种监听数据管道
// Take care of any messages in the store
if !c.options.CleanSession {
c.resume(c.options.ResumeSubs, inboundFromStore)
} else {
c.persist.Reset()
}
} else {
WARN.Println(CLI, "Connect() called but connection established in another goroutine")
}
close(inboundFromStore)
t.flowComplete()
DEBUG.Println(CLI, "exit startClient")
连接建立的代码
还是建立各种管道
// startCommsWorkers is called when the connection is up.
// It starts off all of the routines needed to process incoming and outgoing messages.
// Returns true if the comms workers were started (i.e. they were not already running)
func (c *client) startCommsWorkers(conn net.Conn, inboundFromStore <-chan packets.ControlPacket) bool {
DEBUG.Println(CLI, "startCommsWorkers called")
c.connMu.Lock()
defer c.connMu.Unlock()
if c.conn != nil {
WARN.Println(CLI, "startCommsWorkers called when commsworkers already running")
conn.Close() // No use for the new network connection
return false
}
c.conn = conn // Store the connection
c.stop = make(chan struct{})
if c.options.KeepAlive != 0 {
atomic.StoreInt32(&c.pingOutstanding, 0)
c.lastReceived.Store(time.Now())
c.lastSent.Store(time.Now())
c.workers.Add(1)
go keepalive(c, conn) // 健康监测 一段时间后去Ping
}
// matchAndDispatch will process messages received from the network. It may generate acknowledgements
// It will complete when incomingPubChan is closed and will close ackOut prior to exiting
incomingPubChan := make(chan *packets.PublishPacket)
c.workers.Add(1) // Done will be called when ackOut is closed
ackOut := c.msgRouter.matchAndDispatch(incomingPubChan, c.options.Order, c)
c.setConnected(connected)
DEBUG.Println(CLI, "client is connected/reconnected")
if c.options.OnConnect != nil {
go c.options.OnConnect(c)
}
// c.oboundP and c.obound need to stay active for the life of the client because, depending upon the options,
// messages may be published while the client is disconnected (they will block unless in a goroutine). However
// to keep the comms routines clean we want to shutdown the input messages it uses so create out own channels
// and copy data across.
commsobound := make(chan *PacketAndToken) // outgoing publish packets
commsoboundP := make(chan *PacketAndToken) // outgoing 'priority' packet
c.workers.Add(1)
go func() {
defer c.workers.Done()
for {
select {
case msg := <-c.oboundP:
commsoboundP <- msg
case msg := <-c.obound: // 接受订阅的管道
commsobound <- msg
case msg, ok := <-ackOut:
if !ok {
ackOut = nil // ignore channel going forward
c.workers.Done() // matchAndDispatch has completed
continue // await next message
}
commsoboundP <- msg
case <-c.stop:
// Attempt to transmit any outstanding acknowledgements (this may well fail but should work if this is a clean disconnect)
if ackOut != nil {
for msg := range ackOut {
commsoboundP <- msg
}
c.workers.Done() // matchAndDispatch has completed
}
close(commsoboundP) // Nothing sending to these channels anymore so close them and allow comms routines to exit
close(commsobound)
DEBUG.Println(CLI, "startCommsWorkers output redirector finished")
return
}
}
}()
commsIncomingPub, commsErrors := startComms(c.conn, c, inboundFromStore, commsoboundP, commsobound)
c.commsStopped = make(chan struct{})
go func() {
for {
if commsIncomingPub == nil && commsErrors == nil {
break
}
select {
case pub, ok := <-commsIncomingPub:
if !ok {
// Incoming comms has shutdown
close(incomingPubChan) // stop the router
commsIncomingPub = nil
continue
}
// Care is needed here because an error elsewhere could trigger a deadlock
sendPubLoop:
for {
select {
case incomingPubChan <- pub:
break sendPubLoop
case err, ok := <-commsErrors:
if !ok { // commsErrors has been closed so we can ignore it
commsErrors = nil
continue
}
ERROR.Println(CLI, "Connect comms goroutine - error triggered during send Pub", err)
c.internalConnLost(err) // no harm in calling this if the connection is already down (or shutdown is in progress)
continue
}
}
case err, ok := <-commsErrors:
if !ok {
commsErrors = nil
continue
}
ERROR.Println(CLI, "Connect comms goroutine - error triggered", err)
c.internalConnLost(err) // no harm in calling this if the connection is already down (or shutdown is in progress)
continue
}
}
DEBUG.Println(CLI, "incoming comms goroutine done")
close(c.commsStopped)
}()
DEBUG.Println(CLI, "startCommsWorkers done")
return true
}
核心的发布代码
说实话，就是往conn里面写数据，但是前期的处理真的恶心，垃圾回调函数
// startOutgoingComms initiates a go routine to transmit outgoing packets.
// Pass in an open network connection and channels for outbound messages (including those triggered
// directly from incoming comms).
// Returns a channel that will receive details of any errors (closed when the goroutine exits)
// This function wil only terminate when all input channels are closed
func startOutgoingComms(conn net.Conn,
c commsFns,
oboundp <-chan *PacketAndToken,
obound <-chan *PacketAndToken,
oboundFromIncoming <-chan *PacketAndToken,
) <-chan error {
errChan := make(chan error)
DEBUG.Println(NET, "outgoing started")
go func() {
for {
DEBUG.Println(NET, "outgoing waiting for an outbound message")
// This goroutine will only exits when all of the input channels we receive on have been closed. This approach is taken to avoid any
// deadlocks (if the connection goes down there are limited options as to what we can do with anything waiting on us and
// throwing away the packets seems the best option)
if oboundp == nil && obound == nil && oboundFromIncoming == nil {
DEBUG.Println(NET, "outgoing comms stopping")
close(errChan)
return
}
select {
case pub, ok := <-obound:
if !ok {
obound = nil
continue
}
msg := pub.p.(*packets.PublishPacket)
DEBUG.Println(NET, "obound msg to write", msg.MessageID)
writeTimeout := c.getWriteTimeOut()
if writeTimeout > 0 {
if err := conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil {
ERROR.Println(NET, "SetWriteDeadline ", err)
}
}
if err := msg.Write(conn); err != nil {
ERROR.Println(NET, "outgoing obound reporting error ", err)
pub.t.setError(err)
// report error if it's not due to the connection being closed elsewhere
if !strings.Contains(err.Error(), closedNetConnErrorText) {
errChan <- err
}
continue
}
if writeTimeout > 0 {
// If we successfully wrote, we don't want the timeout to happen during an idle period
// so we reset it to infinite.
if err := conn.SetWriteDeadline(time.Time{}); err != nil {
ERROR.Println(NET, "SetWriteDeadline to 0 ", err)
}
}
if msg.Qos == 0 {
pub.t.flowComplete()
}
DEBUG.Println(NET, "obound wrote msg, id:", msg.MessageID)
case msg, ok := <-oboundp:
if !ok {
oboundp = nil
continue
}
DEBUG.Println(NET, "obound priority msg to write, type", reflect.TypeOf(msg.p))
if err := msg.p.Write(conn); err != nil {
ERROR.Println(NET, "outgoing oboundp reporting error ", err)
if msg.t != nil {
msg.t.setError(err)
}
errChan <- err
continue
}
if _, ok := msg.p.(*packets.DisconnectPacket); ok {
msg.t.(*DisconnectToken).flowComplete()
DEBUG.Println(NET, "outbound wrote disconnect, closing connection")
// As per the MQTT spec "After sending a DISCONNECT Packet the Client MUST close the Network Connection"
// Closing the connection will cause the goroutines to end in sequence (starting with incoming comms)
conn.Close()
}
case msg, ok := <-oboundFromIncoming: // message triggered by an inbound message (PubrecPacket or PubrelPacket)
if !ok {
oboundFromIncoming = nil
continue
}
DEBUG.Println(NET, "obound from incoming msg to write, type", reflect.TypeOf(msg.p), " ID ", msg.p.Details().MessageID)
if err := msg.p.Write(conn); err != nil {
ERROR.Println(NET, "outgoing oboundFromIncoming reporting error", err)
if msg.t != nil {
msg.t.setError(err)
}
errChan <- err
continue
}
}
c.UpdateLastSent() // Record that a packet has been received (for keepalive routine)
}
}()
return errChan
}
订阅代码你特码到底在哪？
虽然找不到了，但是可以推断订阅一定是读取数据报文，我们可以去协议那里，看哪里用到了解析报文
读取报文的地方
查找是订阅的读取报文
// startIncoming initiates a goroutine that reads incoming messages off the wire and sends them to the channel (returned).
// If there are any issues with the network connection then the returned channel will be closed and the goroutine will exit
// (so closing the connection will terminate the goroutine)
func startIncoming(conn io.Reader) <-chan inbound {
var err error
var cp packets.ControlPacket
ibound := make(chan inbound)
DEBUG.Println(NET, "incoming started")
go func() {
for {
if cp, err = packets.ReadPacket(conn); err != nil {
// We do not want to log the error if it is due to the network connection having been closed
// elsewhere (i.e. after sending DisconnectPacket). Detecting this situation is the subject of
// https://github.com/golang/go/issues/4373
if !strings.Contains(err.Error(), closedNetConnErrorText) {
ibound <- inbound{err: err}
}
close(ibound)
DEBUG.Println(NET, "incoming complete")
return
}
DEBUG.Println(NET, "startIncoming Received Message")
ibound <- inbound{cp: cp}
}
}()
return ibound
}
再往上返回
当初确实我眼睛瞎了，没看到订阅读取的地方，全在startComms了
func startIncomingComms(conn io.Reader,
c commsFns,
inboundFromStore <-chan packets.ControlPacket,
) <-chan incomingComms {
// 我日尼玛，这特吗我以为是返回参数
ibound := startIncoming(conn) // Start goroutine that reads from network connection
output := make(chan incomingComms)