91 lines
2.1 KiB
Go
91 lines
2.1 KiB
Go
package mqtt
|
|
|
|
import (
|
|
"fmt"
|
|
"time"
|
|
|
|
"git.piskot.si/SeminarM2/lambdaiot-core/internal/config"
|
|
paho "github.com/eclipse/paho.mqtt.golang"
|
|
)
|
|
|
|
type Client struct {
|
|
client paho.Client
|
|
}
|
|
|
|
// Default is a package-level client used by helpers
|
|
var Default *Client
|
|
|
|
// SetDefault sets the package default client
|
|
func SetDefault(c *Client) {
|
|
Default = c
|
|
}
|
|
|
|
// PublishDefault publishes using the default client
|
|
func PublishDefault(topic string, payload []byte) error {
|
|
if Default == nil {
|
|
return fmt.Errorf("no default mqtt client")
|
|
}
|
|
return Default.Publish(topic, payload)
|
|
}
|
|
|
|
func Connect(cfg config.MQTTConfig) (*Client, error) {
|
|
opts := paho.NewClientOptions()
|
|
opts.AddBroker(cfg.Broker)
|
|
if cfg.ClientID != "" {
|
|
opts.SetClientID(cfg.ClientID)
|
|
}
|
|
if cfg.Username != "" {
|
|
opts.SetUsername(cfg.Username)
|
|
opts.SetPassword(cfg.Password)
|
|
}
|
|
opts.SetAutoReconnect(true)
|
|
opts.SetConnectRetry(true)
|
|
opts.SetConnectTimeout(5 * time.Second)
|
|
|
|
c := paho.NewClient(opts)
|
|
token := c.Connect()
|
|
if ok := token.WaitTimeout(10 * time.Second); !ok {
|
|
return nil, fmt.Errorf("mqtt connect timeout")
|
|
}
|
|
if err := token.Error(); err != nil {
|
|
return nil, err
|
|
}
|
|
return &Client{client: c}, nil
|
|
}
|
|
|
|
func (c *Client) Publish(topic string, payload []byte) error {
|
|
if c == nil || c.client == nil {
|
|
return fmt.Errorf("mqtt client not connected")
|
|
}
|
|
token := c.client.Publish(topic, 0, false, payload)
|
|
if ok := token.WaitTimeout(5 * time.Second); !ok {
|
|
return fmt.Errorf("publish timeout")
|
|
}
|
|
return token.Error()
|
|
}
|
|
|
|
// Subscribe subscribes to the given topic and calls the provided handler
|
|
// for each incoming message.
|
|
func (c *Client) Subscribe(topic string, handler func(topic string, payload []byte)) error {
|
|
if c == nil || c.client == nil {
|
|
return fmt.Errorf("mqtt client not connected")
|
|
}
|
|
mh := func(cli paho.Client, msg paho.Message) {
|
|
if handler != nil {
|
|
handler(msg.Topic(), msg.Payload())
|
|
}
|
|
}
|
|
token := c.client.Subscribe(topic, 0, mh)
|
|
if ok := token.WaitTimeout(5 * time.Second); !ok {
|
|
return fmt.Errorf("subscribe timeout")
|
|
}
|
|
return token.Error()
|
|
}
|
|
|
|
func (c *Client) Close() {
|
|
if c == nil || c.client == nil {
|
|
return
|
|
}
|
|
c.client.Disconnect(250)
|
|
}
|