feat: add mqtt support on raccoon

Author: siddhanta-rath

config/load.go

@@ -29,4 +29,5 @@ func Load() {
 	metricStatsdConfigLoader()
 	eventDistributionConfigLoader()
 	eventConfigLoader()
+	serverMQTTConfigLoader()
 }

config/server.go

@@ -10,6 +10,7 @@ import (
 var Server server
 var ServerWs serverWs
 var ServerGRPC serverGRPC
+var ServerMQTT serverMQTT
 
 type server struct {
 	DedupEnabled bool
@@ -37,6 +38,33 @@ type serverGRPC struct {
 	TLSPublicKey string
 }
 
+type serverMQTT struct {
+	ConsulConfig   consul
+	AuthConfig     auth
+	ConsumerConfig consumer
+	ConnGroup      string
+}
+
+type auth struct {
+	Username string
+	Password string
+}
+
+type consul struct {
+	Address    string
+	HealthOnly bool
+	KVKey      string
+	WaitTime   time.Duration
+}
+
+type consumer struct {
+	RetryIntervalInSec time.Duration
+	LogLevel           string
+	WriteTimeoutInSec  time.Duration
+	PoolSize           int
+	TopicFormat        string
+}
+
 func serverConfigLoader() {
 	viper.SetDefault("SERVER_BATCH_DEDUP_IN_CONNECTION_ENABLED", "false")
 	Server = server{
@@ -85,3 +113,39 @@ func serverGRPCConfigLoader() {
 		TLSPublicKey: util.MustGetString("SERVER_GRPC_TLS_PUBLIC_KEY"),
 	}
 }
+
+func serverMQTTConfigLoader() {
+	viper.SetDefault("SERVER_MQTT_CONSUL_ADDRESS", "consul:8081")
+	viper.SetDefault("SERVER_MQTT_CONSUL_KV_KEY", "kv/path")
+	viper.SetDefault("SERVER_MQTT_CONSUL_HEALTH_ONLY", true)
+	viper.SetDefault("SERVER_MQTT_CONSUL_WAIT_TIME", 300)
+	viper.SetDefault("SERVER_MQTT_AUTH_USERNAME", "test")
+	viper.SetDefault("SERVER_MQTT_AUTH_PASSWORD", "pass")
+	viper.SetDefault("SERVER_MQTT_CONSUMER_RETRY_INTERVAL_IN_SEC", 1)
+	viper.SetDefault("SERVER_MQTT_CONSUMER_WRITE_TIMEOUT_IN_SEC", 1)
+	viper.SetDefault("SERVER_MQTT_CONSUMER_LOG_LEVEL", "warn")
+	viper.SetDefault("SERVER_MQTT_CONSUMER_POOL_SIZE", 1)
+	viper.SetDefault("SERVER_MQTT_CONSUMER_TOPIC_FORMAT", "$share/clickstream/v1/+/+")
+	viper.SetDefault("SERVER_MQTT_CONNECTION_GROUP", "consumer")
+
+	ServerMQTT = serverMQTT{
+		ConsulConfig: consul{
+			Address:    util.MustGetString("SERVER_MQTT_CONSUL_ADDRESS"),
+			HealthOnly: util.MustGetBool("SERVER_MQTT_CONSUL_HEALTH_ONLY"),
+			KVKey:      util.MustGetString("SERVER_MQTT_CONSUL_KV_KEY"),
+			WaitTime:   util.MustGetDuration("SERVER_MQTT_CONSUL_WAIT_TIME", time.Second),
+		},
+		AuthConfig: auth{
+			Username: util.MustGetString("SERVER_MQTT_AUTH_USERNAME"),
+			Password: util.MustGetString("SERVER_MQTT_AUTH_USERNAME"),
+		},
+		ConsumerConfig: consumer{
+			RetryIntervalInSec: util.MustGetDuration("SERVER_MQTT_CONSUMER_RETRY_INTERVAL_IN_SEC", time.Second),
+			LogLevel:           util.MustGetString("SERVER_MQTT_CONSUMER_LOG_LEVEL"),
+			WriteTimeoutInSec:  util.MustGetDuration("SERVER_MQTT_CONSUMER_POOL_SIZE", time.Second),
+			PoolSize:           util.MustGetInt("SERVER_MQTT_CONSUMER_POOL_SIZE"),
+			TopicFormat:        util.MustGetString("SERVER_MQTT_CONSUMER_TOPIC_FORMAT"),
+		},
+		ConnGroup: util.MustGetString("SERVER_MQTT_CONNECTION_GROUP"),
+	}
+}

go.mod

@@ -1,43 +1,65 @@
 module github.com/goto/raccoon
 
-go 1.17
+go 1.24
 
 require (
 	buf.build/gen/go/gotocompany/proton/grpc/go v1.3.0-20230313110213-9a3d240d5293.1
 	buf.build/gen/go/gotocompany/proton/protocolbuffers/go v1.29.0-20230313110213-9a3d240d5293.1
+	github.com/gojek/courier-go v0.7.8
+	github.com/gojek/courier-go/consul v0.7.3
 	github.com/gorilla/mux v1.7.4
-	github.com/gorilla/websocket v1.4.2
+	github.com/gorilla/websocket v1.5.3
+	github.com/goto/raccoon/clients/go v0.0.0-20250203072106-1dbea749aaf3
 	github.com/sirupsen/logrus v1.6.0
 	github.com/spf13/viper v1.7.0
-	github.com/stretchr/testify v1.8.1
+	github.com/stretchr/testify v1.9.0
 	google.golang.org/grpc v1.53.0
 	google.golang.org/protobuf v1.29.0
 	gopkg.in/alexcesaro/statsd.v2 v2.0.0
 	gopkg.in/confluentinc/confluent-kafka-go.v1 v1.4.2
 )
 
 require (
+	github.com/armon/go-metrics v0.4.1 // indirect
 	github.com/confluentinc/confluent-kafka-go v1.4.2 // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/fatih/color v1.16.0 // indirect
 	github.com/fsnotify/fsnotify v1.4.7 // indirect
+	github.com/gojek/paho.mqtt.golang v1.7.0 // indirect
+	github.com/gojekfarm/xtools/generic v0.7.0 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/hashicorp/consul/api v1.32.1 // indirect
+	github.com/hashicorp/errwrap v1.1.0 // indirect
+	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
+	github.com/hashicorp/go-hclog v1.5.0 // indirect
+	github.com/hashicorp/go-immutable-radix v1.3.1 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
+	github.com/hashicorp/go-rootcerts v1.0.2 // indirect
+	github.com/hashicorp/golang-lru v0.5.4 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
+	github.com/hashicorp/serf v0.10.1 // indirect
 	github.com/konsorten/go-windows-terminal-sequences v1.0.3 // indirect
+	github.com/kr/text v0.2.0 // indirect
 	github.com/magiconair/properties v1.8.1 // indirect
-	github.com/mitchellh/mapstructure v1.1.2 // indirect
+	github.com/mattn/go-colorable v0.1.13 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/mitchellh/go-homedir v1.1.0 // indirect
+	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/pelletier/go-toml v1.2.0 // indirect
-	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/spf13/afero v1.9.2 // indirect
 	github.com/spf13/cast v1.3.0 // indirect
 	github.com/spf13/jwalterweatherman v1.0.0 // indirect
 	github.com/spf13/pflag v1.0.3 // indirect
-	github.com/stretchr/objx v0.5.0 // indirect
+	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/subosito/gotenv v1.2.0 // indirect
-	golang.org/x/net v0.5.0 // indirect
-	golang.org/x/sys v0.4.0 // indirect
-	golang.org/x/text v0.6.0 // indirect
+	golang.org/x/exp v0.0.0-20250305212735-054e65f0b394 // indirect
+	golang.org/x/net v0.38.0 // indirect
+	golang.org/x/sync v0.12.0 // indirect
+	golang.org/x/sys v0.31.0 // indirect
+	golang.org/x/text v0.23.0 // indirect
 	google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f // indirect
 	gopkg.in/ini.v1 v1.51.0 // indirect
-	gopkg.in/yaml.v2 v2.2.4 // indirect
+	gopkg.in/yaml.v2 v2.2.5 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )