Merge pull request #1380 from rabbitmq/fix-1378

Fix #1378

Author: michaelklishin

projects/RabbitMQ.Client/client/api/AmqpTcpEndpoint.cs

@@ -76,7 +76,7 @@ public AmqpTcpEndpoint(string hostName, int portOrMinusOne, SslOption ssl, uint
             HostName = hostName;
             _port = portOrMinusOne;
             Ssl = ssl;
-            _maxMessageSize = maxMessageSize;
+            _maxMessageSize = Math.Min(maxMessageSize, ConnectionFactory.DefaultMaxMessageSize);
         }
 
         /// <summary>

projects/RabbitMQ.Client/client/api/ConnectionFactory.cs

@@ -109,7 +109,7 @@ public sealed class ConnectionFactory : ConnectionFactoryBase, IConnectionFactor
         /// Corresponds to the <code>rabbit.max_message_size</code> setting.
         /// Note: the default is 0 which means "unlimited".
         /// </summary>
-        public const uint DefaultMaxMessageSize = 134217728;
+        public const uint DefaultMaxMessageSize = 536870912;
 
         /// <summary>
         /// Default value for desired heartbeat interval. Default is 60 seconds,
@@ -454,7 +454,7 @@ public IConnection CreateConnection(IList<string> hostnames)
         /// </exception>
         public IConnection CreateConnection(IList<string> hostnames, string clientProvidedName)
         {
-            IEnumerable<AmqpTcpEndpoint> endpoints = hostnames.Select(h => new AmqpTcpEndpoint(h, Port, Ssl));
+            IEnumerable<AmqpTcpEndpoint> endpoints = hostnames.Select(h => new AmqpTcpEndpoint(h, Port, Ssl, MaxMessageSize));
             return CreateConnection(EndpointResolverFactory(endpoints), clientProvidedName);
         }
 

projects/Unit/APIApproval.Approve.verified.txt

@@ -154,7 +154,7 @@ namespace RabbitMQ.Client
     {
         public const ushort DefaultChannelMax = 2047;
         public const uint DefaultFrameMax = 0u;
-        public const uint DefaultMaxMessageSize = 134217728u;
+        public const uint DefaultMaxMessageSize = 536870912u;
         public const string DefaultPass = "guest";
         public const string DefaultUser = "guest";
         public const string DefaultVHost = "/";

projects/Unit/TestBasicPublish.cs

@@ -110,22 +110,22 @@ public void CanNotModifyPayloadAfterPublish()
                 QueueDeclareOk q = m.QueueDeclare();
                 byte[] sendBody = new byte[1000];
                 var consumer = new EventingBasicConsumer(m);
-                var are = new AutoResetEvent(false);
+                var receivedMessage = new AutoResetEvent(false);
                 bool modified = true;
                 consumer.Received += (o, a) =>
                 {
                     if (a.Body.Span.IndexOf((byte)1) < 0)
                     {
                         modified = false;
                     }
-                    are.Set();
+                    receivedMessage.Set();
                 };
                 string tag = m.BasicConsume(q.QueueName, true, consumer);
 
                 m.BasicPublish("", q.QueueName, sendBody);
                 sendBody.AsSpan().Fill(1);
 
-                Assert.True(are.WaitOne(2000));
+                Assert.True(receivedMessage.WaitOne(5000));
                 Assert.False(modified, "Payload was modified after the return of BasicPublish");
 
                 m.BasicCancel(tag);

projects/Unit/TestConnectionFactory.cs

@@ -269,5 +269,51 @@ public void TestCreateConnectioUsesValidEndpointWhenMultipleSupplied()
             var ep = new AmqpTcpEndpoint("localhost");
             using (IConnection conn = cf.CreateConnection(new List<AmqpTcpEndpoint> { invalidEp, ep })) { };
         }
+
+        [Fact]
+        public void TestCreateConnectionUsesConfiguredMaxMessageSize()
+        {
+            var cf = new ConnectionFactory();
+            cf.MaxMessageSize = 1500;
+            using (IConnection conn = cf.CreateConnection())
+            {
+                Assert.Equal(cf.MaxMessageSize, conn.Endpoint.MaxMessageSize);
+            };
+        }
+        [Fact]
+        public void TestCreateConnectionWithAmqpEndpointListUsesAmqpTcpEndpointMaxMessageSize()
+        {
+            var cf = new ConnectionFactory();
+            cf.MaxMessageSize = 1500;
+            var ep = new AmqpTcpEndpoint("localhost");
+            Assert.Equal(ConnectionFactory.DefaultMaxMessageSize, ep.MaxMessageSize);
+            using (IConnection conn = cf.CreateConnection(new List<AmqpTcpEndpoint> { ep }))
+            {
+                Assert.Equal(ConnectionFactory.DefaultMaxMessageSize, conn.Endpoint.MaxMessageSize);
+            };
+        }
+
+        [Fact]
+        public void TestCreateConnectionWithAmqpEndpointResolverUsesAmqpTcpEndpointMaxMessageSize()
+        {
+            var cf = new ConnectionFactory();
+            cf.MaxMessageSize = 1500;
+            var ep = new AmqpTcpEndpoint("localhost", -1, new SslOption(), 1200);
+            using (IConnection conn = cf.CreateConnection(new List<AmqpTcpEndpoint> { ep }))
+            {
+                Assert.Equal(ep.MaxMessageSize, conn.Endpoint.MaxMessageSize);
+            };
+        }
+
+        [Fact]
+        public void TestCreateConnectionWithHostnameListUsesConnectionFactoryMaxMessageSize()
+        {
+            var cf = new ConnectionFactory();
+            cf.MaxMessageSize = 1500;
+            using (IConnection conn = cf.CreateConnection(new List<string> { "localhost" }))
+            {
+                Assert.Equal(cf.MaxMessageSize, conn.Endpoint.MaxMessageSize);
+            };
+        }
     }
 }