Merge pull request MarimerLLC#1848 from danielmartind/update-rabbitmq

Fix breaking change from RabbitMq v6

Author: rockfordlhotka

NuGet/Definition/Csla.Channels.RabbitMq.NuSpec

@@ -23,7 +23,7 @@ Data portal channel using RabbitMQ for transport.
     <dependencies>
       <group targetFramework="netstandard2.1">
         <dependency id="Csla" version="[4.6.3-Beta10]" />
-        <dependency id="RabbitMQ.Client" version="5.1.2" />
+        <dependency id="RabbitMQ.Client" version="6.2.1" />
       </group>
     </dependencies>
   </metadata>

Source/Csla.Channels.RabbitMq/Csla.Channels.RabbitMq.csproj

@@ -25,8 +25,8 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="RabbitMQ.Client" Version="5.1.2" />
-    <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All"/>
+    <PackageReference Include="RabbitMQ.Client" Version="6.2.1" />
+    <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
   </ItemGroup>
 
   <ItemGroup>

Source/Csla.Channels.RabbitMq/ProxyListener.cs

@@ -20,10 +20,12 @@ internal class ProxyListener : IDisposable
     /// Gets or sets the connection to the RabbitMQ service.
     /// </summary>
     protected IConnection Connection { get; set; }
+
     /// <summary>
     /// Gets or sets the channel (model) for RabbitMQ.
     /// </summary>
     protected IModel Channel { get; set; }
+
     /// <summary>
     /// Gets or sets the queue for inbound messages
     /// and replies.
@@ -34,7 +36,9 @@ internal class ProxyListener : IDisposable
 
     private static ProxyListener _instance;
 
-    private ProxyListener() { }
+    private ProxyListener()
+    {
+    }
 
     public static ProxyListener GetListener(Uri queueUri)
     {
@@ -113,7 +117,7 @@ public void StartListening()
         Console.WriteLine($"Received reply for {ea.BasicProperties.CorrelationId}");
         if (Wip.WorkInProgress.TryRemove(ea.BasicProperties.CorrelationId, out WipItem item))
         {
-          item.Response = ea.Body;
+          item.Response = ea.Body.ToArray();
           item.ResetEvent.Set();
         }
         else
@@ -148,4 +152,4 @@ public void Dispose()
       IsListening = false;
     }
   }
-}
+}

Source/Csla.Channels.RabbitMq/RabbitMqPortal.cs

@@ -6,7 +6,6 @@
 // <summary>Exposes server-side DataPortal functionality through RabbitMQ</summary>
 //-----------------------------------------------------------------------
 using System;
-using System.IO;
 using System.Security.Principal;
 using System.Threading.Tasks;
 using Csla.Core;
@@ -27,9 +26,11 @@ public class RabbitMqPortal : IDisposable
     /// Gets the URI for the data portal service.
     /// </summary>
     public string DataPortalUrl { get; private set; }
+
     private IConnection Connection;
     private IModel Channel;
     private string DataPortalQueueName;
+
     /// <summary>
     /// Gets or sets the timeout for network
     /// operations in seconds (default is 30 seconds).
@@ -91,16 +92,16 @@ public void StartListening()
       InitializeRabbitMQ();
       Channel.QueueDeclare(
         queue: DataPortalQueueName,
-        durable: false, 
-        exclusive: false, 
-        autoDelete: false, 
+        durable: false,
+        exclusive: false,
+        autoDelete: false,
         arguments: null);
 
       var consumer = new EventingBasicConsumer(Channel);
       consumer.Received += (model, ea) =>
       {
         Console.WriteLine($"Received {ea.BasicProperties.Type} for {ea.BasicProperties.CorrelationId} from {ea.BasicProperties.ReplyTo}");
-        InvokePortal(ea, ea.Body);
+        InvokePortal(ea, ea.Body.ToArray());
       };
       Console.WriteLine($"Listening on queue {DataPortalQueueName}");
       Channel.BasicConsume(queue: DataPortalQueueName, autoAck: true, consumer: consumer);
@@ -159,15 +160,19 @@ private async Task<HttpResponse> CallPortal(string operation, object request)
         case "create":
           result = await Create((CriteriaRequest)request).ConfigureAwait(false);
           break;
+
         case "fetch":
           result = await Fetch((CriteriaRequest)request).ConfigureAwait(false);
           break;
+
         case "update":
           result = await Update((UpdateRequest)request).ConfigureAwait(false);
           break;
+
         case "delete":
           result = await Delete((CriteriaRequest)request).ConfigureAwait(false);
           break;
+
         default:
           throw new InvalidOperationException(operation);
       }
@@ -367,7 +372,7 @@ private static object GetCriteria(byte[] criteriaData)
       return criteria;
     }
 
-    #endregion
+    #endregion Criteria
 
     #region Conversion methods
 
@@ -401,7 +406,7 @@ protected virtual HttpResponse ConvertResponse(HttpResponse response)
       return response;
     }
 
-    #endregion
+    #endregion Conversion methods
 
     /// <summary>
     /// Dispose this object.
@@ -413,4 +418,4 @@ public void Dispose()
       Connection?.Dispose();
     }
   }
-}
+}