NServiceBus с azure очередями хранения будет обрабатывать сообщения без заголовков бесконечно

Question

Я экспериментирую с новым проектом NServiceBus, использующим Azure Очереди хранения для передачи сообщений и JSON сериализации. Я заметил, что когда я запускаю сообщение через очередь, в которой отсутствуют заголовки NServiceBus, например, пустое сообщение JSON: {} Выдается следующее предупреждающее сообщение:

2020-02-06 17:46:35.587 WARN  NServiceBus.Transport.AzureStorageQueues.MessagePump Azure Storage Queue transport failed pushing a message through pipeline
System.ArgumentNullException: Value cannot be null.
Parameter name: nativeMessageId
   at NServiceBus.Transport.IncomingMessage..ctor(String nativeMessageId, Dictionary`2 headers, Byte[] body)
   at NServiceBus.Transport.ErrorContext..ctor(Exception exception, Dictionary`2 headers, String transportMessageId, Byte[] body, TransportTransaction transportTransaction, Int32 immediateProcessingFailures)
   at NServiceBus.Transport.AzureStorageQueues.ReceiveStrategy.CreateErrorContext(MessageRetrieved retrieved, MessageWrapper message, Exception ex, Byte[] body)
   at NServiceBus.Transport.AzureStorageQueues.AtLeastOnceReceiveStrategy.<Receive>d__1.MoveNext()
--- 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()

После чего оно появляется, чтобы остановить обработку сообщения, но оставляет его в очереди. Затем, после ожидания сконфигурированного сообщения невидимого периода, сообщение снова становится видимым в очереди, и NServiceBus будет повторять процесс «предупреждение и остановить обработку» бесконечно. Есть ли способ изменить способ, которым NServiceBus обрабатывает этот сценарий, чтобы он перебрасывал сообщение в сконфигурированную очередь ошибок, когда он не может проанализировать информацию заголовка и не пытается обрабатывать сообщение бесконечно?

Answer 1

Транспорт NServiceBus Storage Queues предполагает, что сообщения приходят с правильным конвертом. Если этот конверт не найден, вы получите исключение, которое видите выше. Для сообщений, которые не созданы NServiceBus или с пользовательским конвертом, пожалуйста, смотрите документацию здесь . Короче говоря, вам нужен пользовательский распаковщик конвертов.

За что отвечает пользовательский распаковщик (обратный вызов), это десериализация сообщения и создание MessageWrapper, с которым NServiceBus ожидает работать.

var transport = endpointConfiguration.UseTransport<AzureStorageQueueTransport>();

transport.UnwrapMessagesWith(cloudQueueMessage =>
{
    using (var stream = new MemoryStream(cloudQueueMessage.AsBytes))
    using (var streamReader = new StreamReader(stream))
    using (var textReader = new JsonTextReader(streamReader))
    {
        //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 = cloudQueueMessage.Id,
            Headers = new Dictionary<string, string>(),
            Body = cloudQueueMessage.AsBytes
        };
    }
});