|
|
|
@ -30,22 +30,26 @@ func (the *MqttHelper) reConn2Subscribe(client mqtt.Client) { |
|
|
|
the.Subscribe(call.topic, call.f) |
|
|
|
} |
|
|
|
} |
|
|
|
func (the *MqttHelper) initialClient(sslEnable bool, caPath string) { |
|
|
|
func (the *MqttHelper) initialClient(sslEnable bool, caCertPath, clientCertPath, clientKeyPath string) { |
|
|
|
maxReConnCount := 3 |
|
|
|
ReConnDurationSec := 30 |
|
|
|
reConn: |
|
|
|
mqttConnectStr := fmt.Sprintf("tcp://%v:%d", the.Host, the.Port) |
|
|
|
tag := "tcp" |
|
|
|
if sslEnable { |
|
|
|
tag = "ssl" |
|
|
|
} |
|
|
|
mqttConnectStr := fmt.Sprintf("%s://%v:%d", tag, the.Host, the.Port) |
|
|
|
opts := mqtt.NewClientOptions().AddBroker(mqttConnectStr) |
|
|
|
opts.SetUsername(the.UserName) |
|
|
|
opts.SetPassword(the.Password) |
|
|
|
opts.SetClientID(the.ClientId) |
|
|
|
opts.SetOnConnectHandler(the.reConn2Subscribe) |
|
|
|
if sslEnable { |
|
|
|
opts.SetTLSConfig(NewTlsConfig(caPath)) |
|
|
|
opts.SetTLSConfig(NewTlsConfig(caCertPath, clientCertPath, clientKeyPath)) |
|
|
|
} |
|
|
|
the.client = mqtt.NewClient(opts) |
|
|
|
if token := the.client.Connect(); token.Wait() && token.Error() != nil { |
|
|
|
log.Printf("mqtt连接状态异常 %v(u:%v,p:%v,cid:%v) [err=%s]", mqttConnectStr, the.UserName, the.Password, the.ClientId, token.Error()) |
|
|
|
log.Printf("mqtt连接状态异常 %v [ u:%v,p:%v,cid:%v ] [err=%s]", mqttConnectStr, the.UserName, the.Password, the.ClientId, token.Error()) |
|
|
|
log.Printf("mqtt重连,%ds后尝试,剩余次数=%d", ReConnDurationSec, maxReConnCount) |
|
|
|
if maxReConnCount > 0 { |
|
|
|
maxReConnCount-- |
|
|
|
@ -59,26 +63,36 @@ reConn: |
|
|
|
|
|
|
|
} |
|
|
|
func (the *MqttHelper) Initial() { |
|
|
|
the.initialClient(false, "") |
|
|
|
the.initialClient(false, "", "", "") |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
func (the *MqttHelper) InitialWithSSL(caPath string) { |
|
|
|
the.initialClient(true, caPath) |
|
|
|
func (the *MqttHelper) InitialWithSSL(caCertPath, clientCertPath, clientKeyPath string) { |
|
|
|
the.initialClient(true, caCertPath, clientCertPath, clientKeyPath) |
|
|
|
} |
|
|
|
|
|
|
|
func NewTlsConfig(sslPath string) *tls.Config { |
|
|
|
func NewTlsConfig(caCertPath, clientCertPath, clientKeyPath string) *tls.Config { |
|
|
|
//"ssl/centerCA.crt"
|
|
|
|
certpool := x509.NewCertPool() |
|
|
|
ca, err := os.ReadFile(sslPath) |
|
|
|
certPool := x509.NewCertPool() |
|
|
|
ca, err := os.ReadFile(caCertPath) |
|
|
|
if err != nil { |
|
|
|
log.Fatalln(err.Error()) |
|
|
|
} |
|
|
|
certpool.AppendCertsFromPEM(ca) |
|
|
|
return &tls.Config{ |
|
|
|
//RootCAs: certpool,
|
|
|
|
certPool.AppendCertsFromPEM(ca) |
|
|
|
|
|
|
|
tlsConfig := &tls.Config{ |
|
|
|
RootCAs: certPool, |
|
|
|
InsecureSkipVerify: true, |
|
|
|
} |
|
|
|
if len(clientCertPath) > 0 && len(clientKeyPath) > 0 { |
|
|
|
// 读取客户端证书和密钥
|
|
|
|
clientCert, err := tls.LoadX509KeyPair(clientCertPath, clientKeyPath) |
|
|
|
if err != nil { |
|
|
|
fmt.Printf("Error loading client certificate and key: %v\n", err) |
|
|
|
} |
|
|
|
tlsConfig.Certificates = append(tlsConfig.Certificates, clientCert) |
|
|
|
} |
|
|
|
return tlsConfig |
|
|
|
} |
|
|
|
|
|
|
|
func (the *MqttHelper) Publish(topic string, messageBytes []byte) { |
|
|
|
@ -123,7 +137,15 @@ func MqttInitial(host string, port int, clientId string, userName string, passwo |
|
|
|
} |
|
|
|
if isSSL && len(caPtah) > 0 { |
|
|
|
log.Println("SSL mqHelpers初始化") |
|
|
|
mqHelpers.InitialWithSSL(caPtah[0]) |
|
|
|
switch len(caPtah) { |
|
|
|
case 1: |
|
|
|
mqHelpers.InitialWithSSL(caPtah[0], "", "") |
|
|
|
case 3: |
|
|
|
mqHelpers.InitialWithSSL(caPtah[0], caPtah[1], caPtah[2]) |
|
|
|
default: |
|
|
|
log.Printf("caPtah 参数量错误，请注意") |
|
|
|
} |
|
|
|
|
|
|
|
} else { |
|
|
|
mqHelpers.Initial() |
|
|
|
} |
|
|
|
|
