[azservicebus] Some naming changes for raw AMQP message support for consistency with other stacks Azure#18430

Some naming changes to be more consistent with other track 2 SDKs as well as the AMQP spec:

Rename the AMQPMessage to be AMQPAnnotatedMessage (track 2 consistency, also matches terminology in the spec)
Create an AMQPAnnotatedMessageBody type to house the .Value, .Sequence and .Data fields. This matches more with the logical format of an AMQP message, where the Body is comprised of sections (Data, Value, Sequence), rather than them just being free-floating fields on a message.

Author: richardpark-msft

sdk/messaging/azservicebus/CHANGELOG.md

@@ -5,7 +5,9 @@
 ### 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()). 
+  - Send AMQP messages using the new `AMQPAnnotatedMessage` type and `Sender.SendAMQPAnnotatedMessage()`.
+  - AMQP messages can be added to MessageBatch's as well using `MessageBatch.AddAMQPAnnotatedMessage()`.
+  - AMQP messages can be scheduled using `Sender.ScheduleAMQPAnnotatedMessages`.
   - Access the full set of AMQP message properties when receiving using the `ReceivedMessage.RawAMQPMessage` property. 
 
 ### Breaking Changes

sdk/messaging/azservicebus/amqp_message.go

@@ -9,7 +9,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/go-amqp"
 )
 
-// AMQPMessage represents the AMQP message, as received from Service Bus.
+// AMQPAnnotatedMessage 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
 //
@@ -22,18 +22,18 @@ import (
 // - string
 // - bool
 // - time.Time
-type AMQPMessage struct {
+type AMQPAnnotatedMessage 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.
+	// The values of the map are restricted to AMQP simple types, as listed in the comment for AMQPAnnotatedMessage.
 	ApplicationProperties map[string]any
 
 	// Body represents the body of an AMQP message.
-	Body AMQPMessageBody
+	Body AMQPAnnotatedMessageBody
 
 	// 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.
+	// The values of the map are restricted to AMQP simple types, as listed in the comment for AMQPAnnotatedMessage.
 	DeliveryAnnotations map[any]any
 
 	// DeliveryTag corresponds to the delivery-tag property of the TRANSFER frame
@@ -42,19 +42,19 @@ type AMQPMessage struct {
 
 	// 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.
+	// The values of the map are restricted to AMQP simple types, as listed in the comment for AMQPAnnotatedMessage.
 	Footer map[any]any
 
 	// Header is the transport headers for this AMQP message.
-	Header *AMQPMessageHeader
+	Header *AMQPAnnotatedMessageHeader
 
 	// 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.
+	// The values of the map are restricted to AMQP simple types, as listed in the comment for AMQPAnnotatedMessage.
 	MessageAnnotations map[any]any
 
 	// Properties corresponds to the properties section of an AMQP message.
-	Properties *AMQPMessageProperties
+	Properties *AMQPAnnotatedMessageProperties
 
 	linkName string
 
@@ -64,10 +64,10 @@ type AMQPMessage struct {
 	inner *amqp.Message
 }
 
-// AMQPMessageProperties represents the properties of an AMQP message.
+// AMQPAnnotatedMessageProperties 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 {
+type AMQPAnnotatedMessageProperties struct {
 	// AbsoluteExpiryTime corresponds to the 'absolute-expiry-time' property.
 	AbsoluteExpiryTime *time.Time
 
@@ -110,29 +110,29 @@ type AMQPMessageProperties struct {
 	UserID []byte
 }
 
-// AMQPMessageBody represents the body of an AMQP message.
+// AMQPAnnotatedMessageBody 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 {
+type AMQPAnnotatedMessageBody 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.
+	// The values of the slices are are restricted to AMQP simple types, as listed in the comment for AMQPAnnotatedMessage.
 	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,
+	// The type of Value can be any of the AMQP simple types, as listed in the comment for AMQPAnnotatedMessage,
 	// as well as slices or maps of AMQP simple types.
 	Value any
 }
 
-// AMQPMessageHeader carries standard delivery details about the transfer
+// AMQPAnnotatedMessageHeader 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 {
+type AMQPAnnotatedMessageHeader struct {
 	// DeliveryCount is the number of unsuccessful previous attempts to deliver this message.
 	// It corresponds to the 'delivery-count' property.
 	DeliveryCount uint32
@@ -152,7 +152,7 @@ type AMQPMessageHeader struct {
 
 // toAMQPMessage converts between our (azservicebus) AMQP message
 // to the underlying message used by go-amqp.
-func (am *AMQPMessage) toAMQPMessage() *amqp.Message {
+func (am *AMQPAnnotatedMessage) toAMQPMessage() *amqp.Message {
 	var header *amqp.MessageHeader
 
 	if am.Header != nil {
@@ -221,11 +221,11 @@ func copyAnnotations(src map[any]any) amqp.Annotations {
 	return dest
 }
 
-func newAMQPMessage(goAMQPMessage *amqp.Message) *AMQPMessage {
-	var header *AMQPMessageHeader
+func newAMQPAnnotatedMessage(goAMQPMessage *amqp.Message) *AMQPAnnotatedMessage {
+	var header *AMQPAnnotatedMessageHeader
 
 	if goAMQPMessage.Header != nil {
-		header = &AMQPMessageHeader{
+		header = &AMQPAnnotatedMessageHeader{
 			DeliveryCount: goAMQPMessage.Header.DeliveryCount,
 			Durable:       goAMQPMessage.Header.Durable,
 			FirstAcquirer: goAMQPMessage.Header.FirstAcquirer,
@@ -234,10 +234,10 @@ func newAMQPMessage(goAMQPMessage *amqp.Message) *AMQPMessage {
 		}
 	}
 
-	var properties *AMQPMessageProperties
+	var properties *AMQPAnnotatedMessageProperties
 
 	if goAMQPMessage.Properties != nil {
-		properties = &AMQPMessageProperties{
+		properties = &AMQPAnnotatedMessageProperties{
 			AbsoluteExpiryTime: goAMQPMessage.Properties.AbsoluteExpiryTime,
 			ContentEncoding:    goAMQPMessage.Properties.ContentEncoding,
 			ContentType:        goAMQPMessage.Properties.ContentType,
@@ -260,10 +260,10 @@ func newAMQPMessage(goAMQPMessage *amqp.Message) *AMQPMessage {
 		footer = (map[any]any)(goAMQPMessage.Footer)
 	}
 
-	return &AMQPMessage{
+	return &AMQPAnnotatedMessage{
 		MessageAnnotations:    map[any]any(goAMQPMessage.Annotations),
 		ApplicationProperties: goAMQPMessage.ApplicationProperties,
-		Body: AMQPMessageBody{
+		Body: AMQPAnnotatedMessageBody{
 			Data:     goAMQPMessage.Data,
 			Sequence: goAMQPMessage.Sequence,
 			Value:    goAMQPMessage.Value,

sdk/messaging/azservicebus/amqp_message_test.go

@@ -9,9 +9,9 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestAMQPMessageUnitTest(t *testing.T) {
+func TestAMQPAnnotatedMessageUnitTest(t *testing.T) {
 	t.Run("Default", func(t *testing.T) {
-		msg := &AMQPMessage{}
+		msg := &AMQPAnnotatedMessage{}
 		amqpMessage := msg.toAMQPMessage()
 
 		// we duplicate/inflate these since we modify them

sdk/messaging/azservicebus/example_receiver_test.go

@@ -132,7 +132,7 @@ func ExampleReceiver_ReceiveMessages_amqpMessage() {
 	// NOTE: For this example we'll assume we received at least one message.
 
 	// Every received message carries a RawAMQPMessage.
-	rawAMQPMessage := messages[0].RawAMQPMessage
+	var rawAMQPMessage *azservicebus.AMQPAnnotatedMessage = messages[0].RawAMQPMessage
 
 	// All the various body encodings available for AMQP messages are exposed via Body
 	_ = rawAMQPMessage.Body.Data

sdk/messaging/azservicebus/example_sender_test.go

@@ -91,14 +91,14 @@ func ExampleSender_ScheduleMessages() {
 	exitOnError("Failed to schedule messages using SendMessage", err)
 }
 
-func ExampleSender_SendAMQPMessage() {
+func ExampleSender_SendAMQPAnnotatedMessage() {
 	// 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.
 
-	message := &azservicebus.AMQPMessage{
-		Body: azservicebus.AMQPMessageBody{
+	message := &azservicebus.AMQPAnnotatedMessage{
+		Body: azservicebus.AMQPAnnotatedMessageBody{
 			// there are three kinds of different body encodings
 			// Data, Value and Sequence. See the azservicebus.AMQPMessageBody
 			// documentation for more details.
@@ -114,7 +114,7 @@ func ExampleSender_SendAMQPMessage() {
 		},
 	}
 
-	err := sender.SendAMQPMessage(context.TODO(), message, nil)
+	err := sender.SendAMQPAnnotatedMessage(context.TODO(), message, nil)
 
 	if err != nil {
 		panic(err)

sdk/messaging/azservicebus/message.go

@@ -124,7 +124,7 @@ type ReceivedMessage struct {
 	// to properties that are not exposed by ReceivedMessage such as payloads encoded into the
 	// Value or Sequence section, payloads sent as multiple Data sections, as well as Footer
 	// and Header fields.
-	RawAMQPMessage *AMQPMessage
+	RawAMQPMessage *AMQPAnnotatedMessage
 
 	// deferred indicates we received it using ReceiveDeferredMessages. These messages
 	// will still go through the normal Receiver.Settle functions but internally will
@@ -314,7 +314,7 @@ func (m *Message) toAMQPMessage() *amqp.Message {
 // serialized byte array in the Data section of the messsage.
 func newReceivedMessage(amqpMsg *amqp.Message) *ReceivedMessage {
 	msg := &ReceivedMessage{
-		RawAMQPMessage: newAMQPMessage(amqpMsg),
+		RawAMQPMessage: newAMQPAnnotatedMessage(amqpMsg),
 		State:          MessageStateActive,
 	}
 

sdk/messaging/azservicebus/message_batch.go

@@ -54,17 +54,17 @@ func (mb *MessageBatch) AddMessage(m *Message, options *AddMessageOptions) error
 	return mb.addAMQPMessage(m)
 }
 
-// AddAMQPMessageOptions contains optional parameters for the AddAMQPMessage function.
-type AddAMQPMessageOptions struct {
+// AddAMQPAnnotatedMessageOptions contains optional parameters for the AddAMQPAnnotatedMessage function.
+type AddAMQPAnnotatedMessageOptions struct {
 	// For future expansion
 }
 
-// AddAMQPMessage adds a message to the batch if the message will not exceed the max size of the batch
+// AddAMQPAnnotatedMessage adds a message to the batch if the message will not exceed the max size of the batch
 // Returns:
 // - ErrMessageTooLarge if the message cannot fit
 // - a non-nil error for other failures
 // - nil, otherwise
-func (mb *MessageBatch) AddAMQPMessage(m *AMQPMessage, options *AddAMQPMessageOptions) error {
+func (mb *MessageBatch) AddAMQPAnnotatedMessage(m *AMQPAnnotatedMessage, options *AddAMQPAnnotatedMessageOptions) error {
 	return mb.addAMQPMessage(m)
 }
 

sdk/messaging/azservicebus/sender.go

@@ -63,15 +63,20 @@ type SendMessageOptions struct {
 // SendMessage sends a Message to a queue or topic.
 // If the operation fails it can return an *azservicebus.Error type if the failure is actionable.
 func (s *Sender) SendMessage(ctx context.Context, message *Message, options *SendMessageOptions) error {
-	return s.sendMessage(ctx, message, options)
+	return s.sendMessage(ctx, message)
 }
 
-// SendAMQPMessage sends an AMQPMessage to a queue or topic.
+// SendAMQPAnnotatedMessageOptions contains optional parameters for the SendAMQPAnnotatedMessage function.
+type SendAMQPAnnotatedMessageOptions struct {
+	// For future expansion
+}
+
+// SendAMQPAnnotatedMessage sends an AMQPMessage to a queue or topic.
 // Using an AMQPMessage allows for advanced use cases, like payload encoding, as well as better
 // interoperability with pure AMQP clients.
 // If the operation fails it can return an *azservicebus.Error type if the failure is actionable.
-func (s *Sender) SendAMQPMessage(ctx context.Context, message *AMQPMessage, options *SendMessageOptions) error {
-	return s.sendMessage(ctx, message, options)
+func (s *Sender) SendAMQPAnnotatedMessage(ctx context.Context, message *AMQPAnnotatedMessage, options *SendAMQPAnnotatedMessageOptions) error {
+	return s.sendMessage(ctx, message)
 }
 
 // SendMessageBatchOptions contains optional parameters for the SendMessageBatch function.
@@ -103,16 +108,16 @@ func (s *Sender) ScheduleMessages(ctx context.Context, messages []*Message, sche
 	return scheduleMessages(ctx, s.links, s.retryOptions, messages, scheduledEnqueueTime)
 }
 
-// ScheduleAMQPMessagesOptions contains optional parameters for the ScheduleAMQPMessages function.
-type ScheduleAMQPMessagesOptions struct {
+// ScheduleAMQPAnnotatedMessagesOptions contains optional parameters for the ScheduleAMQPAnnotatedMessages function.
+type ScheduleAMQPAnnotatedMessagesOptions struct {
 	// For future expansion
 }
 
-// ScheduleAMQPMessages schedules a slice of Messages to appear on Service Bus Queue/Subscription at a later time.
+// ScheduleAMQPAnnotatedMessages schedules a slice of Messages to appear on Service Bus Queue/Subscription at a later time.
 // Returns the sequence numbers of the messages that were scheduled.  Messages that haven't been
 // delivered can be cancelled using `Receiver.CancelScheduleMessage(s)`
 // If the operation fails it can return an *azservicebus.Error type if the failure is actionable.
-func (s *Sender) ScheduleAMQPMessages(ctx context.Context, messages []*AMQPMessage, scheduledEnqueueTime time.Time, options *ScheduleAMQPMessagesOptions) ([]int64, error) {
+func (s *Sender) ScheduleAMQPAnnotatedMessages(ctx context.Context, messages []*AMQPAnnotatedMessage, scheduledEnqueueTime time.Time, options *ScheduleAMQPAnnotatedMessagesOptions) ([]int64, error) {
 	return scheduleMessages(ctx, s.links, s.retryOptions, messages, scheduledEnqueueTime)
 }
 
@@ -161,7 +166,7 @@ func (s *Sender) Close(ctx context.Context) error {
 	return s.links.Close(ctx, true)
 }
 
-func (s *Sender) sendMessage(ctx context.Context, message amqpCompatibleMessage, options *SendMessageOptions) error {
+func (s *Sender) sendMessage(ctx context.Context, message amqpCompatibleMessage) error {
 	err := s.links.Retry(ctx, EventSender, "SendMessage", func(ctx context.Context, lwid *internal.LinksWithID, args *utils.RetryFnArgs) error {
 		return lwid.Sender.Send(ctx, message.toAMQPMessage())
 	}, RetryOptions(s.retryOptions))