opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

1

Browse source
This commit is contained in:
hehaijunandhenry 2026-05-08 14:48:43 +08:00
parent e7c0dc821a
commit f062cb41d7
3 changed files with 30 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
pkg/channels/manager_channel.go
View file

@ -102,6 +102,11 @@ func hiddenValues(key string, value map[string]any, ch *config.Channel) {
} }
} }
value["webhooks"] = webhooks value["webhooks"] = webhooks
case "mqtt":
if settings, ok := v.(*config.MQTTSettings); ok {
value["username"] = settings.Username.String()
value["password"] = settings.Password.String()
}
} }
} }

21
pkg/channels/mqtt/mqtt.go
View file

@ -8,6 +8,7 @@ import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"strings" "strings"
"sync"
"time" "time"
pahomqtt "github.com/eclipse/paho.mqtt.golang" pahomqtt "github.com/eclipse/paho.mqtt.golang"
@ -86,9 +87,13 @@ func (c *MQTTChannel) Start(ctx context.Context) error {
opts.SetPassword(c.cfg.Password.String()) opts.SetPassword(c.cfg.Password.String())
} }
firstSubscribe := make(chan error, 1)
var once sync.Once
opts.SetOnConnectHandler(func(client pahomqtt.Client) { opts.SetOnConnectHandler(func(client pahomqtt.Client) {
logger.InfoC("mqtt", "MQTT connected, subscribing to inbound topic") logger.InfoC("mqtt", "MQTT connected, subscribing to inbound topic")
c.subscribe(client) err := c.subscribe(client)
once.Do(func() { firstSubscribe <- err })
}) })
opts.SetConnectionLostHandler(func(_ pahomqtt.Client, err error) { opts.SetConnectionLostHandler(func(_ pahomqtt.Client, err error) {
@ -98,12 +103,19 @@ func (c *MQTTChannel) Start(ctx context.Context) error {
client := pahomqtt.NewClient(opts) client := pahomqtt.NewClient(opts)
token := client.Connect() token := client.Connect()
if !token.WaitTimeout(10 * time.Second) { if !token.WaitTimeout(10 * time.Second) {
client.Disconnect(250)
return fmt.Errorf("mqtt connect timed out after 10s (broker: %s)", c.cfg.Broker) return fmt.Errorf("mqtt connect timed out after 10s (broker: %s)", c.cfg.Broker)
} }
if err := token.Error(); err != nil { if err := token.Error(); err != nil {
client.Disconnect(250)
return fmt.Errorf("mqtt connect failed: %w", err) return fmt.Errorf("mqtt connect failed: %w", err)
} }
if err := <-firstSubscribe; err != nil {
client.Disconnect(250)
return fmt.Errorf("mqtt subscribe failed: %w", err)
}
c.client = client c.client = client
c.SetRunning(true) c.SetRunning(true)
@ -144,7 +156,7 @@ func (c *MQTTChannel) clientIDFromTopic(topic string) (string, bool) {
} }
// subscribe subscribes to the inbound topic for this agent. // subscribe subscribes to the inbound topic for this agent.
func (c *MQTTChannel) subscribe(client pahomqtt.Client) { func (c *MQTTChannel) subscribe(client pahomqtt.Client) error {
topic := fmt.Sprintf("%s/%s/+/request", c.topicPrefix(), c.cfg.AgentID) topic := fmt.Sprintf("%s/%s/+/request", c.topicPrefix(), c.cfg.AgentID)
token := client.Subscribe(topic, c.qos, func(_ pahomqtt.Client, msg pahomqtt.Message) { token := client.Subscribe(topic, c.qos, func(_ pahomqtt.Client, msg pahomqtt.Message) {
c.handleInbound(msg) c.handleInbound(msg)
@ -155,9 +167,10 @@ func (c *MQTTChannel) subscribe(client pahomqtt.Client) {
"topic": topic, "topic": topic,
"error": err.Error(), "error": err.Error(),
}) })
} else { return err
logger.InfoCF("mqtt", "Subscribed to inbound topic", map[string]any{"topic": topic})
} }
logger.InfoCF("mqtt", "Subscribed to inbound topic", map[string]any{"topic": topic})
return nil
} }
// handleInbound processes an inbound MQTT message. // handleInbound processes an inbound MQTT message.