Repository: Azure/azure-service-bus-go (go)

Issue:
Custom AMQP annotations cannot be sent/received
Raw AMQP messages have an [Annotations](https://pkg.go.dev/github.com/Azure/go-amqp?tab=doc#Annotations) map which can contain arbitrary key/value pairs. This map is used by both Event Hubs and IoT Hub to set their own pieces of metadata, but it is also used by other systems and can be interacted with directly when working with other languages. As a result, other systems may produce or expect additional data in the annotations beyond the specific list provided by IoT Hub and Event Hubs.

### Expected Behavior

Custom annotations can be sent and received in the corresponding [Message](https://pkg.go.dev/github.com/Azure/azure-service-bus-go?tab=doc#Message) payload, regardless of whether they are part of the built-in ones placed in the [SystemProperties](https://pkg.go.dev/github.com/Azure/azure-service-bus-go?tab=doc#SystemProperties) struct.

### Actual Behavior

Only a restricted subset of annotations are placed in the [SystemProperties](https://pkg.go.dev/github.com/Azure/azure-service-bus-go?tab=doc#SystemProperties) struct. The rest of the annotations are hidden from the user and can neither be set nor retrieved.

### Environment
- OS: All (tested on WSL, MacOS)
- Go version: 1.13
- Version of Library: 0.10.0

Related: https://github.com/Azure/azure-event-hubs-go/issues/173

Pull request:
Support send and receiving custom annotations in System Properties
Adds support for sending an receiving custom annotations on messages by adding an `Annotations` field to the `SystemProperties` present in the message. fixes #166 

### Fix or Enhancement?

enhancement

- [x] All tests passed
- [x] Add change to `changelog.md`

### Environment
- OS: MacOS
- Go version: 1.13

Patch:
diff --git a/azuredeploy.tf b/azuredeploy.tf
index c54cbac..7071eee 100644
--- a/azuredeploy.tf
+++ b/azuredeploy.tf
@@ -63,9 +63,9 @@ resource "random_string" "secret" {
 resource "azuread_application" "test" {
   count                       = data.azurerm_client_config.current.service_principal_application_id == "" ? 1 : 0
   name                        = "servicebustest"
-  homepage                    = "https://servicebustest"
-  identifier_uris             = ["https://servicebustest"]
-  reply_urls                  = ["https://servicebustest"]
+  homepage                    = "https://servicebustest-${random_string.name.result}"
+  identifier_uris             = ["https://servicebustest-${random_string.name.result}"]
+  reply_urls                  = ["https://servicebustest-${random_string.name.result}"]
   available_to_other_tenants  = false
   oauth2_allow_implicit_flow  = true
 }
@@ -160,4 +160,4 @@ output "AZURE_CLIENT_ID" {
 output "AZURE_CLIENT_SECRET" {
   value     = compact(concat(azuread_service_principal_password.test.*.value, list(var.azure_client_secret)))[0]
   sensitive = true
-}
\ No newline at end of file
+}
diff --git a/changelog.md b/changelog.md
index 53ad447..ad195c3 100644
--- a/changelog.md
+++ b/changelog.md
@@ -1,5 +1,11 @@
 # Change Log
 
+## `v0.10.2`
+- add support for sending and receiving custom annotations
+- added some missing AMQP span attributes
+- fixed propagation of sender/receiver close context
+- don't panic on empty AMQP payloads
+
 ## `v0.10.1`
 - fix nil pointer dereference for concurrent uses of Send() [issue #149](https://github.com/Azure/azure-service-bus-go/issues/149)
 - fix nil pointer dereference when there are no listeners [PR #151](https://github.com/Azure/azure-service-bus-go/pull/151)
@@ -59,4 +65,4 @@
 - Ensure senders wait for message disposition before returning
 
 ## `v0.1.0`
-- initial tag for Service Bus which includes Queues, Topics and Subscriptions using AMQP
\ No newline at end of file
+- initial tag for Service Bus which includes Queues, Topics and Subscriptions using AMQP
diff --git a/message.go b/message.go
index 233bd9c..ea444f8 100644
--- a/message.go
+++ b/message.go
@@ -68,15 +68,16 @@ type (
 
 	// SystemProperties are used to store properties that are set by the system.
 	SystemProperties struct {
-		LockedUntil            *time.Time `mapstructure:"x-opt-locked-until"`
-		SequenceNumber         *int64     `mapstructure:"x-opt-sequence-number"`
-		PartitionID            *int16     `mapstructure:"x-opt-partition-id"`
-		PartitionKey           *string    `mapstructure:"x-opt-partition-key"`
-		EnqueuedTime           *time.Time `mapstructure:"x-opt-enqueued-time"`
-		DeadLetterSource       *string    `mapstructure:"x-opt-deadletter-source"`
-		ScheduledEnqueueTime   *time.Time `mapstructure:"x-opt-scheduled-enqueue-time"`
-		EnqueuedSequenceNumber *int64     `mapstructure:"x-opt-enqueue-sequence-number"`
-		ViaPartitionKey        *string    `mapstructure:"x-opt-via-partition-key"`
+		LockedUntil            *time.Time             `mapstructure:"x-opt-locked-until"`
+		SequenceNumber         *int64                 `mapstructure:"x-opt-sequence-number"`
+		PartitionID            *int16                 `mapstructure:"x-opt-partition-id"`
+		PartitionKey           *string                `mapstructure:"x-opt-partition-key"`
+		EnqueuedTime           *time.Time             `mapstructure:"x-opt-enqueued-time"`
+		DeadLetterSource       *string                `mapstructure:"x-opt-deadletter-source"`
+		ScheduledEnqueueTime   *time.Time             `mapstructure:"x-opt-scheduled-enqueue-time"`
+		EnqueuedSequenceNumber *int64                 `mapstructure:"x-opt-enqueue-sequence-number"`
+		ViaPartitionKey        *string                `mapstructure:"x-opt-via-partition-key"`
+		Annotations            map[string]interface{} `mapstructure:"-"`
 	}
 
 	mapStructureTag struct {
@@ -364,11 +365,15 @@ func (m *Message) toMsg() (*amqp.Message, error) {
 	}
 
 	if m.SystemProperties != nil {
+		// Set the raw annotations first (they may be nil) and add the explicit
+		// system properties second to ensure they're set properly.
+		amqpMsg.Annotations = addMapToAnnotations(amqpMsg.Annotations, m.SystemProperties.Annotations)
+
 		sysPropMap, err := encodeStructureToMap(m.SystemProperties)
 		if err != nil {
 			return nil, err
 		}
-		amqpMsg.Annotations = annotationsFromMap(sysPropMap)
+		amqpMsg.Annotations = addMapToAnnotations(amqpMsg.Annotations, sysPropMap)
 	}
 
 	if m.LockToken != nil {
@@ -381,8 +386,11 @@ func (m *Message) toMsg() (*amqp.Message, error) {
 	return amqpMsg, nil
 }
 
-func annotationsFromMap(m map[string]interface{}) amqp.Annotations {
-	a := make(amqp.Annotations)
+func addMapToAnnotations(a amqp.Annotations, m map[string]interface{}) amqp.Annotations {
+	if a == nil && len(m) > 0 {
+		a = make(amqp.Annotations)
+	}
+
 	for key, val := range m {
 		a[key] = val
 	}
@@ -434,6 +442,28 @@ func newMessage(data []byte, amqpMsg *amqp.Message) (*Message, error) {
 		if err := mapstructure.Decode(amqpMsg.Annotations, &msg.SystemProperties); err != nil {
 			return msg, err
 		}
+
+		// If we didn't populate any system properties, set up the struct so we
+		// can put the annotations in it
+		if msg.SystemProperties == nil {
+			msg.SystemProperties = new(SystemProperties)
+		}
+
+		// Take all string-keyed annotations because the protocol reserves all
+		// numeric keys for itself and there are no numeric keys defined in the
+		// protocol today:
+		//
+		//	http://www.amqp.org/sites/amqp.org/files/amqp.pdf (section 3.2.10)
+		//
+		// This approach is also consistent with the behavior of .NET:
+		//
+		//	https://docs.microsoft.com/en-us/dotnet/api/azure.messaging.eventhubs.eventdata.systemproperties?view=azure-dotnet#Azure_Messaging_EventHubs_EventData_SystemProperties
+		msg.SystemProperties.Annotations = make(map[string]interface{})
+		for key, val := range amqpMsg.Annotations {
+			if s, ok := key.(string); ok {
+				msg.SystemProperties.Annotations[s] = val
+			}
+		}
 	}
 
 	if amqpMsg.DeliveryTag != nil && len(amqpMsg.DeliveryTag) > 0 {
@@ -500,6 +530,11 @@ func encodeStructureToMap(structPointer interface{}) (map[string]interface{}, er
 				return nil, err
 			}
 
+			// Skip any entries with an exclude tag
+			if tag.Name == "-" {
+				continue
+			}
+
 			if tag != nil {
 				switch f.Kind() {
 				case reflect.Ptr:
diff --git a/message_session.go b/message_session.go
index 91c6fc9..e9920df 100644
--- a/message_session.go
+++ b/message_session.go
@@ -8,8 +8,8 @@ import (
 	"time"
 
 	"github.com/Azure/azure-amqp-common-go/v3/rpc"
-	"github.com/devigned/tab"
 	"github.com/Azure/go-amqp"
+	"github.com/devigned/tab"
 )
 
 // MessageSession represents and allows for interaction with a Service Bus Session.
diff --git a/namespace.go b/namespace.go
index 8be745b..7bcd169 100644
--- a/namespace.go
+++ b/namespace.go
@@ -49,7 +49,7 @@ const (
 	//`
 
 	// Version is the semantic version number
-	Version = "0.10.1"
+	Version = "0.10.2"
 
 	rootUserAgent = "/golang-service-bus"
 )
diff --git a/rpc.go b/rpc.go
index 81beaa8..c55f849 100644
--- a/rpc.go
+++ b/rpc.go
@@ -32,8 +32,8 @@ import (
 
 	"github.com/Azure/azure-amqp-common-go/v3/rpc"
 	"github.com/Azure/azure-amqp-common-go/v3/uuid"
-	"github.com/devigned/tab"
 	"github.com/Azure/go-amqp"
+	"github.com/devigned/tab"
 )
 
 type (
diff --git a/session.go b/session.go
index df9cdff..e74dbfb 100644
--- a/session.go
+++ b/session.go
@@ -28,8 +28,8 @@ import (
 	"sync/atomic"
 
 	"github.com/Azure/azure-amqp-common-go/v3/uuid"
-	"github.com/devigned/tab"
 	"github.com/Azure/go-amqp"
+	"github.com/devigned/tab"
 )
 
 type (


Test patch:
diff --git a/message_test.go b/message_test.go
index 4e3c188..1a63da9 100644
--- a/message_test.go
+++ b/message_test.go
@@ -4,9 +4,9 @@ import (
 	"time"
 
 	"github.com/Azure/azure-amqp-common-go/v3/uuid"
+	"github.com/Azure/go-amqp"
 	"github.com/Azure/go-autorest/autorest/to"
 	"github.com/mitchellh/mapstructure"
-	"github.com/Azure/go-amqp"
 )
 
 func (suite *serviceBusSuite) TestMapStructureEncode() {
@@ -74,6 +74,10 @@ func (suite *serviceBusSuite) TestMessageToAMQPMessage() {
 			ScheduledEnqueueTime:   &until,
 			EnqueuedSequenceNumber: to.Int64Ptr(1),
 			ViaPartitionKey:        to.StringPtr("via"),
+			Annotations: map[string]interface{}{
+				"custom":              "annotation",
+				"x-opt-partition-key": "other value",
+			},
 		},
 		UserProperties: map[string]interface{}{
 			"test": "foo",
@@ -100,6 +104,17 @@ func (suite *serviceBusSuite) TestMessageToAMQPMessage() {
 			for key, val := range sysPropMap {
 				suite.Equal(val, aMsg.Annotations[key], key)
 			}
+
+			for key, val := range msg.SystemProperties.Annotations {
+				// The partition key should be overridden by the value in the
+				// base system properties
+				if key == "x-opt-partition-key" {
+					suite.Equal(*msg.SystemProperties.PartitionKey, aMsg.Annotations[key], key)
+					continue
+				}
+
+				suite.Equal(val, aMsg.Annotations[key], key)
+			}
 		}
 
 		for key, val := range msg.UserProperties {
@@ -147,6 +162,7 @@ func (suite *serviceBusSuite) TestAMQPMessageToMessage() {
 			"x-opt-scheduled-enqueue-time":  until,
 			"x-opt-enqueue-sequence-number": int64(1),
 			"x-opt-via-partition-key":       "via",
+			"custom-annotation":             "value",
 		},
 		ApplicationProperties: map[string]interface{}{
 			"test": "foo",
@@ -177,6 +193,10 @@ func (suite *serviceBusSuite) TestAMQPMessageToMessage() {
 			for key, val := range sysPropMap {
 				suite.Equal(val, aMsg.Annotations[key], key)
 			}
+
+			for key, val := range msg.SystemProperties.Annotations {
+				suite.Equal(val, aMsg.Annotations[key], key)
+			}
 		}
 
 		for key, val := range aMsg.ApplicationProperties {