mirror of
https://github.com/kaspanet/kaspad.git
synced 2025-06-14 10:06:40 +00:00
532e57b61c
* [NOD-427] Send notifications to `dag/selected-tip` * [NOD-442] Add selected tip notification * [NOD-427] Add comment to PublishSelectedTipNotification * [NOD-427] Remove redundant argument from errors.Wrapf * [NOD-427] Add handleBlockAddedMsg function * [NOD-427] Return errors instead of panicking * [NOD-427] Fix findHashOfBluestBlock to use []string instead of dbmodels.Block * [NOD-427] Add constants * [NOD-427] use path.Join instead of topic+address * [NOD-427] Remove redundant select * [NOD-427] Change break to return
75 lines
1.6 KiB
Go
75 lines
1.6 KiB
Go
package mqtt
|
|
|
|
import (
|
|
"encoding/json"
|
|
"github.com/daglabs/btcd/apiserver/config"
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
"github.com/pkg/errors"
|
|
)
|
|
|
|
// client is an instance of the MQTT client, in case we have an active connection
|
|
var client mqtt.Client
|
|
|
|
const (
|
|
qualityOfService = 2
|
|
quiesceMilliseconds = 250
|
|
)
|
|
|
|
// GetClient returns an instance of the MQTT client, in case we have an active connection
|
|
func GetClient() (mqtt.Client, error) {
|
|
if client == nil {
|
|
return nil, errors.New("MQTT is not connected")
|
|
}
|
|
return client, nil
|
|
}
|
|
|
|
func isConnected() bool {
|
|
return client != nil
|
|
}
|
|
|
|
// Connect initiates a connection to the MQTT server, if defined
|
|
func Connect() error {
|
|
cfg := config.ActiveConfig()
|
|
if cfg.MQTTBrokerAddress == "" {
|
|
// MQTT broker not defined -- nothing to do
|
|
return nil
|
|
}
|
|
|
|
options := mqtt.NewClientOptions()
|
|
options.AddBroker(cfg.MQTTBrokerAddress)
|
|
options.SetUsername(cfg.MQTTUser)
|
|
options.SetPassword(cfg.MQTTPassword)
|
|
options.SetAutoReconnect(true)
|
|
|
|
newClient := mqtt.NewClient(options)
|
|
if token := newClient.Connect(); token.Wait() && token.Error() != nil {
|
|
return token.Error()
|
|
}
|
|
client = newClient
|
|
|
|
return nil
|
|
}
|
|
|
|
// Close closes the connection to the MQTT server, if previously connected
|
|
func Close() {
|
|
if client == nil {
|
|
return
|
|
}
|
|
client.Disconnect(quiesceMilliseconds)
|
|
client = nil
|
|
}
|
|
|
|
func publish(topic string, data interface{}) error {
|
|
payload, err := json.Marshal(data)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
token := client.Publish(topic, qualityOfService, false, payload)
|
|
token.Wait()
|
|
if token.Error() != nil {
|
|
return errors.WithStack(token.Error())
|
|
}
|
|
return nil
|
|
}
|