enforcing transaction isolation level on peeks (#311)

Particular · Oct 7, 2016 · d75fb52 · d75fb52
1 parent 7f1645d
commit d75fb52
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 0 deletions.
diff --git a/src/NServiceBus.SqlServer/Legacy/MultiInstance/LegacyQueuePeeker.cs b/src/NServiceBus.SqlServer/Legacy/MultiInstance/LegacyQueuePeeker.cs
@@ -4,6 +4,7 @@
     using System.Data.SqlClient;
     using System.Threading;
     using System.Threading.Tasks;
+    using System.Transactions;
     using Logging;
 
     class LegacyQueuePeeker : IPeekMessagesInQueue
@@ -20,6 +21,7 @@ public async Task<int> Peek(TableBasedQueue inputQueue, RepeatedFailuresOverTime
 
             try
             {
+                using (var scope = new TransactionScope(TransactionScopeOption.RequiresNew, new TransactionOptions {IsolationLevel = IsolationLevel.ReadCommitted}, TransactionScopeAsyncFlowOption.Enabled))
                 using (var connection = await connectionFactory.OpenNewConnection(inputQueue.TransportAddress).ConfigureAwait(false))
                 {
                     messageCount = await inputQueue.TryPeek(connection, cancellationToken).ConfigureAwait(false);
@@ -32,6 +34,8 @@ public async Task<int> Peek(TableBasedQueue inputQueue, RepeatedFailuresOverTime
 
                         await Task.Delay(settings.Delay, cancellationToken).ConfigureAwait(false);
                     }
+
+                    scope.Complete();
                 }
             }
             catch (OperationCanceledException)

diff --git a/src/NServiceBus.SqlServer/Receiving/QueuePeeker.cs b/src/NServiceBus.SqlServer/Receiving/QueuePeeker.cs
@@ -4,6 +4,7 @@
     using System.Data.SqlClient;
     using System.Threading;
     using System.Threading.Tasks;
+    using System.Transactions;
     using Logging;
 
     class QueuePeeker : IPeekMessagesInQueue
@@ -20,6 +21,7 @@ public async Task<int> Peek(TableBasedQueue inputQueue, RepeatedFailuresOverTime
 
             try
             {
+                using (var scope = new TransactionScope(TransactionScopeOption.RequiresNew, new TransactionOptions {IsolationLevel = IsolationLevel.ReadCommitted}, TransactionScopeAsyncFlowOption.Enabled))
                 using (var connection = await connectionFactory.OpenNewConnection().ConfigureAwait(false))
                 {
                     messageCount = await inputQueue.TryPeek(connection, cancellationToken).ConfigureAwait(false);
@@ -32,6 +34,8 @@ public async Task<int> Peek(TableBasedQueue inputQueue, RepeatedFailuresOverTime
 
                         await Task.Delay(settings.Delay, cancellationToken).ConfigureAwait(false);
                     }
+
+                    scope.Complete();
                 }
             }
             catch (OperationCanceledException)