feat: add proto decoder

Author: siddhanta-rath

go.mod

@@ -7,14 +7,15 @@ require (
 	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/gojekfarm/xtools/xproto v0.10.0
 	github.com/gorilla/mux v1.7.4
 	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.9.0
 	google.golang.org/grpc v1.53.0
-	google.golang.org/protobuf v1.29.0
+	google.golang.org/protobuf v1.33.0
 	gopkg.in/alexcesaro/statsd.v2 v2.0.0
 	gopkg.in/confluentinc/confluent-kafka-go.v1 v1.4.2
 )

go.sum

@@ -501,6 +501,8 @@ github.com/gojek/paho.mqtt.golang v1.7.0/go.mod h1:dsB7ojVcuI8Nw1t35g72omp1Swa6/
 github.com/gojek/valkyrie v0.0.0-20180215180059-6aee720afcdf/go.mod h1:QzhUKaYKJmcbTnCYCAVQrroCOY7vOOI8cSQ4NbuhYf0=
 github.com/gojekfarm/xtools/generic v0.7.0 h1:iBLSYE+eekH8pUvsf0qHLyBS77mMPWlKK7xUE1kZFfU=
 github.com/gojekfarm/xtools/generic v0.7.0/go.mod h1:T0yk+yt4s1jwr9Gbtyb/BADXGKRfCYIBz9pN1GnmkQA=
+github.com/gojekfarm/xtools/xproto v0.10.0 h1:cxRe/8OJKkAb5c52NOfnaXvtCmlQkrIwslzHCntQY60=
+github.com/gojekfarm/xtools/xproto v0.10.0/go.mod h1:qbna3ipjT8aEFXuOBi91fFlejLfi8f0YohjXheBRyTU=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
 github.com/golang/glog v1.0.0/go.mod h1:EWib/APOK0SL3dFbYqvxE3UYd8E6s1ouQ7iEp/0LWV4=
 github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
@@ -1402,8 +1404,9 @@ google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQ
 google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
-google.golang.org/protobuf v1.29.0 h1:44S3JjaKmLEE4YIkjzexaP+NzZsudE3Zin5Njn/pYX0=
 google.golang.org/protobuf v1.29.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
+google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
+google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/alexcesaro/statsd.v2 v2.0.0 h1:FXkZSCZIH17vLCO5sO2UucTHsH9pc+17F6pl3JVCwMc=
 gopkg.in/alexcesaro/statsd.v2 v2.0.0/go.mod h1:i0ubccKGzBVNBpdGV5MocxyA/XlLUJzA7SLonnE4drU=

services/mqtt/client/pubsub.go

@@ -5,8 +5,10 @@ import (
 	"fmt"
 	"github.com/gojek/courier-go"
 	"github.com/gojek/courier-go/consul"
+	"github.com/gojekfarm/xtools/xproto"
 	"github.com/goto/raccoon/config"
 	"github.com/goto/raccoon/logger"
+	"io"
 )
 
 // MqttPubSubClient wraps a courier MQTT client with start/stop lifecycle management.
@@ -46,6 +48,7 @@ func NewMqttPubSubClient(ctx context.Context, handler courier.MessageHandler, cl
 		courier.WithWriteTimeout(config.ServerMQTT.ConsumerConfig.WriteTimeoutInSec),
 		courier.WithOnConnect(registerHandler(ctx, handler)),
 		courier.WithLogger(logger.GetLogger()),
+		courier.WithCustomDecoder(protoDecoder),
 	}
 
 	client, err := courier.NewClient(clientOpts...)
@@ -91,3 +94,7 @@ func (m *MqttPubSubClient) Stop() error {
 func (m *MqttPubSubClient) IsConnected() bool {
 	return m.client.IsConnected()
 }
+
+func protoDecoder(ctx context.Context, r io.Reader) courier.Decoder {
+	return xproto.NewDecoder(r)
+}