Add MQTT source/destination

README.md

@@ -117,6 +117,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
 ## RunReveal
@@ -152,6 +175,28 @@ 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
+  "batchSize": 100, // Optional defaults to 100 if not included
+}
+```
+
 # Source / Destination Wishlist
  - Kafka
  - redis

cmd/kawa/config.go

@@ -5,10 +5,12 @@ import (
 
 	"github.com/runreveal/kawa"
 	"github.com/runreveal/kawa/internal/destinations"
+	"github.com/runreveal/kawa/internal/destinations/mqtt"
 	"github.com/runreveal/kawa/internal/destinations/runreveal"
 	s3 "github.com/runreveal/kawa/internal/destinations/s3"
 	"github.com/runreveal/kawa/internal/sources"
 	"github.com/runreveal/kawa/internal/sources/journald"
+	mqttsrc "github.com/runreveal/kawa/internal/sources/mqtt"
 	"github.com/runreveal/kawa/internal/sources/syslog"
 	"github.com/runreveal/kawa/internal/types"
 	"github.com/runreveal/lib/loader"
@@ -29,6 +31,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{}
@@ -39,6 +44,9 @@ 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 &MQTTConfig{}
+	})
 }
 
 type ScannerConfig struct {
@@ -81,6 +89,59 @@ func (c *RunRevealConfig) Configure() (kawa.Destination[types.Event], error) {
 	), nil
 }
 
+type MQTTConfig 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"`
+	BatchSize int  `json:"batchSize"`
+}
+
+func (c *MQTTConfig) Configure() (kawa.Destination[types.Event], error) {
+	slog.Info("configuring mqtt")
+	return mqtt.New(
+		mqtt.WithBroker(c.Broker),
+		mqtt.WithClientID(c.ClientID),
+		mqtt.WithQOS(c.QOS),
+		mqtt.WithBatchSize(c.BatchSize),
+		mqtt.WithTopic(c.Topic),
+		mqtt.WithRetained(c.Retained),
+		mqtt.WithUserName(c.UserName),
+		mqtt.WithPassword(c.Password),
+	), 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"`
+	BatchSize int  `json:"batchSize"`
+}
+
+func (c *MQTTSrcConfig) Configure() (kawa.Source[types.Event], error) {
+	slog.Info("configuring mqtt")
+	return mqttsrc.New(
+		mqttsrc.WithBroker(c.Broker),
+		mqttsrc.WithClientID(c.ClientID),
+		mqttsrc.WithQOS(c.QOS),
+		mqttsrc.WithTopic(c.Topic),
+		mqttsrc.WithRetained(c.Retained),
+		mqttsrc.WithUserName(c.UserName),
+		mqttsrc.WithPassword(c.Password),
+	), nil
+}
+
 type S3Config struct {
 	BucketName   string `json:"bucketName"`
 	PathPrefix   string `json:"pathPrefix"`

examples/mqtt_config.json

@@ -0,0 +1,37 @@
+{
+    // 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.mqtt:1883",
+            "clientID": "kawa_src",
+            "userName": "",
+            "password": "",
+            "topic": "iot/#",
+
+            "qos": 1, // Optional defaults to 1 if not included
+            "retained": false, // Optional defaults to false if not included
+          },
+    ],
+    "destinations": [
+        {
+            "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
+            "batchSize": 100, // Optional defaults to 100 if not included
+          },
+        {
+            "type":"printer"
+        }
+    ],
+  }
+
+

go.mod

@@ -16,6 +16,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=

internal/destinations/mqtt/mqtt.go

@@ -0,0 +1,139 @@
+package mqtt
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"time"
+
+	MQTT "github.com/eclipse/paho.mqtt.golang"
+	"github.com/runreveal/kawa"
+	"github.com/runreveal/kawa/internal/types"
+	batch "github.com/runreveal/kawa/x/batcher"
+)
+
+type Option func(*mqtt)
+
+func WithBroker(broker string) Option {
+	return func(m *mqtt) {
+		m.broker = broker
+	}
+}
+
+func WithClientID(clientID string) Option {
+	return func(m *mqtt) {
+		m.clientID = clientID
+	}
+}
+
+func WithTopic(topic string) Option {
+	return func(m *mqtt) {
+		m.topic = topic
+	}
+}
+
+func WithQOS(qos byte) Option {
+	return func(m *mqtt) {
+		m.qos = qos
+	}
+}
+
+func WithRetained(retained bool) Option {
+	return func(m *mqtt) {
+		m.retained = retained
+	}
+}
+
+func WithBatchSize(batchSize int) Option {
+	return func(m *mqtt) {
+		m.batchSize = batchSize
+	}
+}
+
+func WithUserName(userName string) Option {
+	return func(m *mqtt) {
+		m.userName = userName
+	}
+}
+
+func WithPassword(password string) Option {
+	return func(m *mqtt) {
+		m.password = password
+	}
+}
+
+type mqtt struct {
+	batcher *batch.Destination[types.Event]
+
+	broker   string
+	clientID string
+	topic    string
+
+	userName string
+	password string
+
+	qos      byte
+	retained bool
+
+	batchSize int
+}
+
+func New(opts ...Option) *mqtt {
+	ret := &mqtt{
+		qos:       1,
+		retained:  false,
+		batchSize: 100,
+		topic:     "#",
+	}
+	for _, o := range opts {
+		o(ret)
+	}
+
+	ret.batcher = batch.NewDestination[types.Event](ret,
+		batch.FlushLength(ret.batchSize),
+		batch.FlushFrequency(5*time.Second),
+	)
+	return ret
+}
+
+func (m *mqtt) Run(ctx context.Context) error {
+	if m.broker == "" {
+		return errors.New("missing broker")
+	}
+	if m.clientID == "" {
+		return errors.New("missing clientID")
+	}
+
+	return m.batcher.Run(ctx)
+}
+
+func (m *mqtt) Send(ctx context.Context, ack func(), msgs ...kawa.Message[types.Event]) error {
+	return m.batcher.Send(ctx, ack, msgs...)
+}
+
+// Flush sends the given messages of type kawa.Message[type.Event] to an MQTT topic
+func (m *mqtt) Flush(ctx context.Context, msgs []kawa.Message[types.Event]) error {
+	opts := MQTT.NewClientOptions().AddBroker(m.broker).
+		SetClientID(m.clientID).SetUsername(m.userName).SetPassword(m.password)
+	client := MQTT.NewClient(opts)
+	// m.client = client
+
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		return token.Error()
+	}
+	defer client.Disconnect(250)
+
+	for _, msg := range msgs {
+		jsonData, err := json.Marshal(msg.Value)
+		if err != nil {
+			return err
+		}
+
+		token := client.Publish(m.topic, m.qos, m.retained, jsonData)
+		token.Wait()
+		if token.Error() != nil {
+			return token.Error()
+		}
+	}
+	return nil
+}