Merge branch 'release/3.1.1'

Tapeti.Flow.SQL/SqlConnectionFlowRepository.cs

@@ -1,8 +1,8 @@
 using System;
 using System.Collections.Generic;
 using System.Data;
-using System.Data.SqlClient;
 using System.Threading.Tasks;
+using Microsoft.Data.SqlClient;
 using Newtonsoft.Json;
 
 // Neither of these are available in language version 7 required for .NET Standard 2.0

Tapeti.Flow.SQL/SqlExceptionHelper.cs

@@ -1,7 +1,7 @@
 using System;
 using System.Collections.Generic;
-using System.Data.SqlClient;
 using System.Linq;
+using Microsoft.Data.SqlClient;
 
 // ReSharper disable UnusedMember.Global
 

Tapeti.Flow.SQL/SqlRetryHelper.cs

@@ -1,6 +1,6 @@
 using System;
-using System.Data.SqlClient;
 using System.Threading.Tasks;
+using Microsoft.Data.SqlClient;
 
 namespace Tapeti.Flow.SQL
 {

Tapeti.Flow.SQL/Tapeti.Flow.SQL.csproj

@@ -33,7 +33,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="System.Data.SqlClient" Version="4.8.5" />
+    <PackageReference Include="Microsoft.Data.SqlClient" Version="5.*" />
   </ItemGroup>
 
   <ItemGroup>

Tapeti.Flow/Default/FlowProvider.cs

@@ -407,14 +407,16 @@ namespace Tapeti.Flow.Default
                     if (convergeMethod.Method.DeclaringType != context.HandlerContext.Controller?.GetType())
                         throw new YieldPointException("Converge method must be in the same controller class");
 
-                    await Task.WhenAll(requests.Select(requestInfo =>
-                        flowProvider.SendRequest(
-                            context, 
+                    foreach (var requestInfo in requests)
+                    {
+                        await flowProvider.SendRequest(
+                            context,
                             requestInfo.Message,
                             requestInfo.ResponseHandlerInfo,
                             convergeMethod.Method.Name,
                             convergeMethodSync,
-                            false)));
+                            false);
+                    }
 
                     await context.Store(requests.Any(i => i.ResponseHandlerInfo.IsDurableQueue));
                 });

Tapeti/Connection/TapetiBasicConsumer.cs

@@ -17,7 +17,7 @@ namespace Tapeti.Connection
     /// <summary>
     /// Implements the bridge between the RabbitMQ Client consumer and a Tapeti Consumer
     /// </summary>
-    internal class TapetiBasicConsumer : AsyncDefaultBasicConsumer
+    internal class TapetiBasicConsumer : DefaultBasicConsumer
     {
         private readonly IConsumer consumer;
         private readonly long connectionReference;
@@ -34,7 +34,7 @@ namespace Tapeti.Connection
 
 
         /// <inheritdoc />
-        public override async Task HandleBasicDeliver(string consumerTag,
+        public override void HandleBasicDeliver(string consumerTag,
             ulong deliveryTag,
             bool redelivered,
             string exchange,
@@ -49,16 +49,21 @@ namespace Tapeti.Connection
             //
             // See also: https://github.com/JamesNK/Newtonsoft.Json/issues/1761
             var bodyArray = body.ToArray();
-            
-            try
+
+            // Changing to AsyncDefaultBasicConsumer does not mean HandleBasicDeliver runs in parallel, the Task.Run would
+            // still be necessary, which is why TapetiBasicConsumer is a DefaultBasicConsumer.
+            Task.Run(async () =>
             {
-                var response = await consumer.Consume(exchange, routingKey, new RabbitMQMessageProperties(properties), bodyArray);
-                await onRespond(connectionReference, deliveryTag, response);
-            }
-            catch
-            {
-                await onRespond(connectionReference, deliveryTag, ConsumeResult.Error);
-            }
+                try
+                {
+                    var response = await consumer.Consume(exchange, routingKey, new RabbitMQMessageProperties(properties), bodyArray);
+                    await onRespond(connectionReference, deliveryTag, response);
+                }
+                catch
+                {
+                    await onRespond(connectionReference, deliveryTag, ConsumeResult.Error);
+                }
+            });
         }
     }
 }

Tapeti/Connection/TapetiClient.cs

@@ -777,8 +777,7 @@ namespace Tapeti.Connection
                 Password = connectionParams.Password,
                 AutomaticRecoveryEnabled = false,
                 TopologyRecoveryEnabled = false,
-                RequestedHeartbeat = TimeSpan.FromSeconds(30),
-                DispatchConsumersAsync = true
+                RequestedHeartbeat = TimeSpan.FromSeconds(30)
             };
 
             if (connectionParams.ClientProperties != null)