I am experimenting with a new NServiceBus project utilizing Azure Storage Queues for message transport and JSON serialization using custom message unwrapping logic seen here:
var jsonSerializer = new Newtonsoft.Json.JsonSerializer();
transportExtensions.UnwrapMessagesWith(cloudQueueMessage =>
{
using (var stream = new MemoryStream(cloudQueueMessage.AsBytes))
using (var streamReader = new StreamReader(stream))
using (var textReader = new JsonTextReader(streamReader))
{
try
{
var jObject = JObject.Load(textReader);
using (var jsonReader = jObject.CreateReader())
{
// Try deserialize to a NServiceBus envelope first
var wrapper = jsonSerializer.Deserialize<MessageWrapper>(jsonReader);
if (wrapper.MessageIntent != default)
{
// This was a envelope message
return wrapper;
}
}
// Otherwise this was an EventGrid event
using (var jsonReader = jObject.CreateReader())
{
var @event = jsonSerializer.Deserialize<EventGridEvent>(jsonReader);
var wrapper = new MessageWrapper
{
Id = @event.Id,
Headers = new Dictionary<string, string>
{
{ "NServiceBus.EnclosedMessageTypes", @event.EventType },
{ "NServiceBus.MessageIntent", "Publish" },
{ "EventGrid.topic", @event.Topic },
{ "EventGrid.subject", @event.Subject },
{ "EventGrid.eventTime", @event.EventTime.ToString("u") },
{ "EventGrid.dataVersion", @event.DataVersion },
{ "EventGrid.metadataVersion", @event.MetadataVersion },
},
Body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(@event.Data)),
MessageIntent = MessageIntentEnum.Publish
};
return wrapper;
}
}
catch
{
logger.Error("Message deserialization failed, sending message to error queue");
throw;
}
}
});
The custom message unwrapping logic works correctly for properly formatted JSON messages and when an improperly formatted JSON message is put into the input queue the custom message unwrapping logic will error out on the first line inside the usings where I create the jObject which is the expected behavior. However, when the custom message unwrapping logic fails the error will get caught by the logic in the MessageRetrieved class which is part of the NServiceBus.Azure.Transports.WindowsAzureStorageQueues NuGet package (v8.2.0) seen below:
public async Task<MessageWrapper> Unwrap()
{
try
{
Logger.DebugFormat("Unwrapping message with native ID: '{0}'", rawMessage.Id);
return unwrapper.Unwrap(rawMessage);
}
catch (Exception ex)
{
await errorQueue.AddMessageAsync(rawMessage).ConfigureAwait(false);
await inputQueue.DeleteMessageAsync(rawMessage).ConfigureAwait(false);
throw new SerializationException($"Failed to deserialize message envelope for message with id {rawMessage.Id}. Make sure the configured serializer is used across all endpoints or configure the message wrapper serializer for this endpoint using the `SerializeMessageWrapperWith` extension on the transport configuration. Please refer to the Azure Storage Queue Transport configuration documentation for more details.", ex);
}
}
The first line of the try catch runs correctly adding the message to the configured error queue, however, when it does that, it appears to be changing the message ID and popreceipt of the raw message as seen here:
Then when the next line runs attempting to remove the original message from the input queue it is unable to find it as according to this article Delete Message (REST API) - Azure Storage | Microsoft Learn it requires the original message ID and pop reciept which have now changed leading to the following error being thrown:
2020-04-20 14:17:58,603 WARN : Azure Storage Queue transport failed pushing a message through pipeline
Type: Microsoft.WindowsAzure.Storage.StorageException
Message: The remote server returned an error: (404) Not Found.
Source: Microsoft.WindowsAzure.Storage
StackTrace:
at Microsoft.WindowsAzure.Storage.Core.Executor.Executor.EndExecuteAsync[T](IAsyncResult result) in c:\Program Files (x86)\Jenkins\workspace\release_dotnet_master\Lib\ClassLibraryCommon\Core\Executor\Executor.cs:line 50
at Microsoft.WindowsAzure.Storage.Core.Util.AsyncExtensions.<>c__DisplayClass7.<CreateCallbackVoid>b__5(IAsyncResult ar) in c:\Program Files (x86)\Jenkins\workspace\release_dotnet_master\Lib\ClassLibraryCommon\Core\Util\AsyncExtensions.cs:line 121
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at NServiceBus.Transport.AzureStorageQueues.MessageRetrieved.<Unwrap>d__3.MoveNext() in C:\BuildAgent\work\3c19e2a032c05076\src\Transport\MessageRetrieved.cs:line 40
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at NServiceBus.Transport.AzureStorageQueues.MessagePump.<InnerReceive>d__7.MoveNext() in C:\BuildAgent\work\3c19e2a032c05076\src\Transport\MessagePump.cs:line 153
TargetSite: T EndExecuteAsync[T](System.IAsyncResult)
Is this an issue with the NServiceBus package logic, or is something in my custom message unwrapping logic causing these values to change?