diff --git a/go.mod b/go.mod index ebb57b0..f414092 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,13 @@ module github.com/jwetzell/showbridge-go go 1.25.1 require ( + github.com/eclipse/paho.mqtt.golang v1.5.1 github.com/jwetzell/osc-go v0.0.0-20251114203632-24077a77d6c7 github.com/urfave/cli/v3 v3.6.0 ) + +require ( + github.com/gorilla/websocket v1.5.3 // indirect + golang.org/x/net v0.44.0 // indirect + golang.org/x/sync v0.17.0 // indirect +) diff --git a/go.sum b/go.sum index 313b4b1..49a4f68 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,9 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE= +github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/jwetzell/osc-go v0.0.0-20251114203632-24077a77d6c7 h1:vR4ooQd95vO8pdCugY0Kg7/MSKvuJc0pkHUZlLf6AtM= github.com/jwetzell/osc-go v0.0.0-20251114203632-24077a77d6c7/go.mod h1:mkPoLU72wmg9Wq6mh5P5RjsWFXqaUqq4n64EWqg121A= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= @@ -8,5 +12,9 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/urfave/cli/v3 v3.6.0 h1:oIdArVjkdIXHWg3iqxgmqwQGC8NM0JtdgwQAj2sRwFo= github.com/urfave/cli/v3 v3.6.0/go.mod h1:ysVLtOEmg2tOy6PknnYVhDoouyC/6N42TMeoMzskhso= +golang.org/x/net v0.44.0 h1:evd8IRDyfNBMBTTY5XRF1vaZlD+EmWx6x8PkhR04H/I= +golang.org/x/net v0.44.0/go.mod h1:ECOoLqd5U3Lhyeyo/QDCEVQ4sNgYsqvCZ722XogGieY= +golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= +golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/processing/mqtt-message-encode.go b/internal/processing/mqtt-message-encode.go new file mode 100644 index 0000000..0dbc6e0 --- /dev/null +++ b/internal/processing/mqtt-message-encode.go @@ -0,0 +1,35 @@ +package processing + +import ( + "context" + "fmt" + + mqtt "github.com/eclipse/paho.mqtt.golang" +) + +type MQTTMessageEncode struct { + config ProcessorConfig +} + +func (mme *MQTTMessageEncode) Process(ctx context.Context, payload any) (any, error) { + payloadMessage, ok := payload.(mqtt.Message) + + if !ok { + return nil, fmt.Errorf("mqtt.message.encode processor only accepts an mqtt.Message") + } + + return payloadMessage.Payload(), nil +} + +func (mme *MQTTMessageEncode) Type() string { + return mme.config.Type +} + +func init() { + RegisterProcessor(ProcessorRegistration{ + Type: "mqtt.message.encode", + New: func(config ProcessorConfig) (Processor, error) { + return &MQTTMessageEncode{config: config}, nil + }, + }) +} diff --git a/mqtt-client.go b/mqtt-client.go new file mode 100644 index 0000000..bcd685a --- /dev/null +++ b/mqtt-client.go @@ -0,0 +1,108 @@ +package showbridge + +import ( + "fmt" + "log/slog" + + mqtt "github.com/eclipse/paho.mqtt.golang" +) + +type MQTTClient struct { + config ModuleConfig + router *Router + Broker string + ClientID string + Topic string + client mqtt.Client +} + +func init() { + RegisterModule(ModuleRegistration{ + Type: "net.mqtt.client", + New: func(config ModuleConfig) (Module, error) { + params := config.Params + broker, ok := params["broker"] + + if !ok { + return nil, fmt.Errorf("net.mqtt.client requires a broker parameter") + } + + brokerString, ok := broker.(string) + + if !ok { + return nil, fmt.Errorf("net.mqtt.client host must be string") + } + + topic, ok := params["topic"] + + if !ok { + return nil, fmt.Errorf("net.mqtt.client requires a topic parameter") + } + + topicString, ok := topic.(string) + + if !ok { + return nil, fmt.Errorf("net.mqtt.client host must be string") + } + + clientId, ok := params["clientId"] + + if !ok { + return nil, fmt.Errorf("net.mqtt.client requires a clientId parameter") + } + + clientIdString, ok := clientId.(string) + + if !ok { + return nil, fmt.Errorf("net.mqtt.client host must be string") + } + + return &MQTTClient{config: config, Broker: brokerString, Topic: topicString, ClientID: clientIdString}, nil + }, + }) +} + +func (mc *MQTTClient) Id() string { + return mc.config.Id +} + +func (mc *MQTTClient) Type() string { + return mc.config.Type +} + +func (mc *MQTTClient) RegisterRouter(router *Router) { + mc.router = router +} + +func (mc *MQTTClient) Run() error { + opts := mqtt.NewClientOptions() + opts.AddBroker(mc.Broker) + opts.SetClientID(mc.ClientID) + opts.SetAutoReconnect(true) + opts.SetCleanSession(false) + + opts.OnConnect = func(c mqtt.Client) { + token := mc.client.Subscribe(mc.Topic, 1, func(c mqtt.Client, m mqtt.Message) { + mc.router.HandleInput(mc.config.Id, m) + }) + token.Wait() + } + + mc.client = mqtt.NewClient(opts) + + token := mc.client.Connect() + + token.Wait() + err := token.Error() + if err != nil { + return err + } + + <-mc.router.Context.Done() + slog.Debug("router context done in module", "id", mc.config.Id) + return nil +} + +func (mc *MQTTClient) Output(payload any) error { + return fmt.Errorf("net.mqtt.client output is not implemented") +}