Add MQTT source/destination

README.md

@@ -246,6 +246,29 @@ Read from stdin. Useful for testing or doing something you probably shouldn't.
 }
 ```
 
+### MQTT
+MQTT will listen on the supplied topic for new events.
+
+broker and clientID are required to receive data.
+clientID must be unique from any other mqtt destinations or sources
+If topic is not supplied, it will default to the wildcard `#`.
+
+Do not read events from the same topic that an MQTT destination is sending to otherwise kawa will create an infinite loop and eventually crash.
+
+```
+{
+  "type": "mqtt",
+  "broker": "mqtt://broker.mqtt:1883",
+  "clientID": "kawa_src",
+  "userName": "",
+  "password": "",
+  "topic": "kawa/src",
+
+  "qos": 1, // Optional defaults to 1 if not included
+  "retained": false, // Optional defaults to false if not included
+}
+```
+
 
 ## Destination Configuration
 
@@ -286,6 +309,27 @@ Printer will print the results to stdout. Useful for testing and development.
 }
 ```
 
+### MQTT
+MQTT will send events to the supplied topic.
+
+broker and clientID are required to send data.
+clientID must be unique from any other mqtt destinations or sources
+If topic is not supplied, it will default to the wildcard `#`.
+
+```
+{
+  "type": "mqtt",
+  "broker": "mqtt://broker.mqtt:1883",
+  "clientID": "kawa_dst",
+  "userName": "",
+  "password": "",
+  "topic": "kawa/dest",
+
+  "qos": 1, // Optional defaults to 1 if not included
+  "retained": false, // Optional defaults to false if not included
+}
+```
+
 ## Source / Destination Wishlist
 
  - Kafka

cmd/kawad/config.go

@@ -4,13 +4,16 @@ import (
 	"os"
 
 	"github.com/runreveal/kawa"
+	mqttDstkawad "github.com/runreveal/kawa/cmd/kawad/internal/destinations/mqtt"
 	"github.com/runreveal/kawa/cmd/kawad/internal/destinations/printer"
 	"github.com/runreveal/kawa/cmd/kawad/internal/destinations/runreveal"
 	s3kawad "github.com/runreveal/kawa/cmd/kawad/internal/destinations/s3"
 	"github.com/runreveal/kawa/cmd/kawad/internal/sources/journald"
+	mqttSrckawad "github.com/runreveal/kawa/cmd/kawad/internal/sources/mqtt"
 	"github.com/runreveal/kawa/cmd/kawad/internal/sources/scanner"
 	"github.com/runreveal/kawa/cmd/kawad/internal/sources/syslog"
 	"github.com/runreveal/kawa/cmd/kawad/internal/types"
+	"github.com/runreveal/kawa/x/mqtt"
 	"github.com/runreveal/kawa/x/s3"
 	"github.com/runreveal/lib/loader"
 	"golang.org/x/exp/slog"
@@ -30,6 +33,9 @@ func init() {
 	loader.Register("journald", func() loader.Builder[kawa.Source[types.Event]] {
 		return &JournaldConfig{}
 	})
+	loader.Register("mqtt", func() loader.Builder[kawa.Source[types.Event]] {
+		return &MQTTSrcConfig{}
+	})
 
 	loader.Register("printer", func() loader.Builder[kawa.Destination[types.Event]] {
 		return &PrinterConfig{}
@@ -40,6 +46,10 @@ func init() {
 	loader.Register("runreveal", func() loader.Builder[kawa.Destination[types.Event]] {
 		return &RunRevealConfig{}
 	})
+	loader.Register("mqtt", func() loader.Builder[kawa.Destination[types.Event]] {
+		return &MQTTDestConfig{}
+	})
+
 }
 
 type ScannerConfig struct {
@@ -114,3 +124,57 @@ func (c *JournaldConfig) Configure() (kawa.Source[types.Event], error) {
 	slog.Info("configuring journald")
 	return journald.New(), nil
 }
+
+type MQTTDestConfig struct {
+	Broker   string `json:"broker"`
+	ClientID string `json:"clientID"`
+	Topic    string `json:"topic"`
+
+	UserName string `json:"userName"`
+	Password string `json:"password"`
+
+	QOS      byte `json:"qos"`
+	Retained bool `json:"retained"`
+}
+
+func (c *MQTTDestConfig) Configure() (kawa.Destination[types.Event], error) {
+	slog.Info("configuring mqtt dest")
+	mqttDst := mqttDstkawad.NewMQTT(
+		mqtt.WithBroker(c.Broker),
+		mqtt.WithClientID(c.ClientID),
+		mqtt.WithQOS(c.QOS),
+		mqtt.WithTopic(c.Topic),
+		mqtt.WithRetained(c.Retained),
+		mqtt.WithUserName(c.UserName),
+		mqtt.WithPassword(c.Password),
+	)
+
+	return mqttDst, nil
+}
+
+type MQTTSrcConfig struct {
+	Broker   string `json:"broker"`
+	ClientID string `json:"clientID"`
+	Topic    string `json:"topic"`
+
+	UserName string `json:"userName"`
+	Password string `json:"password"`
+
+	QOS      byte `json:"qos"`
+	Retained bool `json:"retained"`
+}
+
+func (c *MQTTSrcConfig) Configure() (kawa.Source[types.Event], error) {
+	slog.Info("configuring mqtt src")
+	mqttSrc := mqttSrckawad.NewMQTT(
+		mqtt.WithBroker(c.Broker),
+		mqtt.WithClientID(c.ClientID),
+		mqtt.WithQOS(c.QOS),
+		mqtt.WithTopic(c.Topic),
+		mqtt.WithRetained(c.Retained),
+		mqtt.WithUserName(c.UserName),
+		mqtt.WithPassword(c.Password),
+	)
+
+	return mqttSrc, nil
+}

cmd/kawad/internal/destinations/mqtt/mqtt.go

@@ -0,0 +1,31 @@
+package mqttDstkawad
+
+import (
+	"context"
+
+	"github.com/runreveal/kawa"
+	"github.com/runreveal/kawa/cmd/kawad/internal/types"
+	"github.com/runreveal/kawa/x/mqtt"
+)
+
+type MQTT struct {
+	wrapped *mqtt.Destination
+}
+
+func NewMQTT(opts ...mqtt.OptFunc) *MQTT {
+	return &MQTT{wrapped: mqtt.NewDestination(opts...)}
+}
+
+func (p *MQTT) Run(ctx context.Context) error {
+	return p.wrapped.Run(ctx)
+}
+
+func (p *MQTT) Send(ctx context.Context, ack func(), msg ...kawa.Message[types.Event]) error {
+	for _, m := range msg {
+		err := p.wrapped.Send(ctx, ack, kawa.Message[[]byte]{Value: m.Value.RawLog})
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

cmd/kawad/internal/sources/mqtt/mqtt.go

@@ -0,0 +1,41 @@
+package mqttSrckawad
+
+import (
+	"context"
+	"time"
+
+	"github.com/runreveal/kawa"
+	"github.com/runreveal/kawa/cmd/kawad/internal/types"
+	"github.com/runreveal/kawa/x/mqtt"
+)
+
+type MQTT struct {
+	wrapped *mqtt.Source
+}
+
+func NewMQTT(opts ...mqtt.OptFunc) *MQTT {
+	return &MQTT{wrapped: mqtt.NewSource(opts...)}
+}
+
+func (s *MQTT) Run(ctx context.Context) error {
+	return s.wrapped.Run(ctx)
+}
+
+func (s *MQTT) Recv(ctx context.Context) (kawa.Message[types.Event], func(), error) {
+	msg, ack, err := s.wrapped.Recv(ctx)
+	if err != nil {
+		return kawa.Message[types.Event]{}, nil, ctx.Err()
+	}
+
+	eventMsg := kawa.Message[types.Event]{
+		Key: msg.Key,
+		Value: types.Event{
+			Timestamp:  time.Now(),
+			SourceType: "mqtt",
+			RawLog:     msg.Value,
+		}, Topic: msg.Topic,
+		Attributes: msg.Attributes,
+	}
+
+	return eventMsg, ack, err
+}

examples/mqtt_config.json

@@ -0,0 +1,32 @@
+{
+  // this is an example config file for kawa
+  // it is parsed using hujson so you can use comments and trailing commas, but
+  // is otherwise identical to JSON
+  "sources": [
+    {
+      "type": "mqtt",
+      "broker": "mqtt://broker.localhost:1883",
+      "clientID": "kawa_src",
+      "userName": "",
+      "password": "",
+      "topic": "kawad/src",
+      "qos": 1, // Optional defaults to 1 if not included
+      "retained": false, // Optional defaults to false if not included
+    },
+  ],
+  "destinations": [
+    {
+      "type": "mqtt",
+      "broker": "mqtt://broker.localhost:1883",
+      "clientID": "kawa_dst",
+      "userName": "",
+      "password": "",
+      "topic": "kawad/dest",
+      "qos": 1, // Optional defaults to 1 if not included
+      "retained": false, // Optional defaults to false if not included
+    },
+    {
+      "type": "printer"
+    }
+  ],
+}

go.mod

@@ -17,6 +17,8 @@ require (
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/eclipse/paho.mqtt.golang v1.4.3 // indirect
+	github.com/gorilla/websocket v1.5.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
@@ -27,7 +29,8 @@ require (
 	github.com/tidwall/gjson v1.14.4 // indirect
 	github.com/tidwall/match v1.1.1 // indirect
 	github.com/tidwall/pretty v1.2.0 // indirect
-	golang.org/x/net v0.7.0 // indirect
-	golang.org/x/sys v0.5.0 // indirect
+	golang.org/x/net v0.14.0 // indirect
+	golang.org/x/sync v0.3.0 // indirect
+	golang.org/x/sys v0.11.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -6,7 +6,11 @@ github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46t
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 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.4.3 h1:2kwcUGn8seMUfWndX0hGbvH8r7crgcJguQNCyp70xik=
+github.com/eclipse/paho.mqtt.golang v1.4.3/go.mod h1:CSYvoAlsMkhYOXh/oKyxa8EcBci6dVkLCbo5tTC1RIE=
 github.com/google/go-cmp v0.5.8 h1:e6P7q2lk1O+qJJb4BtCQXlK8vWEO8V1ZeuEdJNOqZyg=
+github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
+github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
 github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg=
@@ -55,8 +59,12 @@ golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug
 golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
 golang.org/x/net v0.7.0 h1:rJrUqqhjsgNp7KqAIc25s9pZnjU7TUcSY7HcVZjdn1g=
 golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
+golang.org/x/net v0.14.0 h1:BONx9s002vGdD9umnlX1Po8vOZmrgH34qlHcD1MfK14=
+golang.org/x/net v0.14.0/go.mod h1:PpSgVXXLK0OxS0F31C1/tv6XNguvCrnXIDrFMspZIUI=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.3.0 h1:ftCYgMx6zT/asHUrPw8BLLscYtGznsLAnjq5RH9P66E=
+golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -66,6 +74,8 @@ golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0 h1:MUK/U/4lj1t1oPg0HfuXDN/Z1wv31ZJ/YcPiGccS4DU=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.11.0 h1:eG7RXZHdqOJ1i+0lgLgCpSXAp6M3LYlAo6osgSi0xOM=
+golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
@@ -74,6 +84,7 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo=
+golang.org/x/text v0.12.0 h1:k+n5B8goJNdU7hSvEtMUz3d1Q6D/XW4COJSJR6fN0mc=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=