Add SQS protocol

protocol/sqs/v2/go.mod

@@ -0,0 +1,24 @@
+module github.com/cloudevents/sdk-go/protocol/sqs/v2
+
+go 1.23.0
+
+toolchain go1.23.8
+
+replace github.com/cloudevents/sdk-go/v2 => ../../../v2
+
+require (
+	github.com/aws/aws-sdk-go-v2 v1.38.0
+	github.com/aws/aws-sdk-go-v2/service/sqs v1.41.0
+	github.com/cloudevents/sdk-go/v2 v2.16.1
+)
+
+require (
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.3 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.3 // indirect
+	github.com/aws/smithy-go v1.22.5 // 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
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
+)

protocol/sqs/v2/go.sum

@@ -0,0 +1,37 @@
+github.com/aws/aws-sdk-go-v2 v1.38.0 h1:UCRQ5mlqcFk9HJDIqENSLR3wiG1VTWlyUfLDEvY7RxU=
+github.com/aws/aws-sdk-go-v2 v1.38.0/go.mod h1:9Q0OoGQoboYIAJyslFyF1f5K1Ryddop8gqMhWx/n4Wg=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.3 h1:o9RnO+YZ4X+kt5Z7Nvcishlz0nksIt2PIzDglLMP0vA=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.3/go.mod h1:+6aLJzOG1fvMOyzIySYjOFjcguGvVRL68R+uoRencN4=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.3 h1:joyyUFhiTQQmVK6ImzNU9TQSNRNeD9kOklqTzyk5v6s=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.3/go.mod h1:+vNIyZQP3b3B1tSLI0lxvrU9cfM7gpdRXMFfm67ZcPc=
+github.com/aws/aws-sdk-go-v2/service/sqs v1.41.0 h1:xobvQ4NxlXFUNgVwE6cnMI/ww7K7jtQMWKor2Gi61Xg=
+github.com/aws/aws-sdk-go-v2/service/sqs v1.41.0/go.mod h1:RExz4LhRKY5iogQ1dz7KVa3JyBY0PBotXovrDj850Sc=
+github.com/aws/smithy-go v1.22.5 h1:P9ATCXPMb2mPjYBgueqJNCA5S9UfktsW0tTxi+a7eqw=
+github.com/aws/smithy-go v1.22.5/go.mod h1:t1ufH5HMublsJYulve2RKmHDC15xu1f26kHCp/HgceI=
+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/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/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
+github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
+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/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.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+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=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

protocol/sqs/v2/message.go

@@ -0,0 +1,175 @@
+/*
+Copyright 2023 The CloudEvents Authors
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package sqs
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
+
+	"github.com/cloudevents/sdk-go/v2/binding"
+	"github.com/cloudevents/sdk-go/v2/binding/format"
+	"github.com/cloudevents/sdk-go/v2/binding/spec"
+	"github.com/cloudevents/sdk-go/v2/protocol"
+)
+
+const (
+	prefix      = "ce-"
+	contentType = "Content-Type"
+)
+
+var specs = spec.WithPrefix(prefix)
+
+type SQSDeleteMessageAPI interface {
+	DeleteMessage(ctx context.Context, params *sqs.DeleteMessageInput, optFns ...func(*sqs.Options)) (*sqs.DeleteMessageOutput, error)
+}
+
+// Message implements binding.Message by wrapping an SQS types.Message.
+// This message *can* be read several times safely
+type Message struct {
+	Msg      *types.Message
+	queueURL *string
+	client   SQSDeleteMessageAPI
+
+	version spec.Version
+	format  format.Format
+}
+
+var (
+	_ binding.Message               = (*Message)(nil)
+	_ binding.MessageMetadataReader = (*Message)(nil)
+)
+
+// NewMessage wraps an SQS types.Message in a binding.Message.
+// The returned message *can* be read several times safely
+// The default encoding returned is EncodingStructured unless the SQS message contains a specversion
+// header.
+func NewMessage(msg *types.Message, sqsClient SQSDeleteMessageAPI, queueURL *string) *Message {
+	v := getSpecVersion(msg)
+	fmt := getFormat(msg)
+	if v == nil && fmt == nil {
+		fmt = format.JSON
+	}
+	return &Message{Msg: msg, version: v, format: fmt, client: sqsClient, queueURL: queueURL}
+}
+
+func getSpecVersion(message *types.Message) spec.Version {
+	if sv, ok := message.MessageAttributes[specs.PrefixedSpecVersionName()]; ok {
+		if sv.StringValue != nil {
+			return specs.Version(aws.ToString(sv.StringValue))
+		}
+	}
+	return nil
+}
+
+func getFormat(message *types.Message) format.Format {
+	if sv, ok := message.MessageAttributes[contentType]; ok {
+		if sv.StringValue != nil && format.IsFormat(aws.ToString(sv.StringValue)) {
+			return format.Lookup(aws.ToString(sv.StringValue))
+		}
+	}
+	return nil
+}
+
+// ReadEncoding return the type of the message Encoding.
+func (m *Message) ReadEncoding() binding.Encoding {
+	if m.version != nil {
+		return binding.EncodingBinary
+	}
+	if m.format != nil {
+		return binding.EncodingStructured
+	}
+	return binding.EncodingUnknown
+}
+
+// ReadStructured transfers a structured-mode event to a StructuredWriter.
+func (m *Message) ReadStructured(ctx context.Context, encoder binding.StructuredWriter) error {
+	if m.version != nil {
+		return binding.ErrNotStructured
+	}
+	if m.format == nil {
+		return binding.ErrNotStructured
+	}
+	data := []byte(*m.Msg.Body)
+	return encoder.SetStructuredEvent(ctx, m.format, bytes.NewReader(data))
+}
+
+// ReadBinary transfers a binary-mode event to an BinaryWriter.
+func (m *Message) ReadBinary(ctx context.Context, encoder binding.BinaryWriter) error {
+	if m.format != nil {
+		return binding.ErrNotBinary
+	}
+	var err error
+	for k, attr := range m.Msg.MessageAttributes {
+		v := aws.ToString(attr.StringValue)
+		if strings.HasPrefix(k, prefix) {
+			attr := m.version.Attribute(k)
+			if attr != nil {
+				err = encoder.SetAttribute(attr, v)
+			} else {
+				err = encoder.SetExtension(strings.ToLower(strings.TrimPrefix(k, prefix)), v)
+			}
+		} else if k == contentType {
+			err = encoder.SetAttribute(m.version.AttributeFromKind(spec.DataContentType), v)
+		}
+		if err != nil {
+			return err
+		}
+	}
+
+	if m.Msg.Body != nil {
+		err = encoder.SetData(bytes.NewBuffer([]byte(*m.Msg.Body)))
+	}
+
+	return err
+}
+
+// GetAttribute implements binding.MessageMetadataReader
+func (m *Message) GetAttribute(k spec.Kind) (spec.Attribute, interface{}) {
+	attr := m.version.AttributeFromKind(k)
+	if attr == nil {
+		return nil, nil
+	}
+	key := withPrefix(attr.Name())
+	if msgAttr, ok := m.Msg.MessageAttributes[key]; ok && aws.ToString(msgAttr.StringValue) != "" {
+		return attr, aws.ToString(msgAttr.StringValue)
+	}
+	return nil, nil
+}
+
+// GetExtension implements binding.MessageMetadataReader
+func (m *Message) GetExtension(name string) interface{} {
+	key := withPrefix(name)
+	if attr, ok := m.Msg.MessageAttributes[key]; ok && attr.StringValue != nil {
+		return aws.ToString(attr.StringValue)
+	}
+	return nil
+}
+
+// Finish implements binding.Message
+// It deletes the message from SQS when the CloudEvent has been ACKed.
+func (m *Message) Finish(err error) error {
+	if protocol.IsACK(err) {
+		// If the error is an ACK, we delete the message from SQS.
+		ctx := context.Background()
+		_, err = m.client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
+			QueueUrl:      m.queueURL,
+			ReceiptHandle: m.Msg.ReceiptHandle,
+		})
+		return err
+	}
+	return nil
+}
+
+// withPrefix prepends the prefix to the attribute name
+func withPrefix(attributeName string) string {
+	return fmt.Sprintf("%s%s", prefix, attributeName)
+}

protocol/sqs/v2/message_test.go

@@ -0,0 +1,100 @@
+/*
+Copyright 2023 The CloudEvents Authors
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package sqs
+
+import (
+	"context"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
+	"github.com/cloudevents/sdk-go/v2/event"
+)
+
+type mockGetObjectAPI func(ctx context.Context, params *sqs.DeleteMessageInput, optFns ...func(*sqs.Options)) (*sqs.DeleteMessageOutput, error)
+
+func (m mockGetObjectAPI) DeleteMessage(ctx context.Context, params *sqs.DeleteMessageInput, optFns ...func(*sqs.Options)) (*sqs.DeleteMessageOutput, error) {
+	return m(ctx, params, optFns...)
+}
+
+func TestReadStructured(t *testing.T) {
+	tests := []struct {
+		name      string
+		client    func(t *testing.T) SQSDeleteMessageAPI
+		msg       *types.Message
+		queueName *string
+		wantErr   error
+	}{
+		{
+			name: "nil format",
+			msg: &types.Message{
+				Body: aws.String(""),
+			},
+			client: func(t *testing.T) SQSDeleteMessageAPI {
+				return mockGetObjectAPI(func(ctx context.Context, params *sqs.DeleteMessageInput, optFns ...func(*sqs.Options)) (*sqs.DeleteMessageOutput, error) {
+					return nil, nil
+				})
+			},
+		},
+		{
+			name: "json format",
+			msg: &types.Message{
+				Body: aws.String(""),
+				MessageAttributes: map[string]types.MessageAttributeValue{
+					contentType: stringMessageAttribute(event.ApplicationCloudEventsJSON),
+				},
+			},
+			client: func(t *testing.T) SQSDeleteMessageAPI {
+				return mockGetObjectAPI(func(ctx context.Context, params *sqs.DeleteMessageInput, optFns ...func(*sqs.Options)) (*sqs.DeleteMessageOutput, error) {
+					t.Helper()
+					return nil, nil
+				})
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			msg := NewMessage(tc.msg, tc.client(t), tc.queueName)
+			err := msg.ReadStructured(context.Background(), msgToWriter(tc.msg))
+			if err != tc.wantErr {
+				t.Errorf("Error unexpected. got: %v, want: %v", err, tc.wantErr)
+			}
+		})
+	}
+}
+
+func TestReadBinary(t *testing.T) {
+	msg := &types.Message{
+		Body: aws.String("{hello:world}"),
+		MessageAttributes: map[string]types.MessageAttributeValue{
+			"ce-specversion": stringMessageAttribute("1.0"),
+			"ce-type":        stringMessageAttribute("binary.test"),
+			"ce-source":      stringMessageAttribute("test-source"),
+			"ce-id":          stringMessageAttribute("ABC-123"),
+		},
+		ReceiptHandle: aws.String("test-receipt-handle"),
+	}
+
+	client := func(t *testing.T) SQSDeleteMessageAPI {
+		return mockGetObjectAPI(func(ctx context.Context, params *sqs.DeleteMessageInput, optFns ...func(*sqs.Options)) (*sqs.DeleteMessageOutput, error) {
+			t.Helper()
+			return nil, nil
+		})
+	}
+	message := NewMessage(msg, client(t), aws.String("test-queue"))
+	err := message.ReadBinary(context.Background(), msgToWriter(msg))
+	if err != nil {
+		t.Errorf("Error unexpected. got: %v", err)
+	}
+}
+
+func msgToWriter(msg *types.Message) *sqsMessageWriter {
+	return &sqsMessageWriter{
+		MessageBody:       msg.Body,
+		MessageAttributes: msg.MessageAttributes,
+	}
+}