diff --git a/sdk/eventhub/Azure.Messaging.EventHubs.Processor/CHANGELOG.md b/sdk/eventhub/Azure.Messaging.EventHubs.Processor/CHANGELOG.md index 27add15da9541..f5a047e6bace7 100755 --- a/sdk/eventhub/Azure.Messaging.EventHubs.Processor/CHANGELOG.md +++ b/sdk/eventhub/Azure.Messaging.EventHubs.Processor/CHANGELOG.md @@ -2,6 +2,29 @@ ## 5.5.0-beta.2 (Unreleased) +### Acknowledgments + +Thank you to our developer community members who helped to make the Event Hubs client libraries better with their contributions to this release: + +- Daniel Marbach _([GitHub](https://github.com/danielmarbach))_ + +### Changes + +#### Features Added + +- When stopping, the `EventProcessorClient` will now attempt to force-close the connection to the Event Hubs service to abort in-process read operations blocked on their timeout. This should significantly help reduce the amount of time the processor takes to stop in many scenarios. _(Based on a community prototype contribution, courtesy of [danielmarbach](https://github.com/danielmarbach))_ + +- When the `EventProcessorClient` detects a partition being stolen outside of a load balancing cycle, it will immediately surrender ownership rather than waiting for a load balancing cycle to confirm the ownership change. This will help reduce event duplication from overlapping ownership of processors. + +- The `ConnectionOptions` available when creating a processor now support registering a callback delegate for participating in the validation of SSL certificates when connections are established. This delegate may override the built-in validation and allow or deny certificates based on application-specific logic. + +- The `ConnectionOptions` available when creating a processor now support setting a custom size for the send and receive buffers of the transport. + +#### Key Bugs Fixed + +- The `EventProcessorClient` will now properly respect another another consumer stealing ownership of a partition when the service forcibly terminates the active link in the background. Previously, the client did not observe the error directly and attempted to recover the faulted link which reasserted ownership and caused the partition to "bounce" between owners until a load balancing cycle completed. + +- The `EventProcessorClient` will now be less aggressive when considering whether or not to steal a partition, doing so only when it will correct an imbalance and preferring the status quo when the overall distribution would not change. This will help reduce event duplication due to partitions moving between owners. ## 5.5.0-beta.1 (2021-06-08) @@ -13,7 +36,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### New Features +#### Features Added - When stopping, the `EventProcessorClient` will now attempt to force-close the connection to the Event Hubs service to abort in-process read operations blocked on their timeout. This should significantly help reduce the amount of time the processor takes to stop in many scenarios. _(Based on a community prototype contribution, courtesy of [danielmarbach](https://github.com/danielmarbach))_ @@ -23,7 +46,7 @@ Thank you to our developer community members who helped to make the Event Hubs c - The `ConnectionOptions` available when creating a processor now support setting a custom size for the send and receive buffers of the transport. -#### Key Bug Fixes +#### Key Bugs Fixed - The `EventProcessorClient` will now properly respect another another consumer stealing ownership of a partition when the service forcibly terminates the active link in the background. Previously, the client did not observe the error directly and attempted to recover the faulted link which reasserted ownership and caused the partition to "bounce" between owners until a load balancing cycle completed. @@ -33,13 +56,13 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### New Features +#### Features Added - The processor will now perform validation of core configuration and permissions at startup, in order to attempt to detect unrecoverable problems more deterministically. Validation is non-blocking and will not delay claiming of partitions. One important note is that validation should be considered point-in-time and best effort; it is not meant to replace monitoring of error handler activity. - Partition initialization has been moved to a background operation. This will allow partitions to be more efficiently managed and speed up ownership claims, especially when using the `LoadBalancingStrategy.Greedy` configuration or when the processor is recovering from some error conditions. -#### Key Bug Fixes +#### Key Bugs Fixed - Dependencies have been updated to resolve security warnings for CVE-2021-26701. _(The Event Hubs client library does not make use of the vulnerable components, directly or indirectly)_ @@ -55,13 +78,13 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### New Features +#### Features Added - The `EventProcessorClient` now supports shared key and shared access signature authentication using the `AzureNamedKeyCredential` and `AzureSasCredential` types in addition to the connection string. Use of the credential allows the shared key or SAS to be updated without the need to create a new processor. - Multiple enhancements were made to the AMQP transport paths for reading events to reduce memory allocations and increase performance. _(A community contribution, courtesy of [danielmarbach](https://github.com/danielmarbach))_ -#### Key Bug Fixes +#### Key Bugs Fixed - The AMQP library used for transport has been updated, fixing several issues including a potential unobserved `ObjectDisposedException` that could cause the host process to crash. _(see: [release notes](https://github.com/Azure/azure-amqp/releases/tag/v2.4.13))_ @@ -75,7 +98,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### Key Bug Fixes +#### Key Bugs Fixed - Fixed an issue where long-lived credentials (more than 49 days) were overflowing refresh timer limits and being rejected. @@ -89,7 +112,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### New Features +#### Features Added - Additional options for tuning load balancing have been added to the `EventProcessorClientOptions`. @@ -101,7 +124,7 @@ Thank you to our developer community members who helped to make the Event Hubs c - Documentation used for auto-completion via Intellisense and other tools has been enhanced in many areas, addressing gaps and commonly asked questions. -#### Key Bug Fixes +#### Key Bugs Fixed - Upgraded the `Microsoft.Azure.Amqp` library to resolve crashes occurring in .NET 5. @@ -111,13 +134,13 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### New Features +#### Features Added - Additional options for tuning load balancing have been added to the `EventProcessorClientOptions`. - Documentation used for auto-completion via Intellisense and other tools has been enhanced in many areas, addressing gaps and commonly asked questions. -#### Key Bug Fixes +#### Key Bugs Fixed - Upgraded the `Microsoft.Azure.Amqp` library to resolve crashes occurring in .NET 5. @@ -127,7 +150,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### Key Bug Fixes +#### Key Bugs Fixed - An issue with package publishing which blocked referencing and use has been fixed. @@ -135,7 +158,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### New Features +#### Features Added - The `EventData` representation has been extended with the ability to treat the `Body` as `BinaryData`. `BinaryData` supports a variety of data transformations and allows the ability to provide serialization logic when sending or receiving events. Any type that derives from `ObjectSerializer`, such as `JsonObjectSerializer` can be used, with Schema Registry support available via the `SchemaRegistryAvroObjectSerializer`. @@ -147,7 +170,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### New Features +#### Features Added - Introduction of an option for the various event consumers allowing the prefetch cache to be filled based on a size-based heuristic rather than a count of events. This feature is considered a special case, helpful in scenarios where the size of events being read is not able to be known or predicted upfront and limiting resource use is valued over consistent and predictable performance. @@ -161,7 +184,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### Key Bug Fixes +#### Key Bugs Fixed - The approach used for creation of checkpoints has been updated to interact with Azure Blob storage more efficiently. This will yield major performance improvements when soft delete was enabled and minor improvements otherwise. @@ -169,7 +192,7 @@ Thank you to our developer community members who helped to make the Event Hubs c - Fixed an issue where failure to create an AMQP link would lead to an AMQP session not being explicitly closed, causing connections to the Event Hubs service to remain open until a garbage collection pass was performed. -#### New Features +#### Features Added - Load balancing will now detect when it has reached a balanced state more accurately; this will allow it to operate more efficiently when `LoadBalancingStrategy.Greedy` is in use. diff --git a/sdk/eventhub/Azure.Messaging.EventHubs.Shared/src/Resources.Designer.cs b/sdk/eventhub/Azure.Messaging.EventHubs.Shared/src/Resources.Designer.cs index 41e3b800bc4cd..27d5f83044c62 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs.Shared/src/Resources.Designer.cs +++ b/sdk/eventhub/Azure.Messaging.EventHubs.Shared/src/Resources.Designer.cs @@ -780,5 +780,27 @@ internal static string DictionaryKeyNotFoundMask return ResourceManager.GetString("DictionaryKeyNotFoundMask", resourceCulture); } } + + /// + /// Looks up a localized string similar to {0} is not a supported value body type.. + /// + internal static string InvalidAmqpMessageValueBodyMask + { + get + { + return ResourceManager.GetString("InvalidAmqpMessageValueBodyMask", resourceCulture); + } + } + + /// + /// Looks up a localized string similar to The {0} key `{1}` has a value of type `{2}` which is not supported for AMQP transport.. + /// + internal static string InvalidAmqpMessageDictionaryTypeMask + { + get + { + return ResourceManager.GetString("InvalidAmqpMessageDictionaryTypeMask", resourceCulture); + } + } } } diff --git a/sdk/eventhub/Azure.Messaging.EventHubs.Shared/src/Resources.resx b/sdk/eventhub/Azure.Messaging.EventHubs.Shared/src/Resources.resx index 13e7214092763..a71183654e770 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs.Shared/src/Resources.resx +++ b/sdk/eventhub/Azure.Messaging.EventHubs.Shared/src/Resources.resx @@ -309,4 +309,10 @@ The given key '{0}' was not present in the dictionary. + + {0} is not a supported value body type. + + + The {0} key `{1}` has a value of type `{2}` which is not supported for AMQP transport. + diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/CHANGELOG.md b/sdk/eventhub/Azure.Messaging.EventHubs/CHANGELOG.md index d79f55449bd70..9e415237a0e6e 100755 --- a/sdk/eventhub/Azure.Messaging.EventHubs/CHANGELOG.md +++ b/sdk/eventhub/Azure.Messaging.EventHubs/CHANGELOG.md @@ -2,6 +2,43 @@ ## 5.5.0-beta.2 (Unreleased) +### Acknowledgments + +Thank you to our developer community members who helped to make the Event Hubs client libraries better with their contributions to this release: + +- Daniel Marbach _([GitHub](https://github.com/danielmarbach))_ + +### Changes + +#### Features Added + +- The `EventData` type offers a curated set of the information available for messages using the AMQP protocol. While this results in a simpler and more easily understood API surface for an event, it limits interoperability with other message brokers. To support heterogeneous environments or those with specialized needs, the full AMQP message is now available using the `GetRawAmqpMessage` method. _(Based on a community prototype contribution, courtesy of [danielmarbach](https://github.com/danielmarbach))_ + +- `EventData` now supports construction using a `string` to specify the event body; this will be represented as a set of UTF-8 encoded bytes for transport. + +- `EventData` has been extended to include properties for applications to assign a `MessageId`, `ContentType`, and `CorrelationId` as well-known members rather than embedding them in the `Properties` dictionary. It is important to note that these properties are intended for application use and are not recognized by the Event Hubs service. + +- When stopping, the `EventProcessor` will now attempt to force-close the connection to the Event Hubs service to abort in-process read operations blocked on their timeout. This should significantly help reduce the amount of time the processor takes to stop in many scenarios. _(Based on a community prototype contribution, courtesy of [danielmarbach](https://github.com/danielmarbach))_ + +- When the `EventProcessor` detects a partition being stolen outside of a load balancing cycle, it will immediately surrender ownership rather than waiting for a load balancing cycle to confirm the ownership change. This will help reduce event duplication from overlapping ownership of processors. + +- The `EventProcessor` now exposes the `ListPartitionIdsAsync` method, allowing custom processors to control the set of partitions known to the processor. This can be used to reduce complexity when a custom processor is directly assigned a set of partitions to process rather than using load balancing to control ownership. + +- The `ConnectionOptions` available when creating client types now support registering a callback delegate for participating in the validation of SSL certificates when connections are established. This delegate may override the built-in validation and allow or deny certificates based on application-specific logic. + +- The `ConnectionOptions` available when creating client types now support setting a custom size for the send and receive buffers of the transport. + +- Additional verbose logging has been added to allow monitoring of lower-level AMQP operations such as creating links, terminal exceptions that fault a link without an active operation, and when the service force-closes links. + +#### Key Bugs Fixed + +- The `EventProcessor` will now properly respect another another consumer stealing ownership of a partition when the service forcibly terminates the active link in the background. Previously, the client did not observe the error directly and attempted to recover the faulted link which reasserted ownership and caused the partition to "bounce" between owners until a load balancing cycle completed. + +- The `EventProcessor` will now be less aggressive when considering whether or not to steal a partition, doing so only when it will correct an imbalance and preferring the status quo when the overall distribution would not change. This will help reduce event duplication due to partitions moving between owners. + +- The `EventHubConsumerClient` and `PartitionReceiver` will now properly surface an exception when another another consumer stealing ownership of a partition when the service forcibly terminates the active link in the background. Previously, the client did not observe the error directly and did not make callers attempted to recover the faulted link which reasserted ownership and caused the partition to "bounce" between owners until a load balancing cycle completed. + +- The retry policy used by clients will no longer overflow the `TimeSpan` maximum when using an `Exponential` strategy with a large number of retries and long delay set. ## 5.5.0-beta.1 (2021-06-08) @@ -13,7 +50,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### New Features +#### Features Added - When stopping, the `EventProcessor` will now attempt to force-close the connection to the Event Hubs service to abort in-process read operations blocked on their timeout. This should significantly help reduce the amount of time the processor takes to stop in many scenarios. _(Based on a community prototype contribution, courtesy of [danielmarbach](https://github.com/danielmarbach))_ @@ -27,7 +64,7 @@ Thank you to our developer community members who helped to make the Event Hubs c - Additional verbose logging has been added to allow monitoring of lower-level AMQP operations such as creating links, terminal exceptions that fault a link without an active operation, and when the service force-closes links. -#### Key Bug Fixes +#### Key Bugs Fixed - The `EventProcessor` will now properly respect another another consumer stealing ownership of a partition when the service forcibly terminates the active link in the background. Previously, the client did not observe the error directly and attempted to recover the faulted link which reasserted ownership and caused the partition to "bounce" between owners until a load balancing cycle completed. @@ -41,13 +78,13 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### New Features +#### Features Added - `EventProcessor` will now perform validation of core configuration and permissions at startup, in order to attempt to detect unrecoverable problems more deterministically. Validation is non-blocking and will not delay claiming of partitions. One important note is that validation should be considered point-in-time and best effort; it is not meant to replace monitoring of error handler activity. - Partition initialization for `EventProcessor` has been moved to a background operation. This will allow partitions to be more efficiently managed and speed up ownership claims, especially when using the `LoadBalancingStrategy.Greedy` configuration or when the processor is recovering from some error conditions. -#### Key Bug Fixes +#### Key Bugs Fixed - Dependencies have been updated to resolve security warnings for CVE-2021-26701. _(The Event Hubs client library does not make use of the vulnerable components, directly or indirectly)_ @@ -67,7 +104,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### New Features +#### Features Added - The Event Hubs clients now support shared key and shared access signature authentication using the `AzureNamedKeyCredential` and `AzureSasCredential` types in addition to the connection string. Use of the credential allows the shared key or SAS to be updated without the need to create a new Event Hubs client. @@ -77,7 +114,7 @@ Thank you to our developer community members who helped to make the Event Hubs c - Multiple enhancements were made to the transport paths for publishing and reading events to reduce memory allocations and increase performance. _(A community contribution, courtesy of [danielmarbach](https://github.com/danielmarbach))_ -#### Key Bug Fixes +#### Key Bugs Fixed - The AMQP library used for transport has been updated, fixing several issues including a potential unobserved `ObjectDisposedException` that could cause the host process to crash. _(see: [release notes](https://github.com/Azure/azure-amqp/releases/tag/v2.4.13))_ @@ -85,7 +122,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### New Features +#### Features Added - Returned the idempotent publishing feature to the public API surface. @@ -93,7 +130,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### Key Bug Fixes +#### Key Bugs Fixed - Fixed an issue where long-lived credentials (more than 49 days) were overflowing refresh timer limits and being rejected. @@ -107,7 +144,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### New Features +#### Features Added - Connection strings can now be parsed into their key/value pairs using the `EventHubsConnectionStringProperties` class. @@ -121,7 +158,7 @@ Thank you to our developer community members who helped to make the Event Hubs c - Documentation used for auto-completion via Intellisense and other tools has been enhanced in many areas, addressing gaps and commonly asked questions. -#### Key Bug Fixes +#### Key Bugs Fixed - Upgraded the `Microsoft.Azure.Amqp` library to resolve crashes occurring in .NET 5. @@ -135,7 +172,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### New Features +#### Features Added - Connection strings can now be parsed into their key/value pairs using the `EventHubsConnectionStringProperties` class. @@ -143,7 +180,7 @@ Thank you to our developer community members who helped to make the Event Hubs c - Documentation used for auto-completion via Intellisense and other tools has been enhanced in many areas, addressing gaps and commonly asked questions. -#### Key Bug Fixes +#### Key Bugs Fixed - The `EventHubsException.ToString` result will now properly follow the format of other .NET exception output. @@ -155,7 +192,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### Key Bug Fixes +#### Key Bugs Fixed - An issue with package publishing which blocked referencing and use has been fixed. @@ -163,7 +200,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### New Features +#### Features Added - The `EventData` representation has been extended with the ability to treat the `Body` as `BinaryData`. `BinaryData` supports a variety of data transformations and allows the ability to provide serialization logic when sending or receiving events. Any type that derives from `ObjectSerializer`, such as `JsonObjectSerializer` can be used, with Schema Registry support available via the `SchemaRegistryAvroObjectSerializer`. @@ -177,7 +214,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### New Features +#### Features Added - Introduction of an option for the various event consumers allowing the prefetch cache to be filled based on a size-based heuristic rather than a count of events. This feature is considered a special case, helpful in scenarios where the size of events being read is not able to be known or predicted upfront and limiting resource use is valued over consistent and predictable performance. @@ -191,7 +228,7 @@ Thank you to our developer community members who helped to make the Event Hubs c ### Changes -#### Key Bug Fixes +#### Key Bugs Fixed - The underlying AMQP library has been enhanced for more efficient resource usage; this will result in a noticeable reduction in memory use in common consuming scenarios. (A community contribution, courtesy of _[danielmarbach](https://github.com/danielmarbach))_ @@ -201,7 +238,7 @@ Thank you to our developer community members who helped to make the Event Hubs c - Fixed an issue where failure to create an AMQP link would lead to an AMQP session not being explicitly closed, causing connections to the Event Hubs service to remain open until a garbage collection pass was performed. -#### New Features +#### Features Added - The `EventProcessor` now supports a configurable strategy for load balancing, allowing control over whether it claims ownership of partitions in a balanced manner _(default)_ or more aggressively. The strategy may be set in the `EventProcessorOptions` when creating the processor. More details about strategies can be found in the associated [documentation](https://docs.microsoft.com/dotnet/api/azure.messaging.eventhubs.processor.loadbalancingstrategy?view=azure-dotnet). diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/src/Amqp/AmqpMessageConverter.cs b/sdk/eventhub/Azure.Messaging.EventHubs/src/Amqp/AmqpMessageConverter.cs index 0b6e7419c47d8..f61413a15d764 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs/src/Amqp/AmqpMessageConverter.cs +++ b/sdk/eventhub/Azure.Messaging.EventHubs/src/Amqp/AmqpMessageConverter.cs @@ -9,6 +9,7 @@ using System.Runtime.InteropServices; using System.Runtime.Serialization; using Azure.Core; +using Azure.Core.Amqp; using Azure.Messaging.EventHubs.Diagnostics; using Microsoft.Azure.Amqp; using Microsoft.Azure.Amqp.Encoding; @@ -26,6 +27,23 @@ internal class AmqpMessageConverter /// The size, in bytes, to use as a buffer for stream operations. private const int StreamBufferSizeInBytes = 512; + /// The set of key names for annotations known to be DateTime-based system properties. + private static readonly HashSet SystemPropertyDateTimeKeys = new() + { + AmqpProperty.EnqueuedTime.ToString(), + AmqpProperty.PartitionLastEnqueuedTimeUtc.ToString(), + AmqpProperty.LastPartitionPropertiesRetrievalTimeUtc.ToString() + }; + + /// The set of key names for annotations known to be long-based system properties. + private static readonly HashSet SystemPropertyLongKeys = new() + { + AmqpProperty.SequenceNumber.ToString(), + AmqpProperty.Offset.ToString(), + AmqpProperty.PartitionLastEnqueuedSequenceNumber.ToString(), + AmqpProperty.PartitionLastEnqueuedOffset.ToString() + }; + /// /// Converts a given source into its corresponding /// AMQP representation. @@ -315,26 +333,200 @@ private static AmqpMessage BuildAmqpBatchFromMessages(IEnumerable s private static AmqpMessage BuildAmqpMessageFromEvent(EventData source, string partitionKey) { - if (!MemoryMarshal.TryGetArray(source.EventBody.ToMemory(), out var bodySegment)) + var sourceMessage = source.GetRawAmqpMessage(); + + var message = sourceMessage switch { - bodySegment = new ArraySegment(source.EventBody.ToArray()); + _ when sourceMessage.Body.TryGetData(out var dataBody) => AmqpMessage.Create(TranslateDataBody(dataBody)), + _ when sourceMessage.Body.TryGetSequence(out var sequenceBody) => AmqpMessage.Create(TranslateSequenceBody(sequenceBody)), + _ when sourceMessage.Body.TryGetValue(out var valueBody) => AmqpMessage.Create(TranslateValueBody(valueBody)), + _ => AmqpMessage.Create(new Data { Value = new ArraySegment(Array.Empty()) }) + }; + + // Header + + if (sourceMessage.HasSection(AmqpMessageSection.Header)) + { + if (sourceMessage.Header.DeliveryCount.HasValue) + { + message.Header.DeliveryCount = sourceMessage.Header.DeliveryCount; + } + + if (sourceMessage.Header.Durable.HasValue) + { + message.Header.Durable = sourceMessage.Header.Durable; + } + + if (sourceMessage.Header.Priority.HasValue) + { + message.Header.Priority = sourceMessage.Header.Priority; + } + + if (sourceMessage.Header.TimeToLive.HasValue) + { + message.Header.Ttl = (uint?)sourceMessage.Header.TimeToLive.Value.TotalMilliseconds; + } + + if (sourceMessage.Header.FirstAcquirer.HasValue) + { + message.Header.FirstAcquirer = sourceMessage.Header.FirstAcquirer; + } + + if (sourceMessage.Header.DeliveryCount.HasValue) + { + message.Header.DeliveryCount = sourceMessage.Header.DeliveryCount; + } } - var message = AmqpMessage.Create(new Data { Value = bodySegment }); + // Properties - if ((source.HasProperties) && (source.Properties.Count > 0)) + if (sourceMessage.HasSection(AmqpMessageSection.Properties)) + { + if (sourceMessage.Properties.AbsoluteExpiryTime.HasValue) + { + message.Properties.AbsoluteExpiryTime = sourceMessage.Properties.AbsoluteExpiryTime.Value.UtcDateTime; + } + + if (!string.IsNullOrEmpty(sourceMessage.Properties.ContentEncoding)) + { + message.Properties.ContentEncoding = sourceMessage.Properties.ContentEncoding; + } + + if (!string.IsNullOrEmpty(sourceMessage.Properties.ContentType)) + { + message.Properties.ContentType = sourceMessage.Properties.ContentType; + } + + if (sourceMessage.Properties.CorrelationId.HasValue) + { + message.Properties.CorrelationId = sourceMessage.Properties.CorrelationId.Value.ToString(); + } + + if (sourceMessage.Properties.CreationTime.HasValue) + { + message.Properties.CreationTime = sourceMessage.Properties.CreationTime.Value.UtcDateTime; + } + + if (!string.IsNullOrEmpty(sourceMessage.Properties.GroupId)) + { + message.Properties.GroupId = sourceMessage.Properties.GroupId; + } + + if (sourceMessage.Properties.GroupSequence.HasValue) + { + message.Properties.GroupSequence = sourceMessage.Properties.GroupSequence; + } + + if (sourceMessage.Properties.MessageId.HasValue) + { + message.Properties.MessageId = sourceMessage.Properties.MessageId.Value.ToString(); + } + + if (sourceMessage.Properties.ReplyTo.HasValue) + { + message.Properties.ReplyTo = sourceMessage.Properties.ReplyTo.Value.ToString(); + } + + if (!string.IsNullOrEmpty(sourceMessage.Properties.ReplyToGroupId)) + { + message.Properties.ReplyToGroupId = sourceMessage.Properties.ReplyToGroupId; + } + + if (!string.IsNullOrEmpty(sourceMessage.Properties.Subject)) + { + message.Properties.Subject = sourceMessage.Properties.Subject; + } + + if (sourceMessage.Properties.To.HasValue) + { + message.Properties.To = sourceMessage.Properties.To.Value.ToString(); + } + + if (sourceMessage.Properties.UserId.HasValue) + { + if (MemoryMarshal.TryGetArray(sourceMessage.Properties.UserId.Value, out var segment)) + { + message.Properties.UserId = segment; + } + else + { + message.Properties.UserId = new ArraySegment(sourceMessage.Properties.UserId.Value.ToArray()); + } + } + } + + // Application Properties + + if ((sourceMessage.HasSection(AmqpMessageSection.ApplicationProperties)) && (sourceMessage.ApplicationProperties.Count > 0)) { message.ApplicationProperties ??= new ApplicationProperties(); - foreach (KeyValuePair pair in source.Properties) + foreach (var pair in sourceMessage.ApplicationProperties) { if (TryCreateAmqpPropertyValueForEventProperty(pair.Value, out var amqpValue)) { message.ApplicationProperties.Map[pair.Key] = amqpValue; } + else + { + throw new NotSupportedException(string.Format(CultureInfo.CurrentCulture, Resources.InvalidAmqpMessageDictionaryTypeMask, nameof(sourceMessage.ApplicationProperties), pair.Key, pair.Value.GetType().Name)); + } + } + } + + // Message Annotations + + if (sourceMessage.HasSection(AmqpMessageSection.MessageAnnotations)) + { + foreach (var pair in sourceMessage.MessageAnnotations) + { + if (TryCreateAmqpPropertyValueForEventProperty(pair.Value, out var amqpValue)) + { + message.MessageAnnotations.Map[pair.Key] = amqpValue; + } + else + { + throw new NotSupportedException(string.Format(CultureInfo.CurrentCulture, Resources.InvalidAmqpMessageDictionaryTypeMask, nameof(sourceMessage.MessageAnnotations), pair.Key, pair.Value.GetType().Name)); + } + } + } + + // Delivery Annotations + + if (sourceMessage.HasSection(AmqpMessageSection.DeliveryAnnotations)) + { + foreach (var pair in sourceMessage.DeliveryAnnotations) + { + if (TryCreateAmqpPropertyValueForEventProperty(pair.Value, out var amqpValue)) + { + message.DeliveryAnnotations.Map[pair.Key] = amqpValue; + } + else + { + throw new NotSupportedException(string.Format(CultureInfo.CurrentCulture, Resources.InvalidAmqpMessageDictionaryTypeMask, nameof(sourceMessage.DeliveryAnnotations), pair.Key, pair.Value.GetType().Name)); + } + } + } + + // Footer + + if (sourceMessage.HasSection(AmqpMessageSection.Footer)) + { + foreach (var pair in sourceMessage.Footer) + { + if (TryCreateAmqpPropertyValueForEventProperty(pair.Value, out var amqpValue)) + { + message.Footer.Map[pair.Key] = amqpValue; + } + else + { + throw new NotSupportedException(string.Format(CultureInfo.CurrentCulture, Resources.InvalidAmqpMessageDictionaryTypeMask, nameof(sourceMessage.Footer), pair.Key, pair.Value.GetType().Name)); + } } } + // Special cases + if (!string.IsNullOrEmpty(partitionKey)) { message.MessageAnnotations.Map[AmqpProperty.PartitionKey] = partitionKey; @@ -368,190 +560,337 @@ private static AmqpMessage BuildAmqpMessageFromEvent(EventData source, /// private static EventData BuildEventFromAmqpMessage(AmqpMessage source) { - var body = (source.BodyType.HasFlag(SectionFlag.Data)) - ? ReadAmqpDataBody(source.DataBody) - : new BinaryData(ReadOnlyMemory.Empty); + var message = source switch + { + _ when TryGetDataBody(source, out var dataBody) => new AmqpAnnotatedMessage(dataBody), + _ when TryGetSequenceBody(source, out var sequenceBody) => new AmqpAnnotatedMessage(sequenceBody), + _ when TryGetValueBody(source, out var valueBody) => new AmqpAnnotatedMessage(valueBody), + _ => new AmqpAnnotatedMessage(AmqpMessageBody.FromData(MessageBody.FromReadOnlyMemorySegment(ReadOnlyMemory.Empty))) + }; - ParsedAnnotations systemAnnotations = ParseSystemAnnotations(source); + // Header - // If there were application properties associated with the message, translate them - // to the event. + if ((source.Sections & SectionFlag.Header) > 0) + { + if (source.Header.DeliveryCount.HasValue) + { + message.Header.DeliveryCount = source.Header.DeliveryCount; + } - var properties = default(Dictionary); + if (source.Header.Durable.HasValue) + { + message.Header.Durable = source.Header.Durable; + } - if (source.Sections.HasFlag(SectionFlag.ApplicationProperties)) - { - properties = new Dictionary(); + if (source.Header.Priority.HasValue) + { + message.Header.Priority = source.Header.Priority; + } - foreach (KeyValuePair pair in source.ApplicationProperties.Map) + if (source.Header.FirstAcquirer.HasValue) { - if (TryCreateEventPropertyForAmqpProperty(pair.Value, out object propertyValue)) - { - properties[pair.Key.ToString()] = propertyValue; - } + message.Header.FirstAcquirer = source.Header.FirstAcquirer; + } + + if (source.Header.DeliveryCount.HasValue) + { + message.Header.DeliveryCount = source.Header.DeliveryCount; + } + + if (source.Header.Ttl.HasValue) + { + message.Header.TimeToLive = TimeSpan.FromMilliseconds(source.Header.Ttl.Value); } } - return new EventData( - eventBody: body, - properties: properties, - systemProperties: systemAnnotations.ServiceAnnotations, - sequenceNumber: systemAnnotations.SequenceNumber ?? long.MinValue, - offset: systemAnnotations.Offset ?? long.MinValue, - enqueuedTime: systemAnnotations.EnqueuedTime ?? default, - partitionKey: systemAnnotations.PartitionKey, - lastPartitionSequenceNumber: systemAnnotations.LastSequenceNumber, - lastPartitionOffset: systemAnnotations.LastOffset, - lastPartitionEnqueuedTime: systemAnnotations.LastEnqueuedTime, - lastPartitionPropertiesRetrievalTime: systemAnnotations.LastReceivedTime); - } + // Properties - /// - /// Parses the annotations set by the Event Hubs service on the - /// associated with an event, extracting them into a consumable form. - /// - /// - /// The message to use as the source of the event. - /// - /// The parsed from the source message. - /// - private static ParsedAnnotations ParseSystemAnnotations(AmqpMessage source) - { - var systemProperties = new ParsedAnnotations(); + if ((source.Sections & SectionFlag.Properties) > 0) + { + if (source.Properties.AbsoluteExpiryTime.HasValue) + { + message.Properties.AbsoluteExpiryTime = source.Properties.AbsoluteExpiryTime; + } - object amqpValue; - object propertyValue; + if (!string.IsNullOrEmpty(source.Properties.ContentEncoding.Value)) + { + message.Properties.ContentEncoding = source.Properties.ContentEncoding.Value; + } - // Process the message annotations. + if (!string.IsNullOrEmpty(source.Properties.ContentType.Value)) + { + message.Properties.ContentType = source.Properties.ContentType.Value; + } - if (source.Sections.HasFlag(SectionFlag.MessageAnnotations)) - { - systemProperties.ServiceAnnotations ??= new Dictionary(); + if (source.Properties.CorrelationId != null) + { + message.Properties.CorrelationId = new AmqpMessageId(source.Properties.CorrelationId.ToString()); + } - var annotations = source.MessageAnnotations.Map; - var processed = new HashSet(); + if (source.Properties.CreationTime.HasValue) + { + message.Properties.CreationTime = source.Properties.CreationTime; + } - if ((annotations.TryGetValue(AmqpProperty.EnqueuedTime, out amqpValue)) - && (TryCreateEventPropertyForAmqpProperty(amqpValue, out propertyValue))) + if (!string.IsNullOrEmpty(source.Properties.GroupId)) { - systemProperties.EnqueuedTime = propertyValue switch - { - DateTime dateValue => new DateTimeOffset(dateValue, TimeSpan.Zero), - long longValue => new DateTimeOffset(longValue, TimeSpan.Zero), - _ => (DateTimeOffset)propertyValue - }; + message.Properties.GroupId = source.Properties.GroupId; + } - processed.Add(AmqpProperty.EnqueuedTime.ToString()); + if (source.Properties.GroupSequence.HasValue) + { + message.Properties.GroupSequence = source.Properties.GroupSequence; } - if ((annotations.TryGetValue(AmqpProperty.SequenceNumber, out amqpValue)) - && (TryCreateEventPropertyForAmqpProperty(amqpValue, out propertyValue))) + if (source.Properties.MessageId != null) { - systemProperties.SequenceNumber = (long)propertyValue; - processed.Add(AmqpProperty.SequenceNumber.ToString()); + message.Properties.MessageId = new AmqpMessageId(source.Properties.MessageId.ToString()); } - if ((annotations.TryGetValue(AmqpProperty.Offset, out amqpValue)) - && (TryCreateEventPropertyForAmqpProperty(amqpValue, out propertyValue)) - && (long.TryParse((string)propertyValue, NumberStyles.Integer, CultureInfo.InvariantCulture, out var offset))) + if (source.Properties.ReplyTo != null) { - systemProperties.Offset = offset; - processed.Add(AmqpProperty.Offset.ToString()); + message.Properties.ReplyTo = new AmqpAddress(source.Properties.ReplyTo.ToString()); } - if ((annotations.TryGetValue(AmqpProperty.PartitionKey, out amqpValue)) - && (TryCreateEventPropertyForAmqpProperty(amqpValue, out propertyValue))) + if (!string.IsNullOrEmpty(source.Properties.ReplyToGroupId)) { - systemProperties.PartitionKey = (string)propertyValue; - processed.Add(AmqpProperty.PartitionKey.ToString()); + message.Properties.ReplyToGroupId = source.Properties.ReplyToGroupId; } - string key; + if (!string.IsNullOrEmpty(source.Properties.Subject)) + { + message.Properties.Subject = source.Properties.Subject; + } - foreach (KeyValuePair pair in annotations) + if (source.Properties.To != null) { - key = pair.Key.ToString(); + message.Properties.To = new AmqpAddress(source.Properties.To.ToString()); + } - if ((!processed.Contains(key)) - && (TryCreateEventPropertyForAmqpProperty(pair.Value, out propertyValue))) - { - systemProperties.ServiceAnnotations.Add(key, propertyValue); - processed.Add(key); - } + if (source.Properties.UserId != null) + { + message.Properties.UserId = source.Properties.UserId; } } - // Process the delivery annotations. + // Application Properties - if (source.Sections.HasFlag(SectionFlag.DeliveryAnnotations)) + if ((source.Sections & SectionFlag.ApplicationProperties) > 0) { - if ((source.DeliveryAnnotations.Map.TryGetValue(AmqpProperty.PartitionLastEnqueuedTimeUtc, out amqpValue)) - && (TryCreateEventPropertyForAmqpProperty(amqpValue, out propertyValue))) + foreach (var pair in source.ApplicationProperties.Map) { - systemProperties.LastEnqueuedTime = propertyValue switch + if (TryCreateEventPropertyForAmqpProperty(pair.Value, out var eventValue)) { - DateTime dateValue => new DateTimeOffset(dateValue, TimeSpan.Zero), - long longValue => new DateTimeOffset(longValue, TimeSpan.Zero), - _ => (DateTimeOffset)propertyValue - }; + message.ApplicationProperties[pair.Key.ToString()] = eventValue; + } } + } - if ((source.DeliveryAnnotations.Map.TryGetValue(AmqpProperty.PartitionLastEnqueuedSequenceNumber, out amqpValue)) - && (TryCreateEventPropertyForAmqpProperty(amqpValue, out propertyValue))) - { - systemProperties.LastSequenceNumber = (long)propertyValue; - } + // Message Annotations - if ((source.DeliveryAnnotations.Map.TryGetValue(AmqpProperty.PartitionLastEnqueuedOffset, out amqpValue)) - && (TryCreateEventPropertyForAmqpProperty(amqpValue, out propertyValue)) - && (long.TryParse((string)propertyValue, NumberStyles.Integer, CultureInfo.InvariantCulture, out var offset))) + if ((source.Sections & SectionFlag.MessageAnnotations) > 0) + { + foreach (var pair in source.MessageAnnotations.Map) { - systemProperties.LastOffset = offset; + if (TryCreateEventPropertyForAmqpProperty(pair.Value, out var eventValue)) + { + if (SystemPropertyDateTimeKeys.Contains(pair.Key.ToString())) + { + eventValue = eventValue switch + { + DateTime dateValue => new DateTimeOffset(dateValue, TimeSpan.Zero), + long longValue => new DateTimeOffset(longValue, TimeSpan.Zero), + _ => eventValue + }; + } + else if (SystemPropertyLongKeys.Contains(pair.Key.ToString())) + { + eventValue = eventValue switch + { + string stringValue when long.TryParse(stringValue, NumberStyles.Integer, CultureInfo.InvariantCulture, out var longValue) => longValue, + _ => eventValue + }; + } + + message.MessageAnnotations[pair.Key.ToString()] = eventValue; + } } + } - if ((source.DeliveryAnnotations.Map.TryGetValue(AmqpProperty.LastPartitionPropertiesRetrievalTimeUtc, out amqpValue)) - && (TryCreateEventPropertyForAmqpProperty(amqpValue, out propertyValue))) + // Delivery Annotations + + if ((source.Sections & SectionFlag.DeliveryAnnotations) > 0) + { + foreach (var pair in source.DeliveryAnnotations.Map) { - systemProperties.LastReceivedTime = propertyValue switch + if (TryCreateEventPropertyForAmqpProperty(pair.Value, out var eventValue)) { - DateTime dateValue => new DateTimeOffset(dateValue, TimeSpan.Zero), - long longValue => new DateTimeOffset(longValue, TimeSpan.Zero), - _ => (DateTimeOffset)propertyValue - }; + if (SystemPropertyDateTimeKeys.Contains(pair.Key.ToString())) + { + eventValue = eventValue switch + { + DateTime dateValue => new DateTimeOffset(dateValue, TimeSpan.Zero), + long longValue => new DateTimeOffset(longValue, TimeSpan.Zero), + _ => eventValue + }; + } + else if (SystemPropertyLongKeys.Contains(pair.Key.ToString())) + { + eventValue = eventValue switch + { + string stringValue when long.TryParse(stringValue, NumberStyles.Integer, CultureInfo.InvariantCulture, out var longValue) => longValue, + _ => eventValue + }; + } + + message.DeliveryAnnotations[pair.Key.ToString()] = eventValue; + } } } - // Process the properties annotations + // Footer - if (source.Sections.HasFlag(SectionFlag.Properties)) + if ((source.Sections & SectionFlag.Footer) > 0) { - var properties = source.Properties; - - void conditionalAdd(string name, object value, bool condition) + foreach (var pair in source.Footer.Map) { - if (condition) + if (TryCreateEventPropertyForAmqpProperty(pair.Value, out var eventValue)) { - systemProperties.ServiceAnnotations ??= new Dictionary(); - systemProperties.ServiceAnnotations.Add(name, value); + message.Footer[pair.Key.ToString()] = eventValue; } } + } + + return new EventData(message); + } + + /// + /// Translates the data body segments into the corresponding set of + /// instances. + /// + /// + /// The data body to translate. + /// + /// The set of instances that represents the . + /// + private static IEnumerable TranslateDataBody(IEnumerable> dataBody) + { + foreach (var bodySegment in dataBody) + { + if (!MemoryMarshal.TryGetArray(bodySegment, out ArraySegment dataSegment)) + { + dataSegment = new ArraySegment(bodySegment.ToArray()); + } + + yield return new Data + { + Value = dataSegment + }; + } + } + + /// + /// Translates the data body elements into the corresponding set of + /// instances. + /// + /// + /// The sequence body to translate. + /// + /// The set of instances that represents the in AMQP format. + /// + private static IEnumerable TranslateSequenceBody(IEnumerable> sequenceBody) + { + foreach (var item in sequenceBody) + { + yield return new AmqpSequence((System.Collections.IList)item); + } + } + + /// + /// Translates the data body into the corresponding set of + /// instance. + /// + /// + /// The sequence body to translate. + /// + /// The instance that represents the in AMQP format. + /// + private static AmqpValue TranslateValueBody(object valueBody) + { + if (TryCreateAmqpPropertyValueForEventProperty(valueBody, out var amqpValue, allowBodyTypes: true)) + { + return new AmqpValue { Value = amqpValue }; + } + + throw new NotSupportedException(string.Format(CultureInfo.CurrentCulture, Resources.InvalidAmqpMessageValueBodyMask, valueBody.GetType().Name)); + } + + /// + /// Attempts to read the data body of an . + /// + /// + /// The to read from. + /// The value of the data body, if read. + /// + /// true if the body was successfully read; otherwise, false. + /// + private static bool TryGetDataBody(AmqpMessage source, out AmqpMessageBody dataBody) + { + if (((source.BodyType & SectionFlag.Data) == 0) || (source.DataBody == null)) + { + dataBody = null; + return false; + } + + dataBody = AmqpMessageBody.FromData(MessageBody.FromDataSegments(source.DataBody)); + return true; + } + + /// + /// Attempts to read the sequence body of an . + /// + /// + /// The to read from. + /// The value of the sequence body, if read. + /// + /// true if the body was successfully read; otherwise, false. + /// + private static bool TryGetSequenceBody(AmqpMessage source, out AmqpMessageBody sequenceBody) + { + if ((source.BodyType & SectionFlag.AmqpSequence) == 0) + { + sequenceBody = null; + return false; + } + + sequenceBody = AmqpMessageBody.FromSequence(source.SequenceBody.Select(item => (IList)item.List).ToArray()); + return true; + } + + /// + /// Attempts to read the sequence body of an . + /// + /// + /// The to read from. + /// The value body, if read. + /// + /// true if the body was successfully read; otherwise, false. + /// + private static bool TryGetValueBody(AmqpMessage source, out AmqpMessageBody valueBody) + { + if (((source.BodyType & SectionFlag.AmqpValue) == 0) || (source.ValueBody?.Value == null)) + { + valueBody = null; + return false; + } - conditionalAdd(Properties.MessageIdName, properties.MessageId, properties.MessageId != null); - conditionalAdd(Properties.UserIdName, properties.UserId, properties.UserId.Array != null); - conditionalAdd(Properties.ToName, properties.To, properties.To != null); - conditionalAdd(Properties.SubjectName, properties.Subject, properties.Subject != null); - conditionalAdd(Properties.ReplyToName, properties.ReplyTo, properties.ReplyTo != null); - conditionalAdd(Properties.CorrelationIdName, properties.CorrelationId, properties.CorrelationId != null); - conditionalAdd(Properties.ContentTypeName, properties.ContentType, properties.ContentType.Value != null); - conditionalAdd(Properties.ContentEncodingName, properties.ContentEncoding, properties.ContentEncoding.Value != null); - conditionalAdd(Properties.AbsoluteExpiryTimeName, properties.AbsoluteExpiryTime, properties.AbsoluteExpiryTime != null); - conditionalAdd(Properties.CreationTimeName, properties.CreationTime, properties.CreationTime != null); - conditionalAdd(Properties.GroupIdName, properties.GroupId, properties.GroupId != null); - conditionalAdd(Properties.GroupSequenceName, properties.GroupSequence, properties.GroupSequence != null); - conditionalAdd(Properties.ReplyToGroupIdName, properties.ReplyToGroupId, properties.ReplyToGroupId != null); + if (TryCreateEventPropertyForAmqpProperty(source.ValueBody.Value, out var translatedValue, allowBodyTypes: true)) + { + valueBody = AmqpMessageBody.FromValue(translatedValue); + return true; } - return systemProperties; + throw new NotSupportedException(string.Format(CultureInfo.CurrentCulture, Resources.InvalidAmqpMessageValueBodyMask, source.ValueBody.Value.GetType().Name)); } /// @@ -560,11 +899,13 @@ void conditionalAdd(string name, object value, bool condition) /// /// The value of the event property to create an AMQP property value for. /// The AMQP property value that was created. + /// true to allow an AMQP map to be translated to additional types supported only by a message body; otherwise, false. /// /// true if an AMQP property value was able to be created; otherwise, false. /// private static bool TryCreateAmqpPropertyValueForEventProperty(object eventPropertyValue, - out object amqpPropertyValue) + out object amqpPropertyValue, + bool allowBodyTypes = false) { amqpPropertyValue = null; @@ -611,6 +952,18 @@ private static bool TryCreateAmqpPropertyValueForEventProperty(object eventPrope amqpPropertyValue = new DescribedType(AmqpProperty.Descriptor.TimeSpan, ((TimeSpan)eventPropertyValue).Ticks); break; + case AmqpProperty.Type.Unknown when allowBodyTypes && eventPropertyValue is byte[] byteArray: + amqpPropertyValue = new ArraySegment(byteArray); + break; + + case AmqpProperty.Type.Unknown when allowBodyTypes && eventPropertyValue is System.Collections.IDictionary dict: + amqpPropertyValue = new AmqpMap(dict); + break; + + case AmqpProperty.Type.Unknown when allowBodyTypes && eventPropertyValue is System.Collections.IList: + amqpPropertyValue = eventPropertyValue; + break; + case AmqpProperty.Type.Unknown: var exception = new SerializationException(string.Format(CultureInfo.CurrentCulture, Resources.FailedToSerializeUnsupportedType, eventPropertyValue.GetType().FullName)); EventHubsEventSource.Log.UnexpectedException(exception.Message); @@ -626,11 +979,13 @@ private static bool TryCreateAmqpPropertyValueForEventProperty(object eventPrope /// /// The value of the AMQP property to create an event property value for. /// The event property value that was created. + /// true to allow an AMQP map to be translated to additional types supported only by a message body; otherwise, false. /// /// true if an event property value was able to be created; otherwise, false. /// private static bool TryCreateEventPropertyForAmqpProperty(object amqpPropertyValue, - out object eventPropertyValue) + out object eventPropertyValue, + bool allowBodyTypes = false) { eventPropertyValue = null; @@ -697,6 +1052,19 @@ private static bool TryCreateEventPropertyForAmqpProperty(object amqpPropertyVal eventPropertyValue = TranslateSymbol((AmqpSymbol)described.Descriptor, described.Value); break; + case AmqpMap map when allowBodyTypes: + { + var dict = new Dictionary(); + + foreach (var pair in map) + { + dict.Add(pair.Key.ToString(), pair.Value); + } + + eventPropertyValue = dict; + break; + }; + default: var exception = new SerializationException(string.Format(CultureInfo.CurrentCulture, Resources.FailedToSerializeUnsupportedType, amqpPropertyValue.GetType().FullName)); EventHubsEventSource.Log.UnexpectedException(exception.Message); @@ -779,84 +1147,5 @@ private static ArraySegment ReadStreamToArraySegment(Stream stream) } } } - - /// - /// Reads the data body of an AMQP message, transforming it for use - /// as the body of an instance. - /// - /// - /// The body data set of an AMQP message. - /// - /// A representation of the . - /// - private static BinaryData ReadAmqpDataBody(IEnumerable body) - { - var writer = new ArrayBufferWriter(); - - foreach (var data in body) - { - var dataBytes = GetDataBytes(data); - dataBytes.CopyTo(writer.GetMemory(dataBytes.Length)); - - writer.Advance(dataBytes.Length); - } - - return (writer.WrittenCount > 0) - ? BinaryData.FromBytes(writer.WrittenMemory) - : new BinaryData(Array.Empty()); - } - - /// - /// Gets the bytes that comprise an AMQP data instance. - /// - /// - /// The data to read the bytes from. - /// - /// The set of bytes extracted from the . - /// - private static ReadOnlyMemory GetDataBytes(Data data) - { - return data.Value switch - { - byte[] byteArray => byteArray, - ArraySegment segment => segment, - _ => ReadOnlyMemory.Empty - }; - } - - /// - /// The set of system annotations set on a message received from the - /// Event Hubs service. - /// - /// - private struct ParsedAnnotations - { - /// The set of weakly typed annotations associated with the message. - public Dictionary ServiceAnnotations; - - /// The sequence number of the event associated with the message. - public long? SequenceNumber; - - /// The offset of the event associated with the message. - public long? Offset; - - /// The date and time, in UTC, that the event associated with the message was enqueued. - public DateTimeOffset? EnqueuedTime; - - /// The partition key that the event associated with the message was published with. - public string PartitionKey; - - /// The sequence number of the event that was last enqueued in the partition. - public long? LastSequenceNumber; - - /// The offset of the event that was last enqueued in the partition. - public long? LastOffset; - - /// The date and time, in UTC, that an event was last enqueued in the partition. - public DateTimeOffset? LastEnqueuedTime; - - /// The date and time, in UTC, that the last enqueued event information was retrieved from the service. - public DateTimeOffset? LastReceivedTime; - } } } diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs b/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs index aa79945e3c0c6..151f67b0ea0a6 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs +++ b/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs @@ -294,15 +294,6 @@ public string CorrelationId [EditorBrowsable(EditorBrowsableState.Never)] public Stream BodyAsStream => EventBody.ToStream(); - /// - /// Indicates whether this instance has a populated set of - /// or not, to avoid triggering lazy allocation by checking the property itself. - /// - /// - /// true if this instance has properties; otherwise, false. - /// - internal bool HasProperties => _amqpMessage.HasSection(AmqpMessageSection.ApplicationProperties); - /// /// The sequence number of the event that was last enqueued into the Event Hub partition from which this /// event was received. diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/tests/Amqp/AmqpMessageConverterTests.cs b/sdk/eventhub/Azure.Messaging.EventHubs/tests/Amqp/AmqpMessageConverterTests.cs index 7df623f0a16a0..28f26ad63d9a3 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs/tests/Amqp/AmqpMessageConverterTests.cs +++ b/sdk/eventhub/Azure.Messaging.EventHubs/tests/Amqp/AmqpMessageConverterTests.cs @@ -384,6 +384,150 @@ public void CreateMessageFromEventDoesNotTriggerPropertiesInstantation() Assert.That(eventData.GetRawAmqpMessage().HasSection(AmqpMessageSection.ApplicationProperties), Is.False, "Translation should not have cause the properties dictionary to be instantiated."); } + /// + /// Verifies functionality of the + /// method. + /// + /// + [Test] + public void CreateMessageFromEventPopulatesTheHeader() + { + var sourceMessage = new AmqpAnnotatedMessage(AmqpMessageBody.FromData(new[] { ReadOnlyMemory.Empty })); + sourceMessage.Header.DeliveryCount = 123; + sourceMessage.Header.Durable = true; + sourceMessage.Header.FirstAcquirer = true; + sourceMessage.Header.Priority = 1; + sourceMessage.Header.TimeToLive = TimeSpan.FromDays(2); + + var eventData = new EventData(sourceMessage); + using var message = new AmqpMessageConverter().CreateMessageFromEvent(eventData); + + Assert.That(message, Is.Not.Null, "The AMQP message should have been created."); + Assert.That(message.Sections.HasFlag(SectionFlag.Header), "The AMQP message should have a header section."); + Assert.That(message.Header.DeliveryCount, Is.EqualTo(sourceMessage.Header.DeliveryCount), "The delivery count should match."); + Assert.That(message.Header.Durable, Is.EqualTo(sourceMessage.Header.Durable), "The durable flag should match."); + Assert.That(message.Header.FirstAcquirer, Is.EqualTo(sourceMessage.Header.FirstAcquirer), "The first acquirer flag should match."); + Assert.That(message.Header.Priority, Is.EqualTo(sourceMessage.Header.Priority), "The priority should match."); + Assert.That(message.Header.Ttl, Is.EqualTo(sourceMessage.Header.TimeToLive.Value.TotalMilliseconds), "The time to live should match."); + } + + /// + /// Verifies functionality of the + /// method. + /// + /// + [Test] + public void CreateMessageFromEventPopulatesTheProperties() + { + var sourceMessage = new AmqpAnnotatedMessage(AmqpMessageBody.FromData(new[] { ReadOnlyMemory.Empty })); + sourceMessage.Properties.AbsoluteExpiryTime = new DateTimeOffset(2015, 10, 27, 0, 0 ,0 ,0, TimeSpan.Zero); + sourceMessage.Properties.ContentEncoding = "utf-8"; + sourceMessage.Properties.ContentType = "test/unit"; + sourceMessage.Properties.CorrelationId = new AmqpMessageId("OU812"); + sourceMessage.Properties.CreationTime = new DateTimeOffset(2012, 3, 4, 8, 0, 0, 0, TimeSpan.Zero); + sourceMessage.Properties.GroupId = "Red Squad"; + sourceMessage.Properties.GroupSequence = 76; + sourceMessage.Properties.MessageId = new AmqpMessageId("Bob"); + sourceMessage.Properties.ReplyTo = new AmqpAddress("1407 Graymalkin Lane"); + sourceMessage.Properties.ReplyToGroupId = "Home"; + sourceMessage.Properties.Subject = "You'll never believe this weight loss secret!"; + sourceMessage.Properties.To = new AmqpAddress("http://some.server.com"); + sourceMessage.Properties.UserId = new byte[] { 0x11, 0x22 }; + + var eventData = new EventData(sourceMessage); + using var message = new AmqpMessageConverter().CreateMessageFromEvent(eventData); + + Assert.That(message, Is.Not.Null, "The AMQP message should have been created."); + Assert.That(message.Sections.HasFlag(SectionFlag.Properties), "The AMQP message should have a properties section."); + Assert.That(message.Properties.AbsoluteExpiryTime, Is.EqualTo(sourceMessage.Properties.AbsoluteExpiryTime.Value.UtcDateTime), "The expiry time should match."); + Assert.That(message.Properties.ContentEncoding.ToString(), Is.EqualTo(sourceMessage.Properties.ContentEncoding), "The content encoding should match."); + Assert.That(message.Properties.ContentType.ToString(), Is.EqualTo(sourceMessage.Properties.ContentType), "The content type should match."); + Assert.That(message.Properties.CorrelationId.ToString(), Is.EqualTo(sourceMessage.Properties.CorrelationId.ToString()), "The correlation identifier should match."); + Assert.That(message.Properties.CreationTime, Is.EqualTo(sourceMessage.Properties.CreationTime.Value.UtcDateTime), "The creation time should match."); + Assert.That(message.Properties.GroupId, Is.EqualTo(sourceMessage.Properties.GroupId), "The group identifier should match."); + Assert.That(message.Properties.GroupSequence, Is.EqualTo(sourceMessage.Properties.GroupSequence), "The group sequence should match."); + Assert.That(message.Properties.MessageId.ToString(), Is.EqualTo(sourceMessage.Properties.MessageId.ToString()), "The message identifier should match."); + Assert.That(message.Properties.ReplyTo.ToString(), Is.EqualTo(sourceMessage.Properties.ReplyTo.ToString()), "The reply-to address should match."); + Assert.That(message.Properties.ReplyToGroupId, Is.EqualTo(sourceMessage.Properties.ReplyToGroupId), "The reply-to group identifier should match."); + Assert.That(message.Properties.Subject, Is.EqualTo(sourceMessage.Properties.Subject), "The subject should match."); + Assert.That(message.Properties.To.ToString(), Is.EqualTo(sourceMessage.Properties.To.ToString()), "The to address should match."); + Assert.That(message.Properties.UserId.ToArray(), Is.EquivalentTo(sourceMessage.Properties.UserId.Value.ToArray()), "The user identifier should match."); + } + + /// + /// Verifies functionality of the + /// method. + /// + /// + [Test] + public void CreateMessageFromEventPopulatesMapSections() + { + var sourceMessage = new AmqpAnnotatedMessage(AmqpMessageBody.FromData(new[] { ReadOnlyMemory.Empty })); + + // Delivery Annotations + + sourceMessage.DeliveryAnnotations.Add("Three", 3); + sourceMessage.DeliveryAnnotations.Add("Four", "4"); + + // Message Annotations + + sourceMessage.MessageAnnotations.Add("Five", 5); + sourceMessage.MessageAnnotations.Add("Six", "6"); + + // Footer + + sourceMessage.Footer.Add("Seven", 7); + sourceMessage.Footer.Add("Eight", "8"); + + var eventData = new EventData(sourceMessage); + using var message = new AmqpMessageConverter().CreateMessageFromEvent(eventData); + + Assert.That(message, Is.Not.Null, "The AMQP message should have been created."); + Assert.That(message.Sections.HasFlag(SectionFlag.DeliveryAnnotations), "The AMQP message should have a delivery annotations section."); + Assert.That(message.Sections.HasFlag(SectionFlag.MessageAnnotations), "The AMQP message should have a message annotations section."); + Assert.That(message.Sections.HasFlag(SectionFlag.Footer), "The AMQP message should have a footer section."); + + void validateMap(AmqpMap map, IDictionary expected, string mapName) + { + foreach (var item in map) + { + Assert.That(expected.TryGetValue(item.Key.ToString(), out object expectedValue), Is.True, $"The { mapName } section map did not contain: [{ item.Key }]"); + Assert.That(item.Value, Is.EqualTo(expectedValue), $"The { mapName } section map property value did not match for: [{ item.Key }]"); + } + } + + validateMap(message.DeliveryAnnotations.Map, sourceMessage.DeliveryAnnotations, nameof(sourceMessage.DeliveryAnnotations)); + validateMap(message.MessageAnnotations.Map, sourceMessage.MessageAnnotations, nameof(sourceMessage.MessageAnnotations)); + validateMap(message.Footer.Map, sourceMessage.Footer, nameof(sourceMessage.Footer)); + } + + /// + /// Verifies functionality of the + /// method. + /// + /// + [Test] + public void CreateMessageFromEventPopulatesApplicationProperties() + { + var sourceMessage = new AmqpAnnotatedMessage(AmqpMessageBody.FromData(new[] { ReadOnlyMemory.Empty })); + sourceMessage.ApplicationProperties.Add("Three", 3); + sourceMessage.ApplicationProperties.Add("Four", "4"); + + var eventData = new EventData(sourceMessage); + using var message = new AmqpMessageConverter().CreateMessageFromEvent(eventData); + + Assert.That(message, Is.Not.Null, "The AMQP message should have been created."); + Assert.That(message.Sections.HasFlag(SectionFlag.ApplicationProperties), "The AMQP message should have an application properties section."); + + foreach (var property in sourceMessage.ApplicationProperties.Keys) + { + var containsValue = message.ApplicationProperties.Map.TryGetValue(property, out object value); + + Assert.That(containsValue, Is.True, $"The application properties did not contain: [{ property }]"); + Assert.That(value, Is.EqualTo(eventData.Properties[property]), $"The application property value did not match for: [{ property }]"); + } + } + /// /// Verifies functionality of the /// method. @@ -1245,7 +1389,7 @@ public void CreateEventFromMessagePopulatesLastRetrievalTimeFromTicks() [Test] public void CreateEventFromMessageAllowsAnEmptyMessage() { - var message = AmqpMessage.Create(); + using var message = AmqpMessage.Create(); Assert.That(() => new AmqpMessageConverter().CreateEventFromMessage(message), Throws.Nothing); } @@ -1254,16 +1398,134 @@ public void CreateEventFromMessageAllowsAnEmptyMessage() /// method. /// /// + [Test] + public void CreateEventFromMessagePopulatesTheHeader() + { + var body = new byte[] { 0x11, 0x22, 0x33 }; + using var sourceMessage = AmqpMessage.Create(new Data { Value = body }); + sourceMessage.Header.DeliveryCount = 123; + sourceMessage.Header.Durable = true; + sourceMessage.Header.FirstAcquirer = true; + sourceMessage.Header.Priority = 1; + sourceMessage.Header.Ttl = (uint)TimeSpan.FromDays(2).TotalMilliseconds; + + var converter = new AmqpMessageConverter(); + var eventData = converter.CreateEventFromMessage(sourceMessage); + var message = eventData.GetRawAmqpMessage(); + + Assert.That(eventData, Is.Not.Null, "The event should have been created."); + Assert.That(message.HasSection(AmqpMessageSection.Header), "The message should have a header section."); + Assert.That(message.Header.DeliveryCount, Is.EqualTo(sourceMessage.Header.DeliveryCount), "The delivery count should match."); + Assert.That(message.Header.Durable, Is.EqualTo(sourceMessage.Header.Durable), "The durable flag should match."); + Assert.That(message.Header.FirstAcquirer, Is.EqualTo(sourceMessage.Header.FirstAcquirer), "The first acquirer flag should match."); + Assert.That(message.Header.Priority, Is.EqualTo(sourceMessage.Header.Priority), "The priority should match."); + Assert.That(message.Header.TimeToLive.Value.TotalMilliseconds, Is.EqualTo(sourceMessage.Header.Ttl), "The time to live should match."); + } + + /// + /// Verifies functionality of the + /// method. + /// + /// + [Test] + public void CreateEventFromMessagePopulatesTheProperties() + { + var body = new byte[] { 0x11, 0x22, 0x33 }; + using var sourceMessage = AmqpMessage.Create(new Data { Value = body }); + sourceMessage.Properties.AbsoluteExpiryTime = new DateTimeOffset(2015, 10, 27, 0, 0 ,0 ,0, TimeSpan.Zero).UtcDateTime; + sourceMessage.Properties.ContentEncoding = "utf-8"; + sourceMessage.Properties.ContentType = "test/unit"; + sourceMessage.Properties.CorrelationId = "OU812"; + sourceMessage.Properties.CreationTime = new DateTimeOffset(2012, 3, 4, 8, 0, 0, 0, TimeSpan.Zero).UtcDateTime; + sourceMessage.Properties.GroupId = "Red Squad"; + sourceMessage.Properties.GroupSequence = 76; + sourceMessage.Properties.MessageId = "Bob"; + sourceMessage.Properties.ReplyTo = "1407 Graymalkin Lane"; + sourceMessage.Properties.ReplyToGroupId = "Home"; + sourceMessage.Properties.Subject = "You'll never believe this weight loss secret!"; + sourceMessage.Properties.To = "http://some.server.com"; + sourceMessage.Properties.UserId = new ArraySegment(new byte[] { 0x11, 0x22 }); + + var converter = new AmqpMessageConverter(); + var eventData = converter.CreateEventFromMessage(sourceMessage); + var message = eventData.GetRawAmqpMessage(); + + Assert.That(eventData, Is.Not.Null, "The event should have been created."); + Assert.That(message.HasSection(AmqpMessageSection.Properties), "The message should have a properties section."); + Assert.That(message.Properties.AbsoluteExpiryTime.Value.UtcDateTime, Is.EqualTo(sourceMessage.Properties.AbsoluteExpiryTime), "The expiry time should match."); + Assert.That(message.Properties.ContentEncoding, Is.EqualTo(sourceMessage.Properties.ContentEncoding.ToString()), "The content encoding should match."); + Assert.That(message.Properties.ContentType, Is.EqualTo(sourceMessage.Properties.ContentType.ToString()), "The content type should match."); + Assert.That(message.Properties.CorrelationId.ToString(), Is.EqualTo(sourceMessage.Properties.CorrelationId.ToString()), "The correlation identifier should match."); + Assert.That(message.Properties.CreationTime.Value.UtcDateTime, Is.EqualTo(sourceMessage.Properties.CreationTime), "The creation time should match."); + Assert.That(message.Properties.GroupId, Is.EqualTo(sourceMessage.Properties.GroupId), "The group identifier should match."); + Assert.That(message.Properties.GroupSequence, Is.EqualTo(sourceMessage.Properties.GroupSequence), "The group sequence should match."); + Assert.That(message.Properties.MessageId.ToString(), Is.EqualTo(sourceMessage.Properties.MessageId.ToString()), "The message identifier should match."); + Assert.That(message.Properties.ReplyTo.ToString(), Is.EqualTo(sourceMessage.Properties.ReplyTo.ToString()), "The reply-to address should match."); + Assert.That(message.Properties.ReplyToGroupId, Is.EqualTo(sourceMessage.Properties.ReplyToGroupId), "The reply-to group identifier should match."); + Assert.That(message.Properties.Subject, Is.EqualTo(sourceMessage.Properties.Subject), "The subject should match."); + Assert.That(message.Properties.To.ToString(), Is.EqualTo(sourceMessage.Properties.To.ToString()), "The to address should match."); + Assert.That(message.Properties.UserId.Value.ToArray(), Is.EquivalentTo(sourceMessage.Properties.UserId.ToArray()), "The user identifier should match."); + } + + /// + /// Verifies functionality of the + /// method. + /// + /// + [Test] + public void CreateEventFromMessagePopulatesDictionaryProperties() + { + var body = new byte[] { 0x11, 0x22, 0x33 }; + using var sourceMessage = AmqpMessage.Create(new Data { Value = body }); + + // Delivery Annotations + + sourceMessage.DeliveryAnnotations.Map.Add("Three", 3); + sourceMessage.DeliveryAnnotations.Map.Add("Four", "4"); + + // Message Annotations + + sourceMessage.MessageAnnotations.Map.Add("Five", 5); + sourceMessage.MessageAnnotations.Map.Add("Six", "6"); + + // Footer + + sourceMessage.Footer.Map.Add("Seven", 7); + sourceMessage.Footer.Map.Add("Eight", "8"); + + var converter = new AmqpMessageConverter(); + var eventData = converter.CreateEventFromMessage(sourceMessage); + var message = eventData.GetRawAmqpMessage(); + + Assert.That(eventData, Is.Not.Null, "The event should have been created."); + Assert.That(message.HasSection(AmqpMessageSection.MessageAnnotations), "The message should have a message annotations section."); + Assert.That(message.HasSection(AmqpMessageSection.DeliveryAnnotations), "The message should have a delivery annotations section."); + Assert.That(message.HasSection(AmqpMessageSection.Footer), "The message should have a footer section."); + + void validateMap(AmqpMap expected, IDictionary dictionary, string mapName) + { + foreach (var item in expected) + { + Assert.That(dictionary.TryGetValue(item.Key.ToString(), out object expectedValue), Is.True, $"The { mapName } section map did not contain: [{ item.Key }]"); + Assert.That(item.Value, Is.EqualTo(expectedValue), $"The { mapName } section map property value did not match for: [{ item.Key }]"); + } + } + + validateMap(sourceMessage.DeliveryAnnotations.Map, message.DeliveryAnnotations, nameof(sourceMessage.DeliveryAnnotations)); + validateMap(sourceMessage.MessageAnnotations.Map, message.MessageAnnotations, nameof(sourceMessage.MessageAnnotations)); + validateMap(sourceMessage.Footer.Map, message.Footer, nameof(sourceMessage.Footer)); + } + [Test] public void CreateEventFromMessageAllowsAnEmptyMessageWithProperties() { var propertyValue = 1; - var message = AmqpMessage.Create(); + using var message = AmqpMessage.Create(); message.ApplicationProperties.Map.Add("Test", propertyValue); message.MessageAnnotations.Map.Add(AmqpProperty.Offset, propertyValue.ToString()); - EventData eventData = new AmqpMessageConverter().CreateEventFromMessage(message); + var eventData = new AmqpMessageConverter().CreateEventFromMessage(message); Assert.That(eventData, Is.Not.Null, "The event should have been created."); Assert.That(eventData.Properties.Count, Is.EqualTo(message.ApplicationProperties.Map.Count()), "There should have been properties present."); Assert.That(eventData.Properties.First().Value, Is.EqualTo(propertyValue), "The application property should have been populated."); @@ -1295,7 +1557,7 @@ public void CreateEventFromMessageDoesNotPopulatePropertiesByDefault() /// /// [Test] - public void AnEventCanBeTranslatedToItself() + public void ASimpleEventCanBeTranslatedToItself() { var sourceEvent = new EventData( eventBody: new BinaryData(new byte[] { 0x11, 0x22, 0x33 }), @@ -1310,6 +1572,148 @@ public void AnEventCanBeTranslatedToItself() Assert.That(eventData.IsEquivalentTo(sourceEvent), "The translated event should match the source event."); } + /// + /// Verifies functionality of the + /// method. + /// + /// + [Test] + public void AComplexEventCanBeTranslatedToItself() + { + var sourceMessage = new AmqpAnnotatedMessage(AmqpMessageBody.FromData(new ReadOnlyMemory[] { new byte[] { 0x11, 0x22, 0x33 } })); + var sourceEvent = new EventData(sourceMessage); + + // Header + + sourceMessage.Header.DeliveryCount = 123; + sourceMessage.Header.Durable = true; + sourceMessage.Header.FirstAcquirer = true; + sourceMessage.Header.Priority = 1; + sourceMessage.Header.TimeToLive = TimeSpan.FromDays(2); + + // Properties + + sourceMessage.Properties.AbsoluteExpiryTime = new DateTimeOffset(2015, 10, 27, 0, 0 ,0 ,0, TimeSpan.Zero); + sourceMessage.Properties.ContentEncoding = "utf-8"; + sourceMessage.Properties.ContentType = "test/unit"; + sourceMessage.Properties.CorrelationId = new AmqpMessageId("OU812"); + sourceMessage.Properties.CreationTime = new DateTimeOffset(2012, 3, 4, 8, 0, 0, 0, TimeSpan.Zero); + sourceMessage.Properties.GroupId = "Red Squad"; + sourceMessage.Properties.GroupSequence = 76; + sourceMessage.Properties.MessageId = new AmqpMessageId("Bob"); + sourceMessage.Properties.ReplyTo = new AmqpAddress("1407 Graymalkin Lane"); + sourceMessage.Properties.ReplyToGroupId = "Home"; + sourceMessage.Properties.Subject = "You'll never believe this weight loss secret!"; + sourceMessage.Properties.To = new AmqpAddress("http://some.server.com"); + sourceMessage.Properties.UserId = new byte[] { 0x11, 0x22 }; + + // Application Properties + + sourceMessage.ApplicationProperties.Add("One", TimeSpan.FromMinutes(5)); + sourceMessage.ApplicationProperties.Add("Two", 2); + + // Delivery Annotations + + sourceMessage.DeliveryAnnotations.Add("Three", 3); + sourceMessage.DeliveryAnnotations.Add("Four", new DateTimeOffset(2015, 10, 27, 0, 0, 0, TimeSpan.Zero)); + + // Message Annotations + + sourceMessage.MessageAnnotations.Add("Five", 5); + sourceMessage.MessageAnnotations.Add("Six", 6.0f); + + // Footer + + sourceMessage.Footer.Add("Seven", 7); + sourceMessage.Footer.Add("Eight", "8"); + + var converter = new AmqpMessageConverter(); + using var tempMessage = converter.CreateMessageFromEvent(sourceEvent); + var convertedEvent = converter.CreateEventFromMessage(tempMessage); + var convertedMessage = convertedEvent.GetRawAmqpMessage(); + + Assert.That(tempMessage, Is.Not.Null, "The temporary AMQP message should have been created."); + Assert.That(convertedEvent, Is.Not.Null, "The translated event should have been created."); + Assert.That(convertedMessage.GetEventBody().ToArray(), Is.EquivalentTo(sourceMessage.GetEventBody().ToArray()), "The data body should match."); + Assert.That(convertedMessage.ApplicationProperties, Is.EquivalentTo(sourceMessage.ApplicationProperties), "The application properties should match."); + Assert.That(convertedMessage.DeliveryAnnotations, Is.EquivalentTo(sourceMessage.DeliveryAnnotations), "The delivery annotations should match."); + Assert.That(convertedMessage.MessageAnnotations, Is.EquivalentTo(sourceMessage.MessageAnnotations), "The message annotations should match."); + Assert.That(convertedMessage.Footer, Is.EquivalentTo(sourceMessage.Footer), "The footer should match."); + + // Header + + Assert.That(convertedMessage.Header.DeliveryCount, Is.EqualTo(sourceMessage.Header.DeliveryCount), "The delivery count should match."); + Assert.That(convertedMessage.Header.Durable, Is.EqualTo(sourceMessage.Header.Durable), "The durable flag should match."); + Assert.That(convertedMessage.Header.FirstAcquirer, Is.EqualTo(sourceMessage.Header.FirstAcquirer), "The first acquirer flag should match."); + Assert.That(convertedMessage.Header.Priority, Is.EqualTo(sourceMessage.Header.Priority), "The priority should match."); + Assert.That(convertedMessage.Header.TimeToLive, Is.EqualTo(sourceMessage.Header.TimeToLive), "The time to live should match."); + + // Properties + + Assert.That(convertedMessage.Properties.AbsoluteExpiryTime, Is.EqualTo(sourceMessage.Properties.AbsoluteExpiryTime), "The expiry time should match."); + Assert.That(convertedMessage.Properties.ContentEncoding, Is.EqualTo(sourceMessage.Properties.ContentEncoding), "The content encoding should match."); + Assert.That(convertedMessage.Properties.ContentType, Is.EqualTo(sourceMessage.Properties.ContentType), "The content type should match."); + Assert.That(convertedMessage.Properties.CorrelationId, Is.EqualTo(sourceMessage.Properties.CorrelationId), "The correlation identifier should match."); + Assert.That(convertedMessage.Properties.CreationTime, Is.EqualTo(sourceMessage.Properties.CreationTime), "The creation time should match."); + Assert.That(convertedMessage.Properties.GroupId, Is.EqualTo(sourceMessage.Properties.GroupId), "The group identifier should match."); + Assert.That(convertedMessage.Properties.GroupSequence, Is.EqualTo(sourceMessage.Properties.GroupSequence), "The group sequence should match."); + Assert.That(convertedMessage.Properties.MessageId, Is.EqualTo(sourceMessage.Properties.MessageId), "The message identifier should match."); + Assert.That(convertedMessage.Properties.ReplyTo, Is.EqualTo(sourceMessage.Properties.ReplyTo), "The reply-to address should match."); + Assert.That(convertedMessage.Properties.ReplyToGroupId, Is.EqualTo(sourceMessage.Properties.ReplyToGroupId), "The reply-to group identifier should match."); + Assert.That(convertedMessage.Properties.Subject, Is.EqualTo(sourceMessage.Properties.Subject), "The subject should match."); + Assert.That(convertedMessage.Properties.To, Is.EqualTo(sourceMessage.Properties.To), "The to address should match."); + Assert.That(convertedMessage.Properties.UserId.Value.ToArray(), Is.EquivalentTo(sourceMessage.Properties.UserId.Value.ToArray()), "The user identifier should match."); + } + + /// + /// Verifies functionality of the + /// method. + /// + /// + [Test] + public void AnEventWithValueBodyCanBeTranslatedToItself() + { + var sourceValue = new Dictionary { { "key", "value" } }; + var sourceMessage = new AmqpAnnotatedMessage(AmqpMessageBody.FromValue(sourceValue)); + var sourceEvent = new EventData(sourceMessage); + + var converter = new AmqpMessageConverter(); + using var tempMessage = converter.CreateMessageFromEvent(sourceEvent); + var convertedEvent = converter.CreateEventFromMessage(tempMessage); + var convertedMessage = convertedEvent.GetRawAmqpMessage(); + + Assert.That(tempMessage, Is.Not.Null, "The temporary AMQP message should have been created."); + Assert.That(convertedEvent, Is.Not.Null, "The translated event should have been created."); + Assert.That(convertedMessage.Body.TryGetValue(out var convertedValue), Is.True, "The message should have a value body."); + Assert.That(convertedValue, Is.EquivalentTo(sourceValue), "The value body should match."); + } + + /// + /// Verifies functionality of the + /// method. + /// + /// + [Test] + public void AnEventWithSequenceBodyCanBeTranslatedToItself() + { + var sourceValue = new[] { new List { "1", 2 } }; + var sourceMessage = new AmqpAnnotatedMessage(AmqpMessageBody.FromSequence(sourceValue)); + var sourceEvent = new EventData(sourceMessage); + + var converter = new AmqpMessageConverter(); + using var tempMessage = converter.CreateMessageFromEvent(sourceEvent); + var convertedEvent = converter.CreateEventFromMessage(tempMessage); + var convertedMessage = convertedEvent.GetRawAmqpMessage(); + + Assert.That(tempMessage, Is.Not.Null, "The temporary AMQP message should have been created."); + Assert.That(convertedEvent, Is.Not.Null, "The translated event should have been created."); + Assert.That(convertedMessage.Body.TryGetSequence(out var convertedValue), Is.True, "The message should have a value body."); + + Assert.That(sourceValue.Count, Is.EqualTo(1), "The source sequence should have one embedded list."); + Assert.That(convertedValue.Count, Is.EqualTo(1), "The converted sequence should have one embedded list."); + Assert.That(convertedValue.First(), Is.EquivalentTo(sourceValue.First()), "The sequence embedded list should match."); + } + /// /// Verifies functionality of the /// method. diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/tests/Core/EventDataTests.cs b/sdk/eventhub/Azure.Messaging.EventHubs/tests/Core/EventDataTests.cs index 4b54f426fbd20..f2265ad863b7a 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs/tests/Core/EventDataTests.cs +++ b/sdk/eventhub/Azure.Messaging.EventHubs/tests/Core/EventDataTests.cs @@ -28,7 +28,7 @@ public void ConstructorDoesNotCreatePropertiesByDefault() { var eventData = new EventData(new BinaryData(Array.Empty())); - Assert.That(eventData.HasProperties, Is.False, "The user properties should be created lazily."); + Assert.That(eventData.GetRawAmqpMessage().HasSection(AmqpMessageSection.ApplicationProperties), Is.False, "The user properties should be created lazily."); Assert.That(GetSystemPropertiesBackingStore(eventData), Is.Null, "The system properties should be the static empty set."); } diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/tests/Primitives/PartitionReceiverLiveTests.cs b/sdk/eventhub/Azure.Messaging.EventHubs/tests/Primitives/PartitionReceiverLiveTests.cs index 050d61475dbc5..e369db5f63415 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs/tests/Primitives/PartitionReceiverLiveTests.cs +++ b/sdk/eventhub/Azure.Messaging.EventHubs/tests/Primitives/PartitionReceiverLiveTests.cs @@ -9,6 +9,8 @@ using System.Net.WebSockets; using System.Threading; using System.Threading.Tasks; +using Azure.Core.Amqp; +using Azure.Messaging.EventHubs.Amqp; using Azure.Messaging.EventHubs.Authorization; using Azure.Messaging.EventHubs.Consumer; using Azure.Messaging.EventHubs.Core; @@ -1731,6 +1733,207 @@ public async Task ReceiverRespectsTheWaitTimeWhenReading() } } + /// + /// Verifies that the can read a published + /// event. + /// + /// + [Test] + public async Task ReceiverCanReadEventsWithAFullyPopulatedAmqpMessage() + { + await using (EventHubScope scope = await EventHubScope.CreateAsync(1)) + { + using var cancellationSource = new CancellationTokenSource(); + cancellationSource.CancelAfter(EventHubsTestEnvironment.Instance.TestExecutionTimeLimit); + + var connectionString = EventHubsTestEnvironment.Instance.BuildConnectionStringForEventHub(scope.EventHubName); + var partition = (await QueryPartitionsAsync(connectionString, cancellationSource.Token)).First(); + var message = new AmqpAnnotatedMessage(AmqpMessageBody.FromData(new ReadOnlyMemory[] { new byte[] { 0x11, 0x22, 0x33 } })); + var eventData = new EventData(message); + + // Header + + message.Header.DeliveryCount = 123; + message.Header.Durable = true; + message.Header.FirstAcquirer = true; + message.Header.Priority = 1; + message.Header.TimeToLive = TimeSpan.FromDays(2); + + // Properties + + message.Properties.AbsoluteExpiryTime = new DateTimeOffset(2015, 10, 27, 0, 0 ,0 ,0, TimeSpan.Zero); + message.Properties.ContentEncoding = "utf-8"; + message.Properties.ContentType = "test/unit"; + message.Properties.CorrelationId = new AmqpMessageId("OU812"); + message.Properties.CreationTime = new DateTimeOffset(2012, 3, 4, 8, 0, 0, 0, TimeSpan.Zero); + message.Properties.GroupId = "Red Squad"; + message.Properties.GroupSequence = 76; + message.Properties.MessageId = new AmqpMessageId("Bob"); + message.Properties.ReplyTo = new AmqpAddress("1407 Graymalkin Lane"); + message.Properties.ReplyToGroupId = "Home"; + message.Properties.Subject = "You'll never believe this weight loss secret!"; + message.Properties.To = new AmqpAddress("http://some.server.com"); + message.Properties.UserId = new byte[] { 0x11, 0x22 }; + + // Application Properties + + message.ApplicationProperties.Add("EventGenerator::Identifier", Guid.NewGuid().ToString()); + message.ApplicationProperties.Add("One", TimeSpan.FromMinutes(5)); + message.ApplicationProperties.Add("Two", 2); + + // Delivery Annotations + + message.DeliveryAnnotations.Add("Three", 3); + message.DeliveryAnnotations.Add("Four", new DateTimeOffset(2015, 10, 27, 0, 0, 0, TimeSpan.Zero)); + + // Message Annotations + + message.MessageAnnotations.Add("Five", 5); + message.MessageAnnotations.Add("Six", 6.0f); + + // Footer + + message.Footer.Add("Seven", 7); + message.Footer.Add("Eight", "8"); + + // Publish the event and then read it back. + + await using var producer = new EventHubProducerClient(connectionString); + await producer.SendAsync(new[] { eventData }, new SendEventOptions { PartitionId = partition }); + + await using var receiver = new PartitionReceiver(EventHubConsumerClient.DefaultConsumerGroupName, partition, EventPosition.Earliest, connectionString); + var readState = await ReadEventsAsync(receiver, 1, cancellationSource.Token); + + Assert.That(cancellationSource.IsCancellationRequested, Is.False, "The cancellation token should not have been signaled."); + Assert.That(readState.Events.Count, Is.EqualTo(1), "A single event was sent."); + cancellationSource.Cancel(); + + // Validate the extended event attributes. Note that the header and delivery annotations are per-hop + // values and should not be expected to round-trip. A subset of the other sections are broker-owned + // and should be expected to change. + + var readMessage = readState.Events.First().Value.GetRawAmqpMessage(); + + Assert.That(readMessage.GetEventBody().ToArray(), Is.EquivalentTo(message.GetEventBody().ToArray()), "The data body should match."); + Assert.That(readMessage.ApplicationProperties, Is.EquivalentTo(message.ApplicationProperties), "The application properties should match."); + Assert.That(readMessage.Footer, Is.EquivalentTo(message.Footer), "The footer should match."); + + // Properties + + Assert.That(readMessage.Properties.AbsoluteExpiryTime, Is.EqualTo(message.Properties.AbsoluteExpiryTime), "The expiry time should match."); + Assert.That(readMessage.Properties.ContentEncoding, Is.EqualTo(message.Properties.ContentEncoding), "The content encoding should match."); + Assert.That(readMessage.Properties.ContentType, Is.EqualTo(message.Properties.ContentType), "The content type should match."); + Assert.That(readMessage.Properties.CorrelationId, Is.EqualTo(message.Properties.CorrelationId), "The correlation identifier should match."); + Assert.That(readMessage.Properties.CreationTime, Is.EqualTo(message.Properties.CreationTime), "The creation time should match."); + Assert.That(readMessage.Properties.GroupId, Is.EqualTo(message.Properties.GroupId), "The group identifier should match."); + Assert.That(readMessage.Properties.GroupSequence, Is.EqualTo(message.Properties.GroupSequence), "The group sequence should match."); + Assert.That(readMessage.Properties.MessageId, Is.EqualTo(message.Properties.MessageId), "The message identifier should match."); + Assert.That(readMessage.Properties.ReplyTo, Is.EqualTo(message.Properties.ReplyTo), "The reply-to address should match."); + Assert.That(readMessage.Properties.ReplyToGroupId, Is.EqualTo(message.Properties.ReplyToGroupId), "The reply-to group identifier should match."); + Assert.That(readMessage.Properties.Subject, Is.EqualTo(message.Properties.Subject), "The subject should match."); + Assert.That(readMessage.Properties.To, Is.EqualTo(message.Properties.To), "The to address should match."); + Assert.That(readMessage.Properties.UserId.Value.ToArray(), Is.EquivalentTo(message.Properties.UserId.Value.ToArray()), "The user identifier should match."); + + // Message Annotations + + foreach (var key in message.MessageAnnotations.Keys) + { + Assert.That(readMessage.MessageAnnotations.ContainsKey(key), $"The message annotation key [{ key }] should be present."); + Assert.That(readMessage.MessageAnnotations[key], Is.EqualTo(message.MessageAnnotations[key]), $"The message annotation [{ key }] should match the expected value."); + } + } + } + + /// + /// Verifies that the can read a published + /// event. + /// + /// + [Test] + public async Task ReceiverCanReadEventsWithAValueBody() + { + await using (EventHubScope scope = await EventHubScope.CreateAsync(1)) + { + using var cancellationSource = new CancellationTokenSource(); + cancellationSource.CancelAfter(EventHubsTestEnvironment.Instance.TestExecutionTimeLimit); + + var connectionString = EventHubsTestEnvironment.Instance.BuildConnectionStringForEventHub(scope.EventHubName); + var partition = (await QueryPartitionsAsync(connectionString, cancellationSource.Token)).First(); + var value = new Dictionary { { "key", "value" } }; + var message = new AmqpAnnotatedMessage(AmqpMessageBody.FromValue(value)); + var eventData = new EventData(message); + + message.ApplicationProperties.Add("EventGenerator::Identifier", Guid.NewGuid().ToString()); + + // Publish the event and then read it back. + + await using var producer = new EventHubProducerClient(connectionString); + await producer.SendAsync(new[] { eventData }, new SendEventOptions { PartitionId = partition }); + + await using var receiver = new PartitionReceiver(EventHubConsumerClient.DefaultConsumerGroupName, partition, EventPosition.Earliest, connectionString); + var readState = await ReadEventsAsync(receiver, 1, cancellationSource.Token); + + Assert.That(cancellationSource.IsCancellationRequested, Is.False, "The cancellation token should not have been signaled."); + Assert.That(readState.Events.Count, Is.EqualTo(1), "A single event was sent."); + cancellationSource.Cancel(); + + // Validate the extended event attributes. Note that the header and delivery annotations are per-hop + // values and should not be expected to round-trip. A subset of the other sections are broker-owned + // and should be expected to change. + + var readMessage = readState.Events.First().Value.GetRawAmqpMessage(); + + Assert.That(readMessage.Body.TryGetValue(out var readValue), Is.True, "The message should have a value body."); + Assert.That(readValue, Is.EquivalentTo(value), "The value body should match."); + } + } + + /// + /// Verifies that the can read a published + /// event. + /// + /// + [Test] + public async Task ReceiverCanReadEventsWithASequenceBody() + { + await using (EventHubScope scope = await EventHubScope.CreateAsync(1)) + { + using var cancellationSource = new CancellationTokenSource(); + cancellationSource.CancelAfter(EventHubsTestEnvironment.Instance.TestExecutionTimeLimit); + + var connectionString = EventHubsTestEnvironment.Instance.BuildConnectionStringForEventHub(scope.EventHubName); + var partition = (await QueryPartitionsAsync(connectionString, cancellationSource.Token)).First(); + var value = new[] { new List { "1", 2 } }; + var message = new AmqpAnnotatedMessage(AmqpMessageBody.FromSequence(value)); + var eventData = new EventData(message); + + message.ApplicationProperties.Add("EventGenerator::Identifier", Guid.NewGuid().ToString()); + + // Publish the event and then read it back. + + await using var producer = new EventHubProducerClient(connectionString); + await producer.SendAsync(new[] { eventData }, new SendEventOptions { PartitionId = partition }); + + await using var receiver = new PartitionReceiver(EventHubConsumerClient.DefaultConsumerGroupName, partition, EventPosition.Earliest, connectionString); + var readState = await ReadEventsAsync(receiver, 1, cancellationSource.Token); + + Assert.That(cancellationSource.IsCancellationRequested, Is.False, "The cancellation token should not have been signaled."); + Assert.That(readState.Events.Count, Is.EqualTo(1), "A single event was sent."); + cancellationSource.Cancel(); + + // Validate the extended event attributes. Note that the header and delivery annotations are per-hop + // values and should not be expected to round-trip. A subset of the other sections are broker-owned + // and should be expected to change. + + var readMessage = readState.Events.First().Value.GetRawAmqpMessage(); + + Assert.That(readMessage.Body.TryGetSequence(out var readValue), Is.True, "The message should have a value body."); + Assert.That(value.Count, Is.EqualTo(1), "The source sequence should have one embedded list."); + Assert.That(readValue.Count, Is.EqualTo(1), "The converted sequence should have one embedded list."); + Assert.That(readValue.First(), Is.EquivalentTo(value.First()), "The sequence embedded list should match."); + } + } + /// /// Verifies that the is able to /// connect to the Event Hubs service and perform operations. diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/tests/Producer/EventHubProducerClientLiveTests.cs b/sdk/eventhub/Azure.Messaging.EventHubs/tests/Producer/EventHubProducerClientLiveTests.cs index 3fc9e792c2dd3..9c579f7501adc 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs/tests/Producer/EventHubProducerClientLiveTests.cs +++ b/sdk/eventhub/Azure.Messaging.EventHubs/tests/Producer/EventHubProducerClientLiveTests.cs @@ -9,6 +9,7 @@ using System.Text; using System.Threading; using System.Threading.Tasks; +using Azure.Core.Amqp; using Azure.Messaging.EventHubs.Authorization; using Azure.Messaging.EventHubs.Consumer; using Azure.Messaging.EventHubs.Core; @@ -1101,6 +1102,112 @@ public async Task ProducerCannotSendWhenProxyIsInvalid() } } + /// + /// Verifies that the is able to + /// connect to the Event Hubs service and perform operations. + /// + /// + [Test] + public async Task ProducerCanSendEventsWithAFullyPopulatedAmqpMessage() + { + await using (EventHubScope scope = await EventHubScope.CreateAsync(1)) + { + var message = new AmqpAnnotatedMessage(AmqpMessageBody.FromData(new ReadOnlyMemory[] { new byte[] { 0x11, 0x22, 0x33 } })); + var eventData = new EventData(message); + + // Header + + message.Header.DeliveryCount = 123; + message.Header.Durable = true; + message.Header.FirstAcquirer = true; + message.Header.Priority = 1; + message.Header.TimeToLive = TimeSpan.FromDays(2); + + // Properties + + message.Properties.AbsoluteExpiryTime = new DateTimeOffset(2015, 10, 27, 0, 0 ,0 ,0, TimeSpan.Zero); + message.Properties.ContentEncoding = "utf-8"; + message.Properties.ContentType = "test/unit"; + message.Properties.CorrelationId = new AmqpMessageId("OU812"); + message.Properties.CreationTime = new DateTimeOffset(2012, 3, 4, 8, 0, 0, 0, TimeSpan.Zero); + message.Properties.GroupId = "Red Squad"; + message.Properties.GroupSequence = 76; + message.Properties.MessageId = new AmqpMessageId("Bob"); + message.Properties.ReplyTo = new AmqpAddress("1407 Graymalkin Lane"); + message.Properties.ReplyToGroupId = "Home"; + message.Properties.Subject = "You'll never believe this weight loss secret!"; + message.Properties.To = new AmqpAddress("http://some.server.com"); + message.Properties.UserId = new byte[] { 0x11, 0x22 }; + + // Application Properties + + message.ApplicationProperties.Add("One", TimeSpan.FromMinutes(5)); + message.ApplicationProperties.Add("Two", 2); + + // Delivery Annotations + + message.DeliveryAnnotations.Add("Three", 3); + message.DeliveryAnnotations.Add("Four", new DateTimeOffset(2015, 10, 27, 0, 0, 0, TimeSpan.Zero)); + + // Message Annotations + + message.MessageAnnotations.Add("Five", 5); + message.MessageAnnotations.Add("Six", 6.0f); + + // Footer + + message.Footer.Add("Seven", 7); + message.Footer.Add("Eight", "8"); + + // Attempt to send and validate the operation was not rejected. + + await using var producer = new EventHubProducerClient(EventHubsTestEnvironment.Instance.BuildConnectionStringForEventHub(scope.EventHubName)); + Assert.That(async () => await producer.SendAsync(new[] { eventData }), Throws.Nothing); + } + } + + /// + /// Verifies that the is able to + /// connect to the Event Hubs service and perform operations. + /// + /// + [Test] + public async Task ProducerCanSendEventsWithValueBodies() + { + await using (EventHubScope scope = await EventHubScope.CreateAsync(1)) + { + var value = new Dictionary { { "key", "value" } }; + var message = new AmqpAnnotatedMessage(AmqpMessageBody.FromValue(value)); + var eventData = new EventData(message); + + // Attempt to send and validate the operation was not rejected. + + await using var producer = new EventHubProducerClient(EventHubsTestEnvironment.Instance.BuildConnectionStringForEventHub(scope.EventHubName)); + Assert.That(async () => await producer.SendAsync(new[] { eventData }), Throws.Nothing); + } + } + + /// + /// Verifies that the is able to + /// connect to the Event Hubs service and perform operations. + /// + /// + [Test] + public async Task ProducerCanSendEventsWithSequenceBodies() + { + await using (EventHubScope scope = await EventHubScope.CreateAsync(1)) + { + var sequence = new[] { new List { "1", 2 } }; + var message = new AmqpAnnotatedMessage(AmqpMessageBody.FromSequence(sequence)); + var eventData = new EventData(message); + + // Attempt to send and validate the operation was not rejected. + + await using var producer = new EventHubProducerClient(EventHubsTestEnvironment.Instance.BuildConnectionStringForEventHub(scope.EventHubName)); + Assert.That(async () => await producer.SendAsync(new[] { eventData }), Throws.Nothing); + } + } + /// /// Verifies that the is able to /// connect to the Event Hubs service.