Merge pull request #1224 from rabbitmq/rabbitmq-dotnet-client-1223-6.x

lukebakken · web-flow · commit ec8e55228ac7 · 2022-06-14T12:01:42.000-07:00
Better integrate max message size
diff --git a/.gitignore b/.gitignore
@@ -1,3 +1,5 @@
+_site/
+
 ###################
 ## Generated files
 ###################
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,26 @@
+## Changes Between 6.3.1 and 6.4.0
+
+This release adds the ability to specify a maximum message size when receiving data. The default
+values are:
+
+* RabbitMQ .NET client 7.0.0 and beyond: 128MiB
+* RabbitMQ .NET client 6.4.0 up to 7.0.0: no limit by default
+
+Receiving a frame that specifies a content larger than the limit will throw an execption. This is to
+help prevent situations as described in [this discussion](https://github.com/rabbitmq/rabbitmq-dotnet-client/discussions/1213).
+
+To set a limit, use the set `MaxMessageSize` on your `ConnectionFactory` before opening connections:
+
+```
+// This sets the limit to 512MiB
+var cf = new ConnectionFactory();
+cf.MaxMessageSize = 536870912;
+var conn = cf.CreateConnection()`
+```
+
+GitHub milestone: [`6.4.0`](https://github.com/rabbitmq/rabbitmq-dotnet-client/milestone/58?closed=1)
+Diff: [link](https://github.com/rabbitmq/rabbitmq-dotnet-client/compare/v6.3.1...v6.4.0)
+
 ## Changes Between 6.3.0 and 6.3.1
 
 GitHub milestone: [`6.3.1`](https://github.com/rabbitmq/rabbitmq-dotnet-client/milestone/57?closed=1)
diff --git a/projects/RabbitMQ.Client/client/api/AmqpTcpEndpoint.cs b/projects/RabbitMQ.Client/client/api/AmqpTcpEndpoint.cs
@@ -60,17 +60,32 @@ public class AmqpTcpEndpoint// : ICloneable
 
         private int _port;
 
+        private readonly uint _maxMessageSize;
+
         /// <summary>
         /// Creates a new instance of the <see cref="AmqpTcpEndpoint"/>.
         /// </summary>
         /// <param name="hostName">Hostname.</param>
         /// <param name="portOrMinusOne"> Port number. If the port number is -1, the default port number will be used.</param>
         /// <param name="ssl">Ssl option.</param>
-        public AmqpTcpEndpoint(string hostName, int portOrMinusOne, SslOption ssl)
+        /// <param name="maxMessageSize">Maximum message size from RabbitMQ. 0 means "unlimited"</param>
+        public AmqpTcpEndpoint(string hostName, int portOrMinusOne, SslOption ssl, uint maxMessageSize)
         {
             HostName = hostName;
             _port = portOrMinusOne;
             Ssl = ssl;
+            _maxMessageSize = maxMessageSize;
+        }
+
+        /// <summary>
+        /// Creates a new instance of the <see cref="AmqpTcpEndpoint"/>.
+        /// </summary>
+        /// <param name="hostName">Hostname.</param>
+        /// <param name="portOrMinusOne"> Port number. If the port number is -1, the default port number will be used.</param>
+        /// <param name="ssl">Ssl option.</param>
+        public AmqpTcpEndpoint(string hostName, int portOrMinusOne, SslOption ssl) :
+            this(hostName, portOrMinusOne, ssl, 0)
+        {
         }
 
         /// <summary>
@@ -116,7 +131,7 @@ public AmqpTcpEndpoint(Uri uri) : this(uri.Host, uri.Port)
         /// <returns>A copy with the same hostname, port, and TLS settings</returns>
         public object Clone()
         {
-            return new AmqpTcpEndpoint(HostName, _port, Ssl);
+            return new AmqpTcpEndpoint(HostName, _port, Ssl, _maxMessageSize);
         }
 
         /// <summary>
@@ -126,7 +141,7 @@ public object Clone()
         /// <returns>A copy with the provided hostname and port/TLS settings of this endpoint</returns>
         public AmqpTcpEndpoint CloneWithHostname(string hostname)
         {
-            return new AmqpTcpEndpoint(hostname, _port, Ssl);
+            return new AmqpTcpEndpoint(hostname, _port, Ssl, _maxMessageSize);
         }
 
         /// <summary>
@@ -176,9 +191,12 @@ public IProtocol Protocol
         public SslOption Ssl { get; set; }
 
         /// <summary>
-        /// Set the maximum size for a message in bytes. The default value is 0 (unlimited)
+        /// Get the maximum size for a message in bytes. The default value is 0 (unlimited)
         /// </summary>
-        public uint MaxMessageSize { get; set; }
+        public uint MaxMessageSize
+        {
+            get { return _maxMessageSize; }
+        }
 
         /// <summary>
         /// Construct an instance from a protocol and an address in "hostname:port" format.
diff --git a/projects/RabbitMQ.Client/client/api/ConnectionFactory.cs b/projects/RabbitMQ.Client/client/api/ConnectionFactory.cs
@@ -57,6 +57,7 @@ namespace RabbitMQ.Client
     ///     factory.VirtualHost = ConnectionFactory.DefaultVHost;
     ///     factory.HostName = hostName;
     ///     factory.Port     = AmqpTcpEndpoint.UseDefaultPort;
+    ///     factory.MaxMessageSize = 512 * 1024 * 1024;
     ///     //
     ///     IConnection conn = factory.CreateConnection();
     ///     //
@@ -103,6 +104,13 @@ public sealed class ConnectionFactory : ConnectionFactoryBase, IAsyncConnectionF
         /// </summary>
         public const uint DefaultFrameMax = 0;
 
+        /// <summary>
+        /// Default value for the maximum allowed message size, in bytes, from RabbitMQ.
+        /// Corresponds to the <code>rabbit.max_message_size</code> setting.
+        /// Note: the default is 0 which means "unlimited".
+        /// </summary>
+        public const uint DefaultMaxMessageSize = 0;
+
         /// <summary>
         /// Default value for desired heartbeat interval. Default is 60 seconds,
         /// TimeSpan.Zero means "heartbeats are disabled".
@@ -276,12 +284,13 @@ public ConnectionFactory()
         /// </summary>
         public AmqpTcpEndpoint Endpoint
         {
-            get { return new AmqpTcpEndpoint(HostName, Port, Ssl); }
+            get { return new AmqpTcpEndpoint(HostName, Port, Ssl, MaxMessageSize); }
             set
             {
                 Port = value.Port;
                 HostName = value.HostName;
                 Ssl = value.Ssl;
+                MaxMessageSize = value.MaxMessageSize;
             }
         }
 
@@ -325,6 +334,12 @@ public AmqpTcpEndpoint Endpoint
         /// </summary>
         public string VirtualHost { get; set; } = DefaultVHost;
 
+        /// <summary>
+        /// Maximum allowed message size, in bytes, from RabbitMQ.
+        /// Corresponds to the <code>rabbit.max_message_size</code> setting.
+        /// </summary>
+        public uint MaxMessageSize { get; set; } = DefaultMaxMessageSize;
+
         /// <summary>
         /// The uri to use for the connection.
         /// </summary>
diff --git a/projects/RabbitMQ.Client/client/exceptions/HardProtocolException.cs b/projects/RabbitMQ.Client/client/exceptions/HardProtocolException.cs
@@ -35,8 +35,20 @@ namespace RabbitMQ.Client.Exceptions
     ///requiring a connection.close.</summary>
     public abstract class HardProtocolException : ProtocolException
     {
+        protected readonly bool _canShutdownCleanly = true;
+
         protected HardProtocolException(string message) : base(message)
         {
         }
+
+        protected HardProtocolException(string message, bool canShutdownCleanly) : base(message)
+        {
+            _canShutdownCleanly= canShutdownCleanly;
+        }
+
+        public bool CanShutdownCleanly
+        {
+            get { return _canShutdownCleanly; }
+        }
     }
 }
diff --git a/projects/RabbitMQ.Client/client/exceptions/MalformedFrameException.cs b/projects/RabbitMQ.Client/client/exceptions/MalformedFrameException.cs
@@ -46,6 +46,11 @@ public MalformedFrameException(string message) : base(message)
         {
         }
 
+        public MalformedFrameException(string message, bool canShutdownCleanly) :
+            base(message, canShutdownCleanly)
+        {
+        }
+
         public override ushort ReplyCode
         {
             get { return Constants.FrameError; }
diff --git a/projects/RabbitMQ.Client/client/impl/Connection.cs b/projects/RabbitMQ.Client/client/impl/Connection.cs
@@ -479,7 +479,7 @@ public bool HardProtocolExceptionHandler(HardProtocolException hpe)
                             hpe.ShutdownReason.ReplyCode,
                             hpe.ShutdownReason.ReplyText));
                     }
-                    return true;
+                    return hpe.CanShutdownCleanly;
                 }
                 catch (IOException ioe)
                 {
diff --git a/projects/RabbitMQ.Client/client/impl/Frame.cs b/projects/RabbitMQ.Client/client/impl/Frame.cs
@@ -246,7 +246,8 @@ internal static InboundFrame ReadFrom(Stream reader, byte[] frameHeaderBuffer, A
             int payloadSize = NetworkOrderDeserializer.ReadInt32(new ReadOnlySpan<byte>(frameHeaderBuffer, 2, 4));
             if ((maxMessageSize > 0) && (payloadSize > maxMessageSize))
             {
-                throw new MalformedFrameException($"Frame payload size '{payloadSize}' exceeds maximum of '{maxMessageSize}' bytes");
+                string msg = $"Frame payload size '{payloadSize}' exceeds maximum of '{maxMessageSize}' bytes";
+                throw new MalformedFrameException(message: msg, canShutdownCleanly: false);
             }
 
             const int EndMarkerLength = 1;
diff --git a/projects/Unit/APIApproval.Approve.verified.txt b/projects/Unit/APIApproval.Approve.verified.txt
@@ -11,9 +11,10 @@ namespace RabbitMQ.Client
         public AmqpTcpEndpoint(string hostName, int portOrMinusOne = -1) { }
         public AmqpTcpEndpoint(System.Uri uri, RabbitMQ.Client.SslOption ssl) { }
         public AmqpTcpEndpoint(string hostName, int portOrMinusOne, RabbitMQ.Client.SslOption ssl) { }
+        public AmqpTcpEndpoint(string hostName, int portOrMinusOne, RabbitMQ.Client.SslOption ssl, uint maxMessageSize) { }
         public System.Net.Sockets.AddressFamily AddressFamily { get; set; }
         public string HostName { get; set; }
-        public uint MaxMessageSize { get; set; }
+        public uint MaxMessageSize { get; }
         public int Port { get; set; }
         public RabbitMQ.Client.IProtocol Protocol { get; }
         public RabbitMQ.Client.SslOption Ssl { get; set; }
@@ -72,6 +73,7 @@ namespace RabbitMQ.Client
     {
         public const ushort DefaultChannelMax = 2047;
         public const uint DefaultFrameMax = 0u;
+        public const uint DefaultMaxMessageSize = 0u;
         public const string DefaultPass = "guest";
         public const string DefaultUser = "guest";
         public const string DefaultVHost = "/";
@@ -91,6 +93,7 @@ namespace RabbitMQ.Client
         public System.Func<System.Collections.Generic.IEnumerable<RabbitMQ.Client.AmqpTcpEndpoint>, RabbitMQ.Client.IEndpointResolver> EndpointResolverFactory { get; set; }
         public System.TimeSpan HandshakeContinuationTimeout { get; set; }
         public string HostName { get; set; }
+        public uint MaxMessageSize { get; set; }
         public System.Buffers.ArrayPool<byte> MemoryPool { get; set; }
         public System.TimeSpan NetworkRecoveryInterval { get; set; }
         public string Password { get; set; }
@@ -728,11 +731,15 @@ namespace RabbitMQ.Client.Exceptions
     }
     public abstract class HardProtocolException : RabbitMQ.Client.Exceptions.ProtocolException
     {
+        protected readonly bool _canShutdownCleanly;
         protected HardProtocolException(string message) { }
+        protected HardProtocolException(string message, bool canShutdownCleanly) { }
+        public bool CanShutdownCleanly { get; }
     }
     public class MalformedFrameException : RabbitMQ.Client.Exceptions.HardProtocolException
     {
         public MalformedFrameException(string message) { }
+        public MalformedFrameException(string message, bool canShutdownCleanly) { }
         public override ushort ReplyCode { get; }
     }
     [System.Serializable]
diff --git a/projects/Unit/TestBasicPublish.cs b/projects/Unit/TestBasicPublish.cs
@@ -1,4 +1,5 @@
 using System;
+using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using NUnit.Framework;
@@ -101,5 +102,96 @@ public void CanNotModifyPayloadAfterPublish()
                 m.BasicCancel(tag);
             }
         }
+
+        [Test]
+        public void TestMaxMessageSize()
+        {
+            var re = new ManualResetEventSlim();
+            const ushort maxMsgSize = 1024;
+
+            int count = 0;
+            byte[] msg0 = Encoding.UTF8.GetBytes("hi");
+
+            var r = new System.Random();
+            byte[] msg1 = new byte[maxMsgSize * 2];
+            r.NextBytes(msg1);
+
+            var cf = new ConnectionFactory();
+            cf.AutomaticRecoveryEnabled = false;
+            cf.TopologyRecoveryEnabled = false;
+            cf.MaxMessageSize = maxMsgSize;
+
+            bool sawConnectionShutdown = false;
+            bool sawModelShutdown = false;
+            bool sawConsumerRegistered = false;
+            bool sawConsumerCancelled = false;
+
+            using (IConnection c = cf.CreateConnection())
+            {
+                c.ConnectionShutdown += (o, a) =>
+                {
+                    sawConnectionShutdown= true;
+                };
+
+                Assert.AreEqual(maxMsgSize, cf.MaxMessageSize);
+                Assert.AreEqual(maxMsgSize, cf.Endpoint.MaxMessageSize);
+                Assert.AreEqual(maxMsgSize, c.Endpoint.MaxMessageSize);
+
+                using (IModel m = c.CreateModel())
+                {
+                    m.ModelShutdown += (o, a) =>
+                    {
+                        sawModelShutdown= true;
+                    };
+
+                    m.CallbackException += (o, a) =>
+                    {
+                        Assert.Fail("Unexpected m.CallbackException");
+                    };
+
+                    QueueDeclareOk q = m.QueueDeclare();
+                    IBasicProperties bp = m.CreateBasicProperties();
+
+                    var consumer = new EventingBasicConsumer(m);
+
+                    consumer.Shutdown += (o, a) =>
+                    {
+                        re.Set();
+                    };
+
+                    consumer.Registered += (o, a) =>
+                    {
+                        sawConsumerRegistered = true;
+                    };
+
+                    consumer.Unregistered += (o, a) =>
+                    {
+                        Assert.Fail("Unexpected consumer.Unregistered");
+                    };
+
+                    consumer.ConsumerCancelled += (o, a) =>
+                    {
+                        sawConsumerCancelled = true;
+                    };
+
+                    consumer.Received += (o, a) =>
+                    {
+                        Interlocked.Increment(ref count);
+                    };
+
+                    string tag = m.BasicConsume(q.QueueName, true, consumer);
+
+                    m.BasicPublish("", q.QueueName, bp, msg0);
+                    m.BasicPublish("", q.QueueName, bp, msg1);
+                    Assert.IsTrue(re.Wait(TimeSpan.FromSeconds(5)));
+
+                    Assert.AreEqual(1, count);
+                    Assert.IsTrue(sawConnectionShutdown);
+                    Assert.IsTrue(sawModelShutdown);
+                    Assert.IsTrue(sawConsumerRegistered);
+                    Assert.IsTrue(sawConsumerCancelled);
+                }
+            }
+        }
     }
 }
diff --git a/projects/Unit/TestConnectionFactory.cs b/projects/Unit/TestConnectionFactory.cs

Original file line number	Diff line number	Diff line change
`@@ -1,3 +1,5 @@`
	`1`	`+_site/`
	`2`	`+`
`1`	`3`	`###################`
`2`	`4`	`## Generated files`
`3`	`5`	`###################`
Original file line number	Diff line number	Diff line change
`@@ -35,8 +35,20 @@ namespace RabbitMQ.Client.Exceptions`
`35`	`35`	`///requiring a connection.close.</summary>`
`36`	`36`	`public abstract class HardProtocolException : ProtocolException`
`37`	`37`	`{`
	`38`	`+ protected readonly bool _canShutdownCleanly = true;`
	`39`	`+`
`38`	`40`	`protected HardProtocolException(string message) : base(message)`
`39`	`41`	`{`
`40`	`42`	`}`
	`43`	`+`
	`44`	`+ protected HardProtocolException(string message, bool canShutdownCleanly) : base(message)`
	`45`	`+ {`
	`46`	`+ _canShutdownCleanly= canShutdownCleanly;`
	`47`	`+ }`
	`48`	`+`
	`49`	`+ public bool CanShutdownCleanly`
	`50`	`+ {`
	`51`	`+ get { return _canShutdownCleanly; }`
	`52`	`+ }`
`41`	`53`	`}`
`42`	`54`	`}`
Original file line number	Diff line number	Diff line change
`@@ -46,6 +46,11 @@ public MalformedFrameException(string message) : base(message)`
`46`	`46`	`{`
`47`	`47`	`}`
`48`	`48`
	`49`	`+ public MalformedFrameException(string message, bool canShutdownCleanly) :`
	`50`	`+ base(message, canShutdownCleanly)`
	`51`	`+ {`
	`52`	`+ }`
	`53`	`+`
`49`	`54`	`public override ushort ReplyCode`
`50`	`55`	`{`
`51`	`56`	`get { return Constants.FrameError; }`
Original file line number	Diff line number	Diff line change
`@@ -479,7 +479,7 @@ public bool HardProtocolExceptionHandler(HardProtocolException hpe)`
`479`	`479`	`hpe.ShutdownReason.ReplyCode,`
`480`	`480`	`hpe.ShutdownReason.ReplyText));`
`481`	`481`	`}`
`482`		`- return true;`
	`482`	`+ return hpe.CanShutdownCleanly;`
`483`	`483`	`}`
`484`	`484`	`catch (IOException ioe)`
`485`	`485`	`{`
Original file line number	Diff line number	Diff line change
`@@ -246,7 +246,8 @@ internal static InboundFrame ReadFrom(Stream reader, byte[] frameHeaderBuffer, A`
`246`	`246`	`int payloadSize = NetworkOrderDeserializer.ReadInt32(new ReadOnlySpan<byte>(frameHeaderBuffer, 2, 4));`
`247`	`247`	`if ((maxMessageSize > 0) && (payloadSize > maxMessageSize))`
`248`	`248`	`{`
`249`		`- throw new MalformedFrameException($"Frame payload size '{payloadSize}' exceeds maximum of '{maxMessageSize}' bytes");`
	`249`	`+ string msg = $"Frame payload size '{payloadSize}' exceeds maximum of '{maxMessageSize}' bytes";`
	`250`	`+ throw new MalformedFrameException(message: msg, canShutdownCleanly: false);`
`250`	`251`	`}`
`251`	`252`
`252`	`253`	`const int EndMarkerLength = 1;`