[azservicebus] Adding in raw AMQP support for sending and receiving messages. (Azure#18413)

For sending AMQP messages you can use either Sender.SendAMQPMessages() or schedule them using Sender.ScheduleAMQPMessages().

For receiving, each ReceivedMessage now has a RawAMQPMessage property.

I've also added in a way to export the TLS pre-master key so you can easily wireshark and trace the actual AMQP interaction, which can be helpful in validating that things are encoding properly.

Fixes Azure#15260

Author: richardpark-msft

sdk/messaging/azservicebus/CHANGELOG.md

@@ -1,9 +1,13 @@
 # Release History
 
-## 1.0.2 (Unreleased)
+## 1.0.2-beta.0 (Unreleased)
 
 ### Features Added
 
+- Full access to send and receive all AMQP message properties. (#18413) 
+  - Send AMQP messages using the new `AMQPMessage` type and `Sender.SendAMQPMessage()` (AMQP messages can be added to MessageBatch's as well using MessageBatch.AddAMQPMessage()). 
+  - Access the full set of AMQP message properties when receiving using the `ReceivedMessage.RawAMQPMessage` property. 
+
 ### Breaking Changes
 
 ### Bugs Fixed

sdk/messaging/azservicebus/amqp_message.go

@@ -0,0 +1,279 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package azservicebus
+
+import (
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/go-amqp"
+)
+
+// AMQPMessage represents the AMQP message, as received from Service Bus.
+// For details about these properties, refer to the AMQP specification:
+//   https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#section-message-format
+//
+// Some fields in this struct are typed 'any', which means they will accept AMQP primitives, or in some
+// cases slices and maps.
+//
+// AMQP simple types include:
+// - int (any size), uint (any size)
+// - float (any size)
+// - string
+// - bool
+// - time.Time
+type AMQPMessage struct {
+	// ApplicationProperties corresponds to the "application-properties" section of an AMQP message.
+	//
+	// The values of the map are restricted to AMQP simple types, as listed in the comment for AMQPMessage.
+	ApplicationProperties map[string]any
+
+	// Body represents the body of an AMQP message.
+	Body AMQPMessageBody
+
+	// DeliveryAnnotations corresponds to the "delivery-annotations" section in an AMQP message.
+	//
+	// The values of the map are restricted to AMQP simple types, as listed in the comment for AMQPMessage.
+	DeliveryAnnotations map[any]any
+
+	// DeliveryTag corresponds to the delivery-tag property of the TRANSFER frame
+	// for this message.
+	DeliveryTag []byte
+
+	// Footer is the transport footers for this AMQP message.
+	//
+	// The values of the map are restricted to AMQP simple types, as listed in the comment for AMQPMessage.
+	Footer map[any]any
+
+	// Header is the transport headers for this AMQP message.
+	Header *AMQPMessageHeader
+
+	// MessageAnnotations corresponds to the message-annotations section of an AMQP message.
+	//
+	// The values of the map are restricted to AMQP simple types, as listed in the comment for AMQPMessage.
+	MessageAnnotations map[any]any
+
+	// Properties corresponds to the properties section of an AMQP message.
+	Properties *AMQPMessageProperties
+
+	linkName string
+
+	// inner is the AMQP message we originally received, which contains some hidden
+	// data that's needed to settle with go-amqp. We strip out most of the underlying
+	// data so it's fairly minimal.
+	inner *amqp.Message
+}
+
+// AMQPMessageProperties represents the properties of an AMQP message.
+// See here for more details:
+// http://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-properties
+type AMQPMessageProperties struct {
+	// AbsoluteExpiryTime corresponds to the 'absolute-expiry-time' property.
+	AbsoluteExpiryTime *time.Time
+
+	// ContentEncoding corresponds to the 'content-encoding' property.
+	ContentEncoding *string
+
+	// ContentType corresponds to the 'content-type' property
+	ContentType *string
+
+	// CorrelationID corresponds to the 'correlation-id' property.
+	// The type of CorrelationID can be a uint64, UUID, []byte, or a string
+	CorrelationID any
+
+	// CreationTime corresponds to the 'creation-time' property.
+	CreationTime *time.Time
+
+	// GroupID corresponds to the 'group-id' property.
+	GroupID *string
+
+	// GroupSequence corresponds to the 'group-sequence' property.
+	GroupSequence *uint32
+
+	// MessageID corresponds to the 'message-id' property.
+	// The type of MessageID can be a uint64, UUID, []byte, or string
+	MessageID any
+
+	// ReplyTo corresponds to the 'reply-to' property.
+	ReplyTo *string
+
+	// ReplyToGroupID corresponds to the 'reply-to-group-id' property.
+	ReplyToGroupID *string
+
+	// Subject corresponds to the 'subject' property.
+	Subject *string
+
+	// To corresponds to the 'to' property.
+	To *string
+
+	// UserID corresponds to the 'user-id' property.
+	UserID []byte
+}
+
+// AMQPMessageBody represents the body of an AMQP message.
+// Only one of these fields can be used a a time. They are mutually exclusive.
+type AMQPMessageBody struct {
+	// Data is encoded/decoded as multiple data sections in the body.
+	Data [][]byte
+
+	// Sequence is encoded/decoded as one or more amqp-sequence sections in the body.
+	//
+	// The values of the slices are are restricted to AMQP simple types, as listed in the comment for AMQPMessage.
+	Sequence [][]any
+
+	// Value is encoded/decoded as the amqp-value section in the body.
+	//
+	// The type of Value can be any of the AMQP simple types, as listed in the comment for AMQPMessage,
+	// as well as slices or maps of AMQP simple types.
+	Value any
+}
+
+// AMQPMessageHeader carries standard delivery details about the transfer
+// of a message.
+// See https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-header
+// for more details.
+type AMQPMessageHeader struct {
+	// DeliveryCount is the number of unsuccessful previous attempts to deliver this message.
+	// It corresponds to the 'delivery-count' property.
+	DeliveryCount uint32
+
+	// Durable corresponds to the 'durable' property.
+	Durable bool
+
+	// FirstAcquirer corresponds to the 'first-acquirer' property.
+	FirstAcquirer bool
+
+	// Priority corresponds to the 'priority' property.
+	Priority uint8
+
+	// TTL corresponds to the 'ttl' property.
+	TTL time.Duration
+}
+
+// toAMQPMessage converts between our (azservicebus) AMQP message
+// to the underlying message used by go-amqp.
+func (am *AMQPMessage) toAMQPMessage() *amqp.Message {
+	var header *amqp.MessageHeader
+
+	if am.Header != nil {
+		header = &amqp.MessageHeader{
+			DeliveryCount: am.Header.DeliveryCount,
+			Durable:       am.Header.Durable,
+			FirstAcquirer: am.Header.FirstAcquirer,
+			Priority:      am.Header.Priority,
+			TTL:           am.Header.TTL,
+		}
+	}
+
+	var properties *amqp.MessageProperties
+
+	if am.Properties != nil {
+		properties = &amqp.MessageProperties{
+			AbsoluteExpiryTime: am.Properties.AbsoluteExpiryTime,
+			ContentEncoding:    am.Properties.ContentEncoding,
+			ContentType:        am.Properties.ContentType,
+			CorrelationID:      am.Properties.CorrelationID,
+			CreationTime:       am.Properties.CreationTime,
+			GroupID:            am.Properties.GroupID,
+			GroupSequence:      am.Properties.GroupSequence,
+			MessageID:          am.Properties.MessageID,
+			ReplyTo:            am.Properties.ReplyTo,
+			ReplyToGroupID:     am.Properties.ReplyToGroupID,
+			Subject:            am.Properties.Subject,
+			To:                 am.Properties.To,
+			UserID:             am.Properties.UserID,
+		}
+	} else {
+		properties = &amqp.MessageProperties{}
+	}
+
+	var footer amqp.Annotations
+
+	if am.Footer != nil {
+		footer = (amqp.Annotations)(am.Footer)
+	}
+
+	return &amqp.Message{
+		Annotations:           copyAnnotations(am.MessageAnnotations),
+		ApplicationProperties: am.ApplicationProperties,
+		Data:                  am.Body.Data,
+		DeliveryAnnotations:   amqp.Annotations(am.DeliveryAnnotations),
+		DeliveryTag:           am.DeliveryTag,
+		Footer:                footer,
+		Header:                header,
+		Properties:            properties,
+		Sequence:              am.Body.Sequence,
+		Value:                 am.Body.Value,
+	}
+}
+
+func copyAnnotations(src map[any]any) amqp.Annotations {
+	if src == nil {
+		return amqp.Annotations{}
+	}
+
+	dest := amqp.Annotations{}
+
+	for k, v := range src {
+		dest[k] = v
+	}
+
+	return dest
+}
+
+func newAMQPMessage(goAMQPMessage *amqp.Message) *AMQPMessage {
+	var header *AMQPMessageHeader
+
+	if goAMQPMessage.Header != nil {
+		header = &AMQPMessageHeader{
+			DeliveryCount: goAMQPMessage.Header.DeliveryCount,
+			Durable:       goAMQPMessage.Header.Durable,
+			FirstAcquirer: goAMQPMessage.Header.FirstAcquirer,
+			Priority:      goAMQPMessage.Header.Priority,
+			TTL:           goAMQPMessage.Header.TTL,
+		}
+	}
+
+	var properties *AMQPMessageProperties
+
+	if goAMQPMessage.Properties != nil {
+		properties = &AMQPMessageProperties{
+			AbsoluteExpiryTime: goAMQPMessage.Properties.AbsoluteExpiryTime,
+			ContentEncoding:    goAMQPMessage.Properties.ContentEncoding,
+			ContentType:        goAMQPMessage.Properties.ContentType,
+			CorrelationID:      goAMQPMessage.Properties.CorrelationID,
+			CreationTime:       goAMQPMessage.Properties.CreationTime,
+			GroupID:            goAMQPMessage.Properties.GroupID,
+			GroupSequence:      goAMQPMessage.Properties.GroupSequence,
+			MessageID:          goAMQPMessage.Properties.MessageID,
+			ReplyTo:            goAMQPMessage.Properties.ReplyTo,
+			ReplyToGroupID:     goAMQPMessage.Properties.ReplyToGroupID,
+			Subject:            goAMQPMessage.Properties.Subject,
+			To:                 goAMQPMessage.Properties.To,
+			UserID:             goAMQPMessage.Properties.UserID,
+		}
+	}
+
+	var footer map[any]any
+
+	if goAMQPMessage.Footer != nil {
+		footer = (map[any]any)(goAMQPMessage.Footer)
+	}
+
+	return &AMQPMessage{
+		MessageAnnotations:    map[any]any(goAMQPMessage.Annotations),
+		ApplicationProperties: goAMQPMessage.ApplicationProperties,
+		Body: AMQPMessageBody{
+			Data:     goAMQPMessage.Data,
+			Sequence: goAMQPMessage.Sequence,
+			Value:    goAMQPMessage.Value,
+		},
+		DeliveryAnnotations: map[any]any(goAMQPMessage.DeliveryAnnotations),
+		DeliveryTag:         goAMQPMessage.DeliveryTag,
+		Footer:              footer,
+		Header:              header,
+		linkName:            goAMQPMessage.LinkName(),
+		Properties:          properties,
+		inner:               goAMQPMessage,
+	}
+}

sdk/messaging/azservicebus/amqp_message_test.go

@@ -0,0 +1,22 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package azservicebus
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAMQPMessageUnitTest(t *testing.T) {
+	t.Run("Default", func(t *testing.T) {
+		msg := &AMQPMessage{}
+		amqpMessage := msg.toAMQPMessage()
+
+		// we duplicate/inflate these since we modify them
+		// in various parts of the API.
+		require.NotNil(t, amqpMessage.Properties)
+		require.NotNil(t, amqpMessage.Annotations)
+	})
+}

sdk/messaging/azservicebus/example_receiver_test.go

@@ -116,3 +116,41 @@ func ExampleReceiver_ReceiveMessages() {
 		fmt.Printf("Received and completed the message\n")
 	}
 }
+
+func ExampleReceiver_ReceiveMessages_amqpMessage() {
+	// AMQP is the underlying protocol for all interaction with Service Bus.
+	// You can, if needed, send and receive messages that have a 1:1 correspondence
+	// with an AMQP message. This gives you full control over details that are not
+	// exposed via the azservicebus.ReceivedMessage type.
+
+	messages, err := receiver.ReceiveMessages(context.TODO(), 1, nil)
+
+	if err != nil {
+		panic(err)
+	}
+
+	// NOTE: For this example we'll assume we received at least one message.
+
+	// Every received message carries a RawAMQPMessage.
+	rawAMQPMessage := messages[0].RawAMQPMessage
+
+	// All the various body encodings available for AMQP messages are exposed via Body
+	_ = rawAMQPMessage.Body.Data
+	_ = rawAMQPMessage.Body.Value
+	_ = rawAMQPMessage.Body.Sequence
+
+	// delivery and message annotations
+	_ = rawAMQPMessage.DeliveryAnnotations
+	_ = rawAMQPMessage.MessageAnnotations
+
+	// headers and footers
+	_ = rawAMQPMessage.Header
+	_ = rawAMQPMessage.Footer
+
+	// Settlement (if in azservicebus.ReceiveModePeekLockMode) stil works on the ReceivedMessage.
+	err = receiver.CompleteMessage(context.TODO(), messages[0], nil)
+
+	if err != nil {
+		panic(err)
+	}
+}