Golang-Paho MQTT订阅者在订阅者重新初始化后保持断开连接并出现错误EOF
Golang-Paho MQTT Subscriber keeps disconnecting with error EOF after reinitialization of subscriber
我正在尝试动态更改 mqtt 客户端处理程序和证书，这会在订阅者和发布者连接时导致订阅者 EOF
这就是我想要做的，
1] 我正在初始化 subscriber/publisher（使用 firstPubHandler、firstConnectHandler 和默认证书）
2] 正在使用发布者在服务器上发送注册消息以获取新证书详细信息
3] 服务器将返回证书详细信息，该响应将由主题为 .../id/Certificate[=45 的 firstConnectHandler 处理=] 下载证书。
4] firstPubHandler 将处理服务器的响应并重新初始化 publisher/subscriber（ 使用 messagePubHandler、connectHandler 和新下载的证书），connectHandler 将监听所有主题 /id/+
一切正常，除了 当我重新初始化 subscriber/publisher 时，订阅者不断断开连接并出现错误“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
}
基于对您的代码的快速审查，我相信这是正在发生的事情（因为您没有提供所有代码，所以涉及一些猜测）：
main() 调用 InitializeBroker 创建两个到代理的连接。默认发布处理程序设置为 firstPubHandler 并且在 OnConnect 处理程序中您订阅 SubClientId+"/id/Certificate- 当收到消息时 (
firstPubHandler)，您从消息中获取一个证书，并使用它使用相同的客户端 ID 但不同 OnConnect/ 来建立到代理的一组新连接默认发布处理程序。
所以在第 2 点之后，您实际上有两组独立的代理连接（总共 4 个连接）。但是 MQTT-3.1.4-2（参见 spec）指出：
If the ClientId represents a Client already connected to the Server then the Server MUST disconnect the existing Client.
因此当第二组连接建立时，代理将丢弃第一组连接。这是您看到的 'EOF' 断开连接。第二组连接仍将打开。由于您对第一组和第二组连接使用相同的 connectLostHandler，因此您无法在日志中看到正在终止哪个连接。
总而言之，我相信您的代码确实有效。但是，您可能应该在 firstConnectHandler 中调用 c.Disconnect()，以便在建立第二组连接之前完全关闭初始连接。您还需要将 publisher 存储在某处，以便您可以同时断开该连接。
注意：我很难理解您为什么要这样做。建立初始连接以检索证书似乎会降低系统的整体安全性。标准方法是给每个客户一个唯一的证书，然后在代理上使用 ACL 来应用任何必要的限制。对于许多代理，您可以在 ACL 中使用证书公用名（从而消除了对第二个连接的需要）。
我正在尝试动态更改 mqtt 客户端处理程序和证书，这会在订阅者和发布者连接时导致订阅者 EOF
这就是我想要做的，
1] 我正在初始化 subscriber/publisher（使用 firstPubHandler、firstConnectHandler 和默认证书）
2] 正在使用发布者在服务器上发送注册消息以获取新证书详细信息
3] 服务器将返回证书详细信息，该响应将由主题为 .../id/Certificate[=45 的 firstConnectHandler 处理=] 下载证书。
4] firstPubHandler 将处理服务器的响应并重新初始化 publisher/subscriber（ 使用 messagePubHandler、connectHandler 和新下载的证书），connectHandler 将监听所有主题 /id/+
一切正常，除了 当我重新初始化 subscriber/publisher 时，订阅者不断断开连接并出现错误“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
}
基于对您的代码的快速审查，我相信这是正在发生的事情（因为您没有提供所有代码，所以涉及一些猜测）：
main()调用InitializeBroker创建两个到代理的连接。默认发布处理程序设置为firstPubHandler并且在OnConnect处理程序中您订阅SubClientId+"/id/Certificate- 当收到消息时 (
firstPubHandler)，您从消息中获取一个证书，并使用它使用相同的客户端 ID 但不同OnConnect/ 来建立到代理的一组新连接默认发布处理程序。
所以在第 2 点之后，您实际上有两组独立的代理连接（总共 4 个连接）。但是 MQTT-3.1.4-2（参见 spec）指出：
If the ClientId represents a Client already connected to the Server then the Server MUST disconnect the existing Client.
因此当第二组连接建立时，代理将丢弃第一组连接。这是您看到的 'EOF' 断开连接。第二组连接仍将打开。由于您对第一组和第二组连接使用相同的 connectLostHandler，因此您无法在日志中看到正在终止哪个连接。
总而言之，我相信您的代码确实有效。但是，您可能应该在 firstConnectHandler 中调用 c.Disconnect()，以便在建立第二组连接之前完全关闭初始连接。您还需要将 publisher 存储在某处，以便您可以同时断开该连接。
注意：我很难理解您为什么要这样做。建立初始连接以检索证书似乎会降低系统的整体安全性。标准方法是给每个客户一个唯一的证书，然后在代理上使用 ACL 来应用任何必要的限制。对于许多代理，您可以在 ACL 中使用证书公用名（从而消除了对第二个连接的需要）。