Add ability to specify maximum message size

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

@@ -175,6 +175,11 @@ public IProtocol Protocol
         /// </summary>
         public SslOption Ssl { get; set; }
 
+        /// <summary>
+        /// Set the maximum size for a message in bytes. The default value is 0 (unlimited)
+        /// </summary>
+        public uint MaxMessageSize { get; set; }
+
         /// <summary>
         /// Construct an instance from a protocol and an address in "hostname:port" format.
         /// </summary>

projects/RabbitMQ.Client/client/impl/Frame.cs

@@ -205,7 +205,7 @@ private static void ProcessProtocolHeader(Stream reader)
             }
         }
 
-        internal static InboundFrame ReadFrom(Stream reader, byte[] frameHeaderBuffer, ArrayPool<byte> pool)
+        internal static InboundFrame ReadFrom(Stream reader, byte[] frameHeaderBuffer, ArrayPool<byte> pool, uint maxMessageSize)
         {
             int type = default;
             try
@@ -239,7 +239,11 @@ internal static InboundFrame ReadFrom(Stream reader, byte[] frameHeaderBuffer, A
 
             reader.Read(frameHeaderBuffer, 0, frameHeaderBuffer.Length);
             int channel = NetworkOrderDeserializer.ReadUInt16(new ReadOnlySpan<byte>(frameHeaderBuffer));
-            int payloadSize = NetworkOrderDeserializer.ReadInt32(new ReadOnlySpan<byte>(frameHeaderBuffer, 2, 4)); // FIXME - throw exn on unreasonable value
+            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");
+            }
 
             const int EndMarkerLength = 1;
             // Is returned by InboundFrame.Dispose in Connection.MainLoopIteration

projects/RabbitMQ.Client/client/impl/SocketFrameHandler.cs

@@ -62,6 +62,7 @@ public static async Task TimeoutAfter(this Task task, TimeSpan timeout)
 
     class SocketFrameHandler : IFrameHandler
     {
+        private readonly AmqpTcpEndpoint _endpoint;
         // Socket poll timeout in ms. If the socket does not
         // become writeable in this amount of time, we throw
         // an exception.
@@ -82,7 +83,7 @@ public SocketFrameHandler(AmqpTcpEndpoint endpoint,
             Func<AddressFamily, ITcpClient> socketFactory,
             TimeSpan connectionTimeout, TimeSpan readTimeout, TimeSpan writeTimeout)
         {
-            Endpoint = endpoint;
+            _endpoint = endpoint;
             _frameHeaderBuffer = new byte[6];
             var channel = Channel.CreateUnbounded<ReadOnlyMemory<byte>>(
                 new UnboundedChannelOptions
@@ -135,7 +136,11 @@ public SocketFrameHandler(AmqpTcpEndpoint endpoint,
             WriteTimeout = writeTimeout;
             _writerTask = Task.Run(WriteLoop, CancellationToken.None);
         }
-        public AmqpTcpEndpoint Endpoint { get; set; }
+
+        public AmqpTcpEndpoint Endpoint
+        {
+            get { return _endpoint; }
+        }
 
         internal ArrayPool<byte> MemoryPool
         {
@@ -229,7 +234,7 @@ public void Close()
 
         public InboundFrame ReadFrame()
         {
-            return InboundFrame.ReadFrom(_reader, _frameHeaderBuffer, MemoryPool);
+            return InboundFrame.ReadFrom(_reader, _frameHeaderBuffer, MemoryPool, _endpoint.MaxMessageSize);
         }
 
         public void SendHeader()

projects/Unit/APIApproval.Approve.verified.txt

@@ -13,6 +13,7 @@ namespace RabbitMQ.Client
         public AmqpTcpEndpoint(string hostName, int portOrMinusOne, RabbitMQ.Client.SslOption ssl) { }
         public System.Net.Sockets.AddressFamily AddressFamily { get; set; }
         public string HostName { get; set; }
+        public uint MaxMessageSize { get; set; }
         public int Port { get; set; }
         public RabbitMQ.Client.IProtocol Protocol { get; }
         public RabbitMQ.Client.SslOption Ssl { get; set; }