diff --git a/protocol/mqttv3_paho/go.mod b/protocol/mqttv3_paho/go.mod new file mode 100644 index 000000000..9c46d5580 --- /dev/null +++ b/protocol/mqttv3_paho/go.mod @@ -0,0 +1,26 @@ +module github.com/cloudevents/sdk-go/protocol/mqttv3_paho + +go 1.25.1 + +replace github.com/cloudevents/sdk-go/v2 => ../../v2 + +require ( + github.com/cloudevents/sdk-go/v2 v2.16.1 + github.com/eclipse/paho.mqtt.golang v1.5.1 + github.com/stretchr/testify v1.11.1 +) + +require ( + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/gorilla/websocket v1.5.3 // indirect + github.com/json-iterator/go v1.1.12 // indirect + github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect + github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + go.uber.org/multierr v1.11.0 // indirect + go.uber.org/zap v1.27.0 // indirect + golang.org/x/net v0.44.0 // indirect + golang.org/x/sync v0.17.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/protocol/mqttv3_paho/go.sum b/protocol/mqttv3_paho/go.sum new file mode 100644 index 000000000..22bcf72cc --- /dev/null +++ b/protocol/mqttv3_paho/go.sum @@ -0,0 +1,48 @@ +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.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE= +github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs= +github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw= +github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= +go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= +go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8= +go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= +golang.org/x/net v0.44.0 h1:evd8IRDyfNBMBTTY5XRF1vaZlD+EmWx6x8PkhR04H/I= +golang.org/x/net v0.44.0/go.mod h1:ECOoLqd5U3Lhyeyo/QDCEVQ4sNgYsqvCZ722XogGieY= +golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= +golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE= +golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f h1:BLraFXnmrev5lT+xlilqcH8XK9/i0At2xKjWk4p6zsU= +gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/protocol/mqttv3_paho/mqttv3.go b/protocol/mqttv3_paho/mqttv3.go new file mode 100644 index 000000000..5f4b75585 --- /dev/null +++ b/protocol/mqttv3_paho/mqttv3.go @@ -0,0 +1,155 @@ +package mqttv3_paho + +import ( + "context" + "errors" + "fmt" + "io" + "sync" + + mqtt "github.com/eclipse/paho.mqtt.golang" + + "github.com/cloudevents/sdk-go/v2/binding" + cecontext "github.com/cloudevents/sdk-go/v2/context" + "github.com/cloudevents/sdk-go/v2/protocol" +) + +const defaultQuiesce = 250 // milliseconds + +type Protocol struct { + client mqtt.Client + quiesce uint + + incoming chan mqtt.Message + openerMutex sync.Mutex + closeChan chan struct{} + + subscriptions map[string]byte + + topic string + qos byte + retained bool +} + +var ( + _ protocol.Opener = (*Protocol)(nil) + _ protocol.Sender = (*Protocol)(nil) + _ protocol.Receiver = (*Protocol)(nil) + _ protocol.Closer = (*Protocol)(nil) +) + +func New(ctx context.Context, clientOptions *mqtt.ClientOptions, opts ...Option) (*Protocol, error) { + p := &Protocol{ + client: mqtt.NewClient(clientOptions), + quiesce: defaultQuiesce, + incoming: make(chan mqtt.Message), + closeChan: make(chan struct{}), + } + + for _, opt := range opts { + if err := opt(p); err != nil { + return nil, err + } + } + + token := p.client.Connect() + + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-token.Done(): + if token.Error() != nil { + return nil, token.Error() + } + + return p, nil + case <-p.closeChan: + return nil, errors.New("client closed") + } +} + +func (p *Protocol) Receive(ctx context.Context) (binding.Message, error) { + select { + case m, ok := <-p.incoming: + if !ok { + return nil, io.EOF + } + msg := NewMessage(m.Payload()) + return msg, nil + case <-ctx.Done(): + return nil, io.EOF + } +} + +func (p *Protocol) Send(ctx context.Context, m binding.Message, transformers ...binding.Transformer) error { + logger := cecontext.LoggerFrom(ctx) + + var err error + defer func() { + if fErr := m.Finish(err); fErr != nil { + logger.Warnf("failed to finish message: %v", fErr) + } + }() + + topic := p.topic + if cecontext.TopicFrom(ctx) != "" { + topic = cecontext.TopicFrom(ctx) + cecontext.WithTopic(ctx, "") + } + + payload, err := WritePubMessage(ctx, m, transformers...) + if err != nil { + return err + } + + token := p.client.Publish(topic, p.qos, p.retained, payload) + + <-token.Done() + if token.Error() != nil { + err = fmt.Errorf("publish to %q: %w", topic, token.Error()) + logger.Error(err) + return err + } + + return nil +} + +func (p *Protocol) OpenInbound(ctx context.Context) error { + if len(p.subscriptions) == 0 { + return errors.New("no subscriptions available") + } + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + p.openerMutex.Lock() + defer p.openerMutex.Unlock() + + logger := cecontext.LoggerFrom(ctx) + logger.Infof("subscribing to topics: %v", p.subscriptions) + + token := p.client.SubscribeMultiple(p.subscriptions, func(_ mqtt.Client, message mqtt.Message) { + p.incoming <- message + }) + <-token.Done() + + if err := token.Error(); err != nil { + err = fmt.Errorf("subscribe to %v failed: %w", p.subscriptions, err) + logger.Error(err) + return err + } + + select { + case <-ctx.Done(): + case <-p.closeChan: + cancel() + } + + p.client.Disconnect(p.quiesce) + return nil +} + +func (p *Protocol) Close(context.Context) error { + close(p.closeChan) + return nil +} diff --git a/protocol/mqttv3_paho/msgs.go b/protocol/mqttv3_paho/msgs.go new file mode 100644 index 000000000..aef5f0de9 --- /dev/null +++ b/protocol/mqttv3_paho/msgs.go @@ -0,0 +1,77 @@ +package mqttv3_paho + +import ( + "bytes" + "context" + "io" + + ce "github.com/cloudevents/sdk-go/v2" + "github.com/cloudevents/sdk-go/v2/binding" + "github.com/cloudevents/sdk-go/v2/binding/format" +) + +type message []byte + +var ( + _ binding.StructuredWriter = (*message)(nil) + _ binding.Message = (*message)(nil) +) + +func NewMessage(b []byte) binding.Message { + m := message(b) + return &m +} + +func WritePubMessage(ctx context.Context, m binding.Message, transformers ...binding.Transformer) ([]byte, error) { + ctx = binding.WithForceStructured(ctx) + + var msg message + + _, err := binding.Write( + ctx, + m, + &msg, + nil, + transformers..., + ) + if err != nil { + return nil, err + } + + return msg, nil +} + +func (m *message) SetStructuredEvent(_ context.Context, _ format.Format, event io.Reader) error { + res, err := io.ReadAll(event) + if err != nil { + return err + } + + *m = res + return nil +} + +func (m *message) ReadEncoding() binding.Encoding { + var ev ce.Event + if err := ev.UnmarshalJSON(*m); err != nil { + return binding.EncodingUnknown + } + + if err := ev.Validate(); err != nil { + return binding.EncodingUnknown + } + + return binding.EncodingStructured +} + +func (m *message) ReadStructured(ctx context.Context, encoder binding.StructuredWriter) error { + return encoder.SetStructuredEvent(ctx, format.JSON, bytes.NewReader(*m)) +} + +func (m *message) ReadBinary(context.Context, binding.BinaryWriter) error { + return binding.ErrNotBinary +} + +func (m *message) Finish(error) error { + return nil +} diff --git a/protocol/mqttv3_paho/msgs_test.go b/protocol/mqttv3_paho/msgs_test.go new file mode 100644 index 000000000..d197e25fa --- /dev/null +++ b/protocol/mqttv3_paho/msgs_test.go @@ -0,0 +1,67 @@ +package mqttv3_paho + +import ( + "context" + "testing" + + ce "github.com/cloudevents/sdk-go/v2" + "github.com/cloudevents/sdk-go/v2/binding" + "github.com/stretchr/testify/require" +) + +func TestProtocolMessage(t *testing.T) { + ev := ce.NewEvent() + ev.SetID("123") + ev.SetSource("unit-test") + ev.SetType("unit-test-type") + ev.SetExtension("key", "value") + + err := ev.SetData(ce.ApplicationJSON, map[string]string{ + "hello": "world", + }) + require.NoError(t, err) + + t.Run("marshal event", func(t *testing.T) { + msg := (binding.EventMessage)(ev) + + b, err := WritePubMessage(context.Background(), &msg) + require.NoError(t, err) + require.NotEmpty(t, b) + + var result ce.Event + err = result.UnmarshalJSON(b) + require.NoError(t, err) + + require.Equal(t, ev, result) + }) + + t.Run("unmarshal event", func(t *testing.T) { + payload, err := ev.MarshalJSON() + require.NoError(t, err) + + msg := NewMessage(payload) + + result, err := binding.ToEvent(context.Background(), msg) + require.NoError(t, err) + + require.Equal(t, &ev, result) + }) + + t.Run("unmarshal event gives valid encoding", func(t *testing.T) { + payload, err := ev.MarshalJSON() + require.NoError(t, err) + + msg := NewMessage(payload) + require.Equal(t, binding.EncodingStructured, msg.ReadEncoding()) + }) + + t.Run("unmarshal event gives unknown encoding for non ce JSON", func(t *testing.T) { + msg := NewMessage([]byte(`{"hello": "world"}`)) + require.Equal(t, binding.EncodingUnknown, msg.ReadEncoding()) + }) + + t.Run("unmarshal event gives unknown encoding for malformed payloads", func(t *testing.T) { + msg := NewMessage([]byte(`{"`)) + require.Equal(t, binding.EncodingUnknown, msg.ReadEncoding()) + }) +} diff --git a/protocol/mqttv3_paho/options.go b/protocol/mqttv3_paho/options.go new file mode 100644 index 000000000..7201dddfb --- /dev/null +++ b/protocol/mqttv3_paho/options.go @@ -0,0 +1,61 @@ +package mqttv3_paho + +type Option func(*Protocol) error + +func WithSubscribeMap(subscribeMap map[string]byte) Option { + return func(p *Protocol) error { + if subscribeMap == nil || p == nil { + return nil + } + + if p.subscriptions == nil { + p.subscriptions = make(map[string]byte) + } + + for topic, qos := range subscribeMap { + p.subscriptions[topic] = qos + } + return nil + } +} + +func WithSubscribeTopic(topic string, qos byte) Option { + return func(p *Protocol) error { + if topic == "" || p == nil { + return nil + } + + if p.subscriptions == nil { + p.subscriptions = make(map[string]byte) + } + + p.subscriptions[topic] = qos + return nil + } +} + +func WithPublishTopic(topic string, qos byte, retained bool) Option { + return func(p *Protocol) error { + if topic == "" || p == nil { + return nil + } + + p.topic = topic + p.qos = qos + p.retained = retained + + return nil + } +} + +func WithDisconnectQuiesce(quiesce uint) Option { + return func(p *Protocol) error { + if quiesce == 0 || p == nil { + return nil + } + + p.quiesce = quiesce + + return nil + } +} diff --git a/protocol/mqttv3_paho/options_test.go b/protocol/mqttv3_paho/options_test.go new file mode 100644 index 000000000..3910da05c --- /dev/null +++ b/protocol/mqttv3_paho/options_test.go @@ -0,0 +1,107 @@ +package mqttv3_paho + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestWithSubscribeMap(t *testing.T) { + t.Run("should do nothing for nil map", func(t *testing.T) { + p := &Protocol{subscriptions: map[string]byte{"existing/topic": 1}} + err := WithSubscribeMap(nil)(p) + + require.NoError(t, err) + require.Equal(t, map[string]byte{"existing/topic": 1}, p.subscriptions) + }) + + t.Run("should do nothing for nil protocol", func(t *testing.T) { + err := WithSubscribeMap(map[string]byte{"test/topic": 1})(nil) + + require.NoError(t, err) + }) + + t.Run("should add subscriptions", func(t *testing.T) { + p := &Protocol{} + err := WithSubscribeMap(map[string]byte{"test/topic": 1, "another/topic": 0})(p) + + require.NoError(t, err) + require.Equal(t, map[string]byte{"test/topic": 1, "another/topic": 0}, p.subscriptions) + }) +} + +func TestWithSubscribeTopic(t *testing.T) { + t.Run("should do nothing for empty topic", func(t *testing.T) { + p := &Protocol{subscriptions: map[string]byte{"existing/topic": 1}} + err := WithSubscribeTopic("", 1)(p) + + require.NoError(t, err) + require.Equal(t, map[string]byte{"existing/topic": 1}, p.subscriptions) + }) + + t.Run("should do nothing for nil protocol", func(t *testing.T) { + err := WithSubscribeTopic("test/topic", 1)(nil) + + require.NoError(t, err) + }) + + t.Run("should add subscription", func(t *testing.T) { + p := &Protocol{} + err := WithSubscribeTopic("test/topic", 1)(p) + + require.NoError(t, err) + require.Equal(t, map[string]byte{"test/topic": 1}, p.subscriptions) + }) +} + +func TestWithDisconnectQuiesce(t *testing.T) { + t.Run("should do nothing for zero quiesce", func(t *testing.T) { + p := &Protocol{quiesce: 5} + err := WithDisconnectQuiesce(0)(p) + + require.NoError(t, err) + require.Equal(t, uint(5), p.quiesce) + }) + + t.Run("should do nothing for nil protocol", func(t *testing.T) { + err := WithDisconnectQuiesce(10)(nil) + + require.NoError(t, err) + }) + + t.Run("should set quiesce", func(t *testing.T) { + p := &Protocol{} + err := WithDisconnectQuiesce(10)(p) + + require.NoError(t, err) + require.Equal(t, uint(10), p.quiesce) + }) +} + +func TestWithPublishTopic(t *testing.T) { + t.Run("should not set empty topic", func(t *testing.T) { + p := &Protocol{} + err := WithPublishTopic("", 1, true)(p) + + require.NoError(t, err) + require.Equal(t, "", p.topic) + require.Equal(t, byte(0), p.qos) + require.Equal(t, false, p.retained) + }) + + t.Run("should do nothing for nil protocol", func(t *testing.T) { + err := WithPublishTopic("test/topic", 1, true)(nil) + + require.NoError(t, err) + }) + + t.Run("should set topic", func(t *testing.T) { + p := &Protocol{} + err := WithPublishTopic("test/topic", 1, true)(p) + + require.NoError(t, err) + require.Equal(t, "test/topic", p.topic) + require.Equal(t, byte(1), p.qos) + require.Equal(t, true, p.retained) + }) +}