Merge pull request #361 from Particular/hotfix-8.1.4

Hotfix 8.1.4
Particular · Apr 16, 2019 · 7e51a50 · 7e51a50
2 parents 4e08b1f + fa20c92
commit 7e51a50
Show file tree

Hide file tree

Showing 5 changed files with 10 additions and 13 deletions.
diff --git a/src/Transport/AtLeastOnceReceiveStrategy.cs b/src/Transport/AtLeastOnceReceiveStrategy.cs
@@ -11,10 +11,11 @@ namespace NServiceBus.Transport.AzureStorageQueues
 
     class AtLeastOnceReceiveStrategy : ReceiveStrategy
     {
-        public AtLeastOnceReceiveStrategy(Func<MessageContext, Task> pipeline, Func<ErrorContext, Task<ErrorHandleResult>> errorPipe)
+        public AtLeastOnceReceiveStrategy(Func<MessageContext, Task> pipeline, Func<ErrorContext, Task<ErrorHandleResult>> errorPipe, CriticalError criticalError)
         {
             this.pipeline = pipeline;
             this.errorPipe = errorPipe;
+            this.criticalError = criticalError;
         }
 
         public override async Task Receive(MessageRetrieved retrieved, MessageWrapper message)
@@ -56,8 +57,10 @@ public override async Task Receive(MessageRetrieved retrieved, MessageWrapper me
                 }
                 catch (Exception e)
                 {
-                    Logger.Warn("The error pipeline wasn't able to handle the exception.", e);
+                    criticalError.Raise($"Failed to execute recoverability policy for message with native ID: `{message.Id}`", e);
+
                     await retrieved.Nack().ConfigureAwait(false);
+
                     return;
                 }
 
@@ -78,6 +81,7 @@ public override async Task Receive(MessageRetrieved retrieved, MessageWrapper me
 
         readonly Func<MessageContext, Task> pipeline;
         readonly Func<ErrorContext, Task<ErrorHandleResult>> errorPipe;
+        readonly CriticalError criticalError;
 
         static readonly ILog Logger = LogManager.GetLogger<ReceiveStrategy>();
     }

diff --git a/src/Transport/MessagePump.cs b/src/Transport/MessagePump.cs
@@ -30,7 +30,7 @@ public Task Init(Func<MessageContext, Task> onMessage, Func<ErrorContext, Task<E
             circuitBreaker = new RepeatedFailuresOverTimeCircuitBreaker("AzureStorageQueue-MessagePump", TimeToWaitBeforeTriggering, ex => criticalError.Raise("Failed to receive message from Azure Storage Queue.", ex));
             messageReceiver.PurgeOnStartup = settings.PurgeOnStartup;
 
-            receiveStrategy = ReceiveStrategy.BuildReceiveStrategy(onMessage, onError, settings.RequiredTransactionMode);
+            receiveStrategy = ReceiveStrategy.BuildReceiveStrategy(onMessage, onError, settings.RequiredTransactionMode, criticalError);
 
             return messageReceiver.Init(settings.InputQueue, settings.ErrorQueue);
         }

diff --git a/src/Transport/ReceiveStrategy.cs b/src/Transport/ReceiveStrategy.cs
@@ -9,15 +9,15 @@ abstract class ReceiveStrategy
     {
         public abstract Task Receive(MessageRetrieved retrieved, MessageWrapper message);
 
-        public static ReceiveStrategy BuildReceiveStrategy(Func<MessageContext, Task> pipe, Func<ErrorContext, Task<ErrorHandleResult>> errorPipe, TransportTransactionMode transactionMode)
+        public static ReceiveStrategy BuildReceiveStrategy(Func<MessageContext, Task> pipe, Func<ErrorContext, Task<ErrorHandleResult>> errorPipe, TransportTransactionMode transactionMode, CriticalError criticalError)
         {
             // ReSharper disable once SwitchStatementMissingSomeCases
             switch (transactionMode)
             {
                 case TransportTransactionMode.None:
                     return new AtMostOnceReceiveStrategy(pipe, errorPipe);
                 case TransportTransactionMode.ReceiveOnly:
-                    return new AtLeastOnceReceiveStrategy(pipe, errorPipe);
+                    return new AtLeastOnceReceiveStrategy(pipe, errorPipe, criticalError);
                 default:
                     throw new NotSupportedException($"The TransportTransactionMode {transactionMode} is not supported");
             }

diff --git a/src/TransportTests/ConfigureAzureStorageQueueTransportInfrastructure.cs b/src/TransportTests/ConfigureAzureStorageQueueTransportInfrastructure.cs
@@ -9,7 +9,6 @@
 using NServiceBus.Settings;
 using NServiceBus.TransportTests;
 using NServiceBus.Unicast.Messages;
-using NUnit.Framework;
 
 public class ConfigureAzureStorageQueueTransportInfrastructure : IConfigureTransportInfrastructure
 {
@@ -27,12 +26,6 @@ public TransportConfigurationResult Configure(SettingsHolder settings, Transport
             settings.Set(registry);
         }
 
-        var methodName = TestContext.CurrentContext.Test.MethodName;
-        if (methodName == nameof(When_on_error_throws.Should_reinvoke_on_error_with_original_exception))
-        {
-            throw new IgnoreException("ASQ uses a circuit breaker that is triggered after specific period of time. Critical errors are not reported immediately");
-        }
-
         settings.Set(AzureStorageQueueTransport.SerializerSettingsKey, Tuple.Create<SerializationDefinition, SettingsHolder>(new XmlSerializer(), settings));
 
         var transportExtension = new TransportExtensions<AzureStorageQueueTransport>(settings);

diff --git a/src/TransportTests/NServiceBus.AzureStorageQueues.TransportTests.csproj b/src/TransportTests/NServiceBus.AzureStorageQueues.TransportTests.csproj
@@ -17,7 +17,7 @@
     <PackageReference Include="NUnit3TestAdapter" Version="3.12.0" />
     <PackageReference Include="NUnit" Version="3.11.0" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.9.0" />
-    <PackageReference Include="NServiceBus.TransportTests.Sources" Version="7.1.6" />
+    <PackageReference Include="NServiceBus.TransportTests.Sources" Version="7.2.0-beta0079" />
   </ItemGroup>
 
   <ItemGroup>