Merge pull request #6924 from Particular/1005_delayed_retries

make it clear that the user needs to deal with NServiceBus messages in addition to native messages

Author: jasontaylordev

Snippets/ASQ/ASQN_12/CustomEnvelopeUnwrapper.cs

@@ -16,27 +16,22 @@ class CustomEnvelopeUnwrapper
         {
             MessageUnwrapper = queueMessage =>
             {
-                using (var stream = new MemoryStream(Convert.FromBase64String(queueMessage.MessageText)))
-                using (var streamReader = new StreamReader(stream))
-                using (var textReader = new JsonTextReader(streamReader))
+                //Determine whether the message is one expected by the standard program flow.
+                // All other messages should be forwarded to the framework by returning null.
+                //NOTE: More complex methods may be needed in some scenarios to determine
+                // whether the message is of an expected type, but this check
+                // should be kept as lightweight as possible. Deserialization of the message
+                // body happens later in the pipeline.
+                return queueMessage.MessageText.Contains("MyMessageIdFieldName") &&
+                       queueMessage.MessageText.Contains("MyMessageCustomPropertyFieldName")
+                //this was a native message just return the body as is with no headers
+                ? new MessageWrapper
                 {
-                    //try deserialize to a NServiceBus envelope first
-                    var wrapper = jsonSerializer.Deserialize<MessageWrapper>(textReader);
-
-                    if (wrapper.Id != null)
-                    {
-                        //this was a envelope message
-                        return wrapper;
-                    }
-
-                    //this was a native message just return the body as is with no headers
-                    return new MessageWrapper
-                    {
-                        Id = queueMessage.MessageId,
-                        Headers = new Dictionary<string, string>(),
-                        Body = Convert.FromBase64String(queueMessage.MessageText)
-                    };
+                    Id = queueMessage.MessageId,
+                    Headers = new Dictionary<string, string>(),
+                    Body = queueMessage.Body.ToArray()
                 }
+                : null;
             }
         };
 

Snippets/ASQ/ASQN_13/CustomEnvelopeUnwrapper.cs

@@ -1,5 +1,4 @@
-using System;
-using System.Collections.Generic;
+using System.Buffers.Text;
 using System.Text.Json;
 using NServiceBus;
 using NServiceBus.Azure.Transports.WindowsAzureStorageQueues;
@@ -15,24 +14,22 @@ class CustomEnvelopeUnwrapper
         {
             MessageUnwrapper = queueMessage =>
             {
-                var messageText = Convert.FromBase64String(queueMessage.MessageText);
-
-                //try deserialize to a NServiceBus envelope first
-                var wrapper = JsonSerializer.Deserialize<MessageWrapper>(messageText);
-
-                if (wrapper?.Id != null)
-                {
-                    //this was a envelope message
-                    return wrapper;
-                }
-
+                //Determine whether the message is one expected by the standard program flow.
+                // All other messages should be forwarded to the framework by returning null.
+                //NOTE: More complex methods may be needed in some scenarios to determine
+                // whether the message is of an expected type, but this check
+                // should be kept as lightweight as possible. Deserialization of the message
+                // body happens later in the pipeline.
+                return queueMessage.MessageText.Contains("MyMessageIdFieldName") &&
+                       queueMessage.MessageText.Contains("MyMessageCustomPropertyFieldName")
                 //this was a native message just return the body as is with no headers
-                return new MessageWrapper
+                ? new MessageWrapper 
                 {
                     Id = queueMessage.MessageId,
-                    Headers = new Dictionary<string, string>(),
-                    Body = messageText
-                };
+                    Headers = [],
+                    Body = queueMessage.Body.ToArray()
+                }
+                : null;
             }
         };
 

samples/azure/native-integration-asq/ASQN_12/NativeSender/Program.cs

@@ -27,6 +27,7 @@ static async Task Main()
 
             var nativeMessage = new NativeMessage
             {
+                NativeMessageId = Guid.NewGuid(),
                 Content = $"Hello from native sender @ {DateTimeOffset.Now}"
             };
 

samples/azure/native-integration-asq/ASQN_12/Receiver/Program.cs

@@ -21,15 +21,19 @@ static async Task Main()
 
         #region Native-message-mapping
 
-        transport.MessageUnwrapper = message => new MessageWrapper
-        {
-            Id = message.MessageId,
-            Body = message.Body.ToArray(),
-            Headers = new Dictionary<string, string>
+        transport.MessageUnwrapper = message =>
+            message.MessageText.Contains("NativeMessageId") &&
+            message.MessageText.Contains("Content")
+            ? new MessageWrapper
             {
-                { Headers.EnclosedMessageTypes, typeof(NativeMessage).FullName }
+                Id = message.MessageId,
+                Body = message.Body.ToArray(),
+                Headers = new Dictionary<string, string>
+                {
+                    { Headers.EnclosedMessageTypes, typeof(NativeMessage).FullName }
+                }
             }
-        };
+            : null; // not a raw native message - allow the framework to deal with it
 
         #endregion
 

samples/azure/native-integration-asq/ASQN_12/Shared/NativeMessage.cs

@@ -1,6 +1,8 @@
+using System;
 using NServiceBus;
 
 public class NativeMessage : IMessage
 {
+    public Guid NativeMessageId { get; set; }
     public string Content { get; set; }
 }

samples/azure/native-integration-asq/ASQN_13/NativeSender/Program.cs

@@ -23,6 +23,7 @@
 
     var nativeMessage = new NativeMessage
     {
+        NativeMessageId = Guid.NewGuid(),
         Content = $"Hello from native sender @ {DateTimeOffset.Now}"
     };
 

samples/azure/native-integration-asq/ASQN_13/Receiver/Program.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Buffers.Text;
 using System.Collections.Generic;
 
 using NServiceBus;
@@ -18,15 +19,19 @@
 
 #region Native-message-mapping
 
-transport.MessageUnwrapper = message => new MessageWrapper
-{
-    Id = message.MessageId,
-    Body = message.Body.ToArray(),
-    Headers = new Dictionary<string, string>
-            {
-                { Headers.EnclosedMessageTypes, typeof(NativeMessage).FullName }
-            }
-};
+transport.MessageUnwrapper = message =>
+    message.MessageText.Contains("NativeMessageId") &&
+    message.MessageText.Contains("Content")
+    ? new MessageWrapper
+    {
+        Id = message.MessageId,
+        Body = message.Body.ToArray(),
+        Headers = new Dictionary<string, string>
+        {
+            { Headers.EnclosedMessageTypes, typeof(NativeMessage).FullName }
+        }
+    }
+    : null; // not a raw native message - allow the framework to deal with it
 
 #endregion
 

samples/azure/native-integration-asq/ASQN_13/Shared/NativeMessage.cs

@@ -1,6 +1,8 @@
+using System;
 using NServiceBus;
 
 public class NativeMessage : IMessage
 {
+    public Guid NativeMessageId { get; set; }
     public string Content { get; set; }
 }

transports/azure-storage-queues/native-integration_sending_asqn_[11,).partial.md

@@ -0,0 +1,15 @@
+### Custom envelope unwrapper
+
+Azure Storage Queues lacks native header support. NServiceBus solves this by wrapping headers and message body in a custom envelope structure. This envelope is serialized using the configured [serializer](/nservicebus/serialization) for the endpoint before being sent.
+
+Creating this envelope can cause unnecessary complexity if headers are not needed, as is the case in native integration scenarios. For this reason, NServiceBus.Transport.AzureStorageQueues 9.0 and above support configuring a custom envelope unwrapper. 
+
+> [!WARNING]
+> In this scenario, NServiceBus may place messages in your queue in addition to the native messages that are expected, for example if a message results in a [delayed retry](delayed-delivery.md). Any custom envelope unwrapper must verify if the incoming message is capable of being deserialized as a native message, and the resulting body must be serialized according to the configured endpoint serializer. If either of these conditions cannot be met then `MessageUnwrapper` should return `null` to allow the default unwrapper to handle the message.
+
+The snippet below shows custom unwrapping logic that enables both NServiceBus formatted and plain serialized messages to be consumed.
+
+snippet: CustomEnvelopeUnwrapper
+
+> [!NOTE]
+> This feature is currently NOT compatible with ServiceControl. A [ServiceControl transport adapter](/servicecontrol/transport-adapter.md) is required to leverage both.