Implement missing MQTTv3 protocol support

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
+)

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=

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
+}

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
+}

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())
+	})
+}