Golang-Paho MQTT 订阅者在重新初始化订阅者后不断断开连接并出现错误 EOF

Question

我正在尝试动态更改 mqtt 客户端处理程序和证书，当订阅者和发布者连接时，这会导致订阅者 EOF

这就是我想要做的，

1] 我正在初始化订阅者/发布者（使用 firstPubHandler、firstConnectHandler 和默认证书）

2] 使用发布者在服务器上发送注册消息以获取新证书详细信息

3] 服务器将返回证书详细信息，该响应将由主题 .../id/Certificate 上的 firstConnectHandler 处理以下载证书。

4] firstPubHandler 将处理服务器的响应并重新初始化发布者/订阅者（使用 messagePubHandler、connectHandler 和新下载的证书），connectHandler 将侦听所有主题 /id/+

一切正常，除了当我重新初始化订阅者/发布者时，订阅者不断断开连接并出现错误“EOF”

我在这里做错了吗？或者有没有更好的方法来做到这一点？ 任何帮助表示赞赏

-- 主函数

var opt Params
var publisher mqtt.Client
var subscriber mqtt.Client

func main() {
    InitializeBroker(firstPubHandler, firstConnectHandler)

    //Ultimately it will trigger message on ".../id/Certificate" topic which will be handled byfirstConnectHandler 
    PublishRegistrationMessage(publisher)

    c := make(chan os.Signal, 1)
    signal.Notify(c, os.Interrupt, syscall.SIGTERM)
    done := make(chan os.Signal, 1)
    signal.Notify(done, os.Interrupt, syscall.SIGTERM)
    go func() {
        for {
        }
    }()
    <-done
    <-c
    DisconnectBrocker()
}

-- 处理程序

// First handlers
var firstPubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
    DownloadCertificates(msg.Payload())
    InitializeBroker(messagePubHandler, connectHandler)
}

var firstConnectHandler mqtt.OnConnectHandler = func(c mqtt.Client) {
    if token := c.Subscribe(opt.SubClientId+"/id/Certificate", 0, firstPubHandler); token.Wait() && token.Error() != nil {
        log.Error(token.Error())
    }
}

// Second handlers
var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
    ProcessMessage(msg.Payload())
}

var connectHandler mqtt.OnConnectHandler = func(c mqtt.Client) {
    if token := c.Subscribe(opt.SubClientId+"/id/+", 0, messagePubHandler); token.Wait() && token.Error() != nil {
        log.Error(token.Error())
    }
}

// Common handler
var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
    log.Info(err)
}

-- Mqtt 代理初始化

func InitializeBroker(lMessageHandler mqtt.MessageHandler, lConnectHandler mqtt.OnConnectHandler) {
    statusPublishTopic := opt.PubClientId/id
    nodeSubscribeTopic := opt.SubClientId/id

    // Build the options for the publish client
    publisherOptions := mqtt.NewClientOptions()
    publisherOptions.AddBroker(opt.Broker)
    publisherOptions.SetClientID(statusPublishTopic)
    publisherOptions.SetDefaultPublishHandler(lMessageHandler)
    publisherOptions.OnConnectionLost = connectLostHandler

    // Build the options for the subscribe client
    subscriberOptions := mqtt.NewClientOptions()
    subscriberOptions.AddBroker(opt.Broker)
    subscriberOptions.SetClientID(nodeSubscribeTopic)
    subscriberOptions.SetDefaultPublishHandler(lMessageHandler)
    subscriberOptions.OnConnectionLost = connectLostHandler
    subscriberOptions.OnConnect = lConnectHandler

    if !opt.NoTLS {
        tlsconfig, err := NewTLSConfig()
        if err != nil {
            log.Fatalf(err)
        }

        subscriberOptions.SetTLSConfig(tlsconfig)
        publisherOptions.SetTLSConfig(tlsconfig)
    }

    publisher = mqtt.NewClient(publisherOptions)
    if token := publisher.Connect(); token.Wait() && token.Error() != nil {
        log.Fatalf(token.Error())
    }

    subscriber = mqtt.NewClient(subscriberOptions)
    if token := subscriber.Connect(); token.Wait() && token.Error() != nil {
        log.Fatalf(token.Error())
    }
}

func NewTLSConfig() (config *tls.Config, err error) {
    certpool := x509.NewCertPool()
    pemCerts, err := ioutil.ReadFile(rootCert)
    if err != nil {
        return nil, err
    }
    certpool.AppendCertsFromPEM(pemCerts)

    cert, err := tls.LoadX509KeyPair(nodeCertFilePath, pvtKeyFilePath)
    if err != nil {
        return nil, err
    }

    config = &tls.Config{
        RootCAs:      certpool,
        ClientAuth:   tls.NoClientCert,
        ClientCAs:    nil,
        Certificates: []tls.Certificate{cert},
    }
    return config, nil
}

Answer 1

基于对您代码的快速审查，我认为这是正在发生的事情（因为您没有提供所有代码，所以涉及一些猜测）：

1. main() 调用 InitializeBroker 创建两个到代理的连接。默认发布处理程序设置为firstPubHandler，在OnConnect 处理程序中您订阅SubClientId+"/id/Certificate
2. 当收到消息 (firstPubHandler) 时，您会从消息中获取一个证书，并使用它与使用相同客户端 ID 但不同 OnConnect/默认发布处理程序的代理建立一组新连接。李>

所以在第 2 点之后，您实际上有两组独立的代理连接（总共 4 个连接）。然而MQTT-3.1.4-2（见spec）声明：

如果 ClientId 代表一个已经连接到服务器的客户端，那么服务器必须断开现有的客户端。

因此，当建立第二组连接时，代理将丢弃第一组连接。这是您看到的“EOF”断开连接。第二组连接仍将启动。由于您对第一组和第二组连接使用相同的connectLostHandler，因此您无法在日志中看到哪个连接被终止。

总之，我相信您的代码确实有效。但是，您可能应该在firstConnectHandler 中调用c.Disconnect()，以便在建立第二组连接之前完全关闭初始连接。您还需要将publisher 存储在某处，以便同时断开该连接。

注意：我很难理解您为什么要这样做。建立初始连接以检索证书似乎会降低系统的整体安全性。标准方法是给每个客户端一个唯一的证书，然后使用代理上的 ACL 来应用任何必要的限制。对于许多代理，您可以在 ACL 中使用证书公用名（从而无需第二次连接）。