feat(watermill): add sonic marshaler

adapter/watermill/marshaler.go

@@ -0,0 +1,40 @@
+package watermill
+
+import (
+	"github.com/ThreeDotsLabs/watermill/components/cqrs"
+	"github.com/ThreeDotsLabs/watermill/message"
+	"github.com/bytedance/sonic"
+	"github.com/google/uuid"
+)
+
+var _ cqrs.CommandEventMarshaler = SonicMarshaler{}
+
+type SonicMarshaler struct{}
+
+func NewSonicMarshaler() SonicMarshaler {
+	return SonicMarshaler{}
+}
+
+func (m SonicMarshaler) Marshal(v interface{}) (*message.Message, error) {
+	b, err := sonic.Marshal(v)
+	if err != nil {
+		return nil, err
+	}
+
+	msg := message.NewMessage(uuid.NewString(), b)
+	msg.Metadata.Set("name", m.Name(v))
+
+	return msg, nil
+}
+
+func (SonicMarshaler) Unmarshal(msg *message.Message, v interface{}) (err error) {
+	return sonic.Unmarshal(msg.Payload, v)
+}
+
+func (m SonicMarshaler) Name(cmdOrEvent interface{}) string {
+	return cqrs.FullyQualifiedStructName(cmdOrEvent)
+}
+
+func (m SonicMarshaler) NameFromMessage(msg *message.Message) string {
+	return msg.Metadata.Get("name")
+}

adapter/watermill/module.go

@@ -20,9 +20,10 @@ var (
 
 	Providers = fx.Options(
 		fx.Provide(newLogger, fx.Private),
-		fx.Provide(newMarshaler, fx.Private),
+		fx.Provide(NewSonicMarshaler, fx.Private),
 		fx.Provide(newRouter, fx.Private),
 		fx.Provide(newGoChannel, fx.Private),
+		fx.Provide(func(m SonicMarshaler) cqrs.CommandEventMarshaler { return m }, fx.Private),
 		fx.Provide(func(c *gochannel.GoChannel) message.Publisher { return c }, fx.Private),
 		fx.Provide(func(c *gochannel.GoChannel) message.Subscriber { return c }, fx.Private),
 
@@ -36,10 +37,6 @@ var (
 	)
 )
 
-func newMarshaler() cqrs.CommandEventMarshaler {
-	return cqrs.JSONMarshaler{}
-}
-
 func newGoChannel(logger watermill.LoggerAdapter) *gochannel.GoChannel {
 	return gochannel.NewGoChannel(gochannel.Config{}, logger)
 }