开源包mqtt源码_Connect
知识点:
options
包头
官方最简单的代码示例
/*
* Copyright (c) 2021 IBM Corp and others.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
* and Eclipse Distribution License v1.0 which accompany this distribution.
*
* The Eclipse Public License is available at
* https://www.eclipse.org/legal/epl-2.0/
* and the Eclipse Distribution License is available at
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* Contributors:
* Seth Hoenig
* Allan Stockdill-Mander
* Mike Robertson
*/
package main
import (
"fmt"
"log"
"os"
"time"
"github.com/eclipse/paho.mqtt.golang"
)
var f mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
fmt.Printf("TOPIC: %s
", msg.Topic())
fmt.Printf("MSG: %s
", msg.Payload())
}
func main() {
mqtt.DEBUG = log.New(os.Stdout, "", 0)
mqtt.ERROR = log.New(os.Stdout, "", 0)
opts := mqtt.NewClientOptions().AddBroker("tcp://iot.eclipse.org:1883").SetClientID("gotrivial")
opts.SetKeepAlive(2 * time.Second)
opts.SetDefaultPublishHandler(f)
opts.SetPingTimeout(1 * time.Second)
c := mqtt.NewClient(opts)
token := c.Connect()
if token.Wait() && token.Error() != nil {
panic(token.Error())
}
if token := c.Subscribe("go-mqtt/sample", 0, nil); token.Wait() && token.Error() != nil {
fmt.Println(token.Error())
os.Exit(1)
}
for i := 0; i < 5; i++ {
text := fmt.Sprintf("this is msg #%d!", i)
token := c.Publish("go-mqtt/sample", 0, false, text)
token.Wait()
}
time.Sleep(6 * time.Second)
if token := c.Unsubscribe("go-mqtt/sample"); token.Wait() && token.Error() != nil {
fmt.Println(token.Error())
os.Exit(1)
}
c.Disconnect(250)
time.Sleep(1 * time.Second)
}
分析
新建client解构体
options方式,但是有区别于之前的选项模式， 下面每个设置参数函数都会再次返回options，就可以一直用 . 下去
type ClientOptions struct {
Servers []*url.URL
ClientID string
Username string
Password string
CredentialsProvider CredentialsProvider
CleanSession bool
Order bool
WillEnabled bool
WillTopic string
WillPayload []byte
WillQos byte
WillRetained bool
ProtocolVersion uint
protocolVersionExplicit bool
TLSConfig *tls.Config
KeepAlive int64
PingTimeout time.Duration
ConnectTimeout time.Duration
MaxReconnectInterval time.Duration
AutoReconnect bool
ConnectRetryInterval time.Duration
ConnectRetry bool
Store Store
DefaultPublishHandler MessageHandler
OnConnect OnConnectHandler
OnConnectionLost ConnectionLostHandler
OnReconnecting ReconnectHandler
OnConnectAttempt ConnectionAttemptHandler
WriteTimeout time.Duration
MessageChannelDepth uint
ResumeSubs bool
HTTPHeaders http.Header
WebsocketOptions *WebsocketOptions
MaxResumePubInFlight int // // 0 = no limit; otherwise this is the maximum simultaneous messages sent while resuming
}
// NewClientOptions will create a new ClientClientOptions type with some
// default values.
// Port: 1883
// CleanSession: True
// Order: True (note: it is recommended that this be set to FALSE unless order is important)
// KeepAlive: 30 (seconds)
// ConnectTimeout: 30 (seconds)
// MaxReconnectInterval 10 (minutes)
// AutoReconnect: True
func NewClientOptions() *ClientOptions {
o := &ClientOptions{
Servers: nil,
ClientID: "",
Username: "",
Password: "",
CleanSession: true,
Order: true,
WillEnabled: false,
WillTopic: "",
WillPayload: nil,
WillQos: 0,
WillRetained: false,
ProtocolVersion: 0,
protocolVersionExplicit: false,
KeepAlive: 30,
PingTimeout: 10 * time.Second,
ConnectTimeout: 30 * time.Second,
MaxReconnectInterval: 10 * time.Minute,
AutoReconnect: true,
ConnectRetryInterval: 30 * time.Second,
ConnectRetry: false,
Store: nil,
OnConnect: nil,
OnConnectionLost: DefaultConnectionLostHandler,
OnConnectAttempt: nil,
WriteTimeout: 0, // 0 represents timeout disabled
ResumeSubs: false,
HTTPHeaders: make(map[string][]string),
WebsocketOptions: &WebsocketOptions{},
}
return o
}
// AddBroker adds a broker URI to the list of brokers to be used. The format should be
// scheme://host:port
// Where "scheme" is one of "tcp", "ssl", or "ws", "host" is the ip-address (or hostname)
// and "port" is the port on which the broker is accepting connections.
//
// Default values for hostname is "127.0.0.1", for schema is "tcp://".
//
// An example broker URI would look like: tcp://foobar.com:1883
func (o *ClientOptions) AddBroker(server string) *ClientOptions {
if len(server) > 0 && server[0] == ':' {
server = "127.0.0.1" + server
}
if !strings.Contains(server, "://") {
server = "tcp://" + server
}
brokerURI, err := url.Parse(server)
if err != nil {
ERROR.Println(CLI, "Failed to parse %q broker address: %s", server, err)
return o
}
o.Servers = append(o.Servers, brokerURI)
return o
}
// AddBroker adds a broker URI to the list of brokers to be used. The format should be
// scheme://host:port
// Where "scheme" is one of "tcp", "ssl", or "ws", "host" is the ip-address (or hostname)
// and "port" is the port on which the broker is accepting connections.
//
// Default values for hostname is "127.0.0.1", for schema is "tcp://".
//
// An example broker URI would look like: tcp://foobar.com:1883
func (o *ClientOptions) AddBroker(server string) *ClientOptions {
if len(server) > 0 && server[0] == ':' {
server = "127.0.0.1" + server
}
if !strings.Contains(server, "://") {
server = "tcp://" + server
}
brokerURI, err := url.Parse(server)
if err != nil {
ERROR.Println(CLI, "Failed to parse %q broker address: %s", server, err)
return o
}
o.Servers = append(o.Servers, brokerURI)
return o
}
关键代码 Connect()函数
从简单示例点下去，可以看到
// Connect will create a connection to the message broker, by default
// it will attempt to connect at v3.1.1 and auto retry at v3.1 if that
// fails
// Note: If using QOS1+ and CleanSession=false it is advisable to add
// routes (or a DefaultPublishHandler) prior to calling Connect()
// because queued messages may be delivered immediately post connection
func (c *client) Connect() Token {
// todo： 这个真尼玛是花活
t := newToken(packets.Connect).(*ConnectToken)
DEBUG.Println(CLI, "Connect()")
// 我感觉这段代码没啥用
if c.options.ConnectRetry && atomic.LoadUint32(&c.status) != disconnected {
// if in any state other than disconnected and ConnectRetry is
// enabled then the connection will come up automatically
// client can assume connection is up
WARN.Println(CLI, "Connect() called but not disconnected")
t.returnCode = packets.Accepted
t.flowComplete()
return t
}
c.persist.Open()
// 没有重试就不用看
if c.options.ConnectRetry {
c.reserveStoredPublishIDs() // Reserve IDs to allow publish before connect complete
}
c.setConnected(connecting) // 设置正在连接状态，基本上时间很短
go func() {
if len(c.options.Servers) == 0 {
t.setError(fmt.Errorf("no servers defined to connect to"))
return
}
RETRYCONN:
var conn net.Conn
var rc byte
var err error
conn, rc, t.sessionPresent, err = c.attemptConnection() // 尝试连接, 重点代码
if err != nil {
if c.options.ConnectRetry { // 简单没有重试，这段代码不用看
DEBUG.Println(CLI, "Connect failed, sleeping for", int(c.options.ConnectRetryInterval.Seconds()), "seconds and will then retry")
time.Sleep(c.options.ConnectRetryInterval)
if atomic.LoadUint32(&c.status) == connecting {
goto RETRYCONN
}
}
ERROR.Println(CLI, "Failed to connect to a broker")
c.setConnected(disconnected)
c.persist.Close() // 不知道这个仓库有啥用，就给关上了
t.returnCode = rc
t.setError(err)
return
}
// 下面是连接 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")
}()
return t
}
c.attemptConnection() 函数解析
// attemptConnection makes a single attempt to connect to each of the brokers
// the protocol version to use is passed in (as c.options.ProtocolVersion)
// Note: Does not set c.conn in order to minimise race conditions
// Returns:
// net.Conn - Connected network connection
// byte - Return code (packets.Accepted indicates a successful connection).
// bool - SessionPresent flag from the connect ack (only valid if packets.Accepted)
// err - Error (err != nil guarantees that conn has been set to active connection).
func (c *client) attemptConnection() (net.Conn, byte, bool, error) {
protocolVersion := c.options.ProtocolVersion
var (
sessionPresent bool
conn net.Conn
err error
rc byte
)
c.optionsMu.Lock() // Protect c.options.Servers so that servers can be added in test cases
brokers := c.options.Servers
c.optionsMu.Unlock()
for _, broker := range brokers {
// 从 options 获取一些用户名密码啥的， 返回 packets.ConnectPacket 连接的包头
cm := newConnectMsgFromOptions(&c.options, broker)
DEBUG.Println(CLI, "about to write new connect msg")
CONN:
// 垃圾代码
tlsCfg := c.options.TLSConfig // 这个用不上啊
if c.options.OnConnectAttempt != nil { // 尝试连接的 hook 函数
DEBUG.Println(CLI, "using custom onConnectAttempt handler...")
tlsCfg = c.options.OnConnectAttempt(broker, c.options.TLSConfig)
}
// ----
// 很多参数没用上，简单的代码连接仅仅是获取个 tcp 的连接
// Start by opening the network connection (tcp, tls, ws) etc
conn, err = openConnection(broker, tlsCfg, c.options.ConnectTimeout, c.options.HTTPHeaders, c.options.WebsocketOptions)
if err != nil {
ERROR.Println(CLI, err.Error())
WARN.Println(CLI, "failed to connect to broker, trying next")
rc = packets.ErrNetworkError
continue
}
DEBUG.Println(CLI, "socket connected to broker")
// 开始发送 cm 是连接包头
// Now we send the perform the MQTT connection handshake
rc, sessionPresent, err = connectMQTT(conn, cm, protocolVersion)
if rc == packets.Accepted {
break // successfully connected 成功连接，跳出去，注意，是跳出到 for
}
// We may be have to attempt the connection with MQTT 3.1
if conn != nil {
_ = conn.Close()
}
if !c.options.protocolVersionExplicit && protocolVersion == 4 { // try falling back to 3.1?
DEBUG.Println(CLI, "Trying reconnect using MQTT 3.1 protocol")
protocolVersion = 3
goto CONN
}
if c.options.protocolVersionExplicit { // to maintain logging from previous version
ERROR.Println(CLI, "Connecting to", broker, "CONNACK was not CONN_ACCEPTED, but rather", packets.ConnackReturnCodes[rc])
}
}
// If the connection was successful we set member variable and lock in the protocol version for future connection attempts (and users)
if rc == packets.Accepted {
c.options.ProtocolVersion = protocolVersion
c.options.protocolVersionExplicit = true
} else {
// Maintain same error format as used previously
if rc != packets.ErrNetworkError { // mqtt error
err = packets.ConnErrors[rc]
} else { // network error (if this occurred in ConnectMQTT then err will be nil)
err = fmt.Errorf("%s : %s", packets.ConnErrors[rc], err)
}
}
return conn, rc, sessionPresent, err
}
OnConnection() 代码
直接定位到tcp那里，再回到上面，就是建立了一个tcp连接
//
// This just establishes the network connection; once established the type of connection should be irrelevant
//
// openConnection opens a network connection using the protocol indicated in the URL.
// Does not carry out any MQTT specific handshakes.
func openConnection(uri *url.URL, tlsc *tls.Config, timeout time.Duration, headers http.Header, websocketOptions *WebsocketOptions) (net.Conn, error) {
switch uri.Scheme {
case "ws":
conn, err := NewWebsocket(uri.String(), nil, timeout, headers, websocketOptions)
return conn, err
case "wss":
conn, err := NewWebsocket(uri.String(), tlsc, timeout, headers, websocketOptions)
return conn, err
case "mqtt", "tcp":
// todo: 连接处代码
allProxy := os.Getenv("all_proxy")
if len(allProxy) == 0 { // 简单代码连接看这里
conn, err := net.DialTimeout("tcp", uri.Host, timeout)
if err != nil {
return nil, err
}
return conn, nil
}
proxyDialer := proxy.FromEnvironment()
conn, err := proxyDialer.Dial("tcp", uri.Host)
if err != nil {
return nil, err
}
return conn, nil
case "unix":
conn, err := net.DialTimeout("unix", uri.Host, timeout)
if err != nil {
return nil, err
}
return conn, nil
case "ssl", "tls", "mqtts", "mqtt+ssl", "tcps":
allProxy := os.Getenv("all_proxy")
if len(allProxy) == 0 {
conn, err := tls.DialWithDialer(&net.Dialer{Timeout: timeout}, "tcp", uri.Host, tlsc)
if err != nil {
return nil, err
}
return conn, nil
}
proxyDialer := proxy.FromEnvironment()
conn, err := proxyDialer.Dial("tcp", uri.Host)
if err != nil {
return nil, err
}
tlsConn := tls.Client(conn, tlsc)
err = tlsConn.Handshake()
if err != nil {
_ = conn.Close()
return nil, err
}
return tlsConn, nil
}
return nil, errors.New("unknown protocol")
}
回到client.go 继续看连接Mq发送的报文
// 开始发送 cm 是连接包头
// Now we send the perform the MQTT connection handshake
rc, sessionPresent, err = connectMQTT(conn, cm, protocolVersion)
if rc == packets.Accepted {
break // successfully connected 成功连接，跳出去，注意，是跳出到 for
}
ConnectMQTT() 函数
发送 Connect包头， 请求ack
// ConnectMQTT takes a connected net.Conn and performs the initial MQTT handshake. Parameters are:
// conn - Connected net.Conn
// cm - Connect Packet with everything other than the protocol name/version populated (historical reasons)
// protocolVersion - The protocol version to attempt to connect with
//
// Note that, for backward compatibility, ConnectMQTT() suppresses the actual connection error (compare to connectMQTT()).
func ConnectMQTT(conn net.Conn, cm *packets.ConnectPacket, protocolVersion uint) (byte, bool) {
rc, sessionPresent, _ := connectMQTT(conn, cm, protocolVersion)
return rc, sessionPresent
}
func connectMQTT(conn io.ReadWriter, cm *packets.ConnectPacket, protocolVersion uint) (byte, bool, error) {
switch protocolVersion {
case 3:
DEBUG.Println(CLI, "Using MQTT 3.1 protocol")
cm.ProtocolName = "MQIsdp"
cm.ProtocolVersion = 3
case 0x83:
DEBUG.Println(CLI, "Using MQTT 3.1b protocol")
cm.ProtocolName = "MQIsdp"
cm.ProtocolVersion = 0x83
case 0x84:
DEBUG.Println(CLI, "Using MQTT 3.1.1b protocol")
cm.ProtocolName = "MQTT"
cm.ProtocolVersion = 0x84
default:
DEBUG.Println(CLI, "Using MQTT 3.1.1 protocol")
cm.ProtocolName = "MQTT"
cm.ProtocolVersion = 4
}
// todo: 握手建立连接代码
// 我草，这里牛逼，里面的 写 io ，不仅可以调试本地的 buffer 也可以去写连接，牛逼
if err := cm.Write(conn); err != nil {
ERROR.Println(CLI, err)
return packets.ErrNetworkError, false, err
}
rc, sessionPresent, err := verifyCONNACK(conn)
return rc, sessionPresent, err
}