Skip to content

Commit e5e960b

Browse files
michaelklishinmichaelklishin
authored
Merge pull request #1226 from rabbitmq/rabbitmq-dotnet-client-1225
Merge pull request #1224 from rabbitmq/rabbitmq-dotnet-client-1223-6.x
2 parents fd95377 + cf2b303 commit e5e960b

12 files changed

+210
-27
lines changed

‎.gitignore

+2
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
1+
_site/
2+
13
###################
24
## Generated files
35
###################

‎CHANGELOG.md

+23
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,26 @@
1+
## Changes Between 6.3.1 and 6.4.0
2+
3+
This release adds the ability to specify a maximum message size when receiving data. The default
4+
values are:
5+
6+
* RabbitMQ .NET client 7.0.0 and beyond: 128MiB
7+
* RabbitMQ .NET client 6.4.0 up to 7.0.0: no limit by default
8+
9+
Receiving a frame that specifies a content larger than the limit will throw an execption. This is to
10+
help prevent situations as described in [this discussion](https://github.com/rabbitmq/rabbitmq-dotnet-client/discussions/1213).
11+
12+
To set a limit, use the set `MaxMessageSize` on your `ConnectionFactory` before opening connections:
13+
14+
```
15+
// This sets the limit to 512MiB
16+
var cf = new ConnectionFactory();
17+
cf.MaxMessageSize = 536870912;
18+
var conn = cf.CreateConnection()`
19+
```
20+
21+
GitHub milestone: [`6.4.0`](https://github.com/rabbitmq/rabbitmq-dotnet-client/milestone/58?closed=1)
22+
Diff: [link](https://github.com/rabbitmq/rabbitmq-dotnet-client/compare/v6.3.1...v6.4.0)
23+
124
## Changes Between 6.3.0 and 6.3.1
225

326
GitHub milestone: [`6.3.1`](https://github.com/rabbitmq/rabbitmq-dotnet-client/milestone/57?closed=1)

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

+19-16
Original file line numberDiff line numberDiff line change
@@ -61,19 +61,33 @@ public class AmqpTcpEndpoint// : ICloneable
6161
public const int UseDefaultPort = -1;
6262

6363
private int _port;
64-
private uint _maxMessageSize = Constants.DefaultMaxMessageSizeInBytes;
64+
65+
private readonly uint _maxMessageSize;
6566

6667
/// <summary>
6768
/// Creates a new instance of the <see cref="AmqpTcpEndpoint"/>.
6869
/// </summary>
6970
/// <param name="hostName">Hostname.</param>
7071
/// <param name="portOrMinusOne"> Port number. If the port number is -1, the default port number will be used.</param>
7172
/// <param name="ssl">Ssl option.</param>
72-
public AmqpTcpEndpoint(string hostName, int portOrMinusOne, SslOption ssl)
73+
/// <param name="maxMessageSize">Maximum message size from RabbitMQ. 0 means "unlimited"</param>
74+
public AmqpTcpEndpoint(string hostName, int portOrMinusOne, SslOption ssl, uint maxMessageSize)
7375
{
7476
HostName = hostName;
7577
_port = portOrMinusOne;
7678
Ssl = ssl;
79+
_maxMessageSize = maxMessageSize;
80+
}
81+
82+
/// <summary>
83+
/// Creates a new instance of the <see cref="AmqpTcpEndpoint"/>.
84+
/// </summary>
85+
/// <param name="hostName">Hostname.</param>
86+
/// <param name="portOrMinusOne"> Port number. If the port number is -1, the default port number will be used.</param>
87+
/// <param name="ssl">Ssl option.</param>
88+
public AmqpTcpEndpoint(string hostName, int portOrMinusOne, SslOption ssl) :
89+
this(hostName, portOrMinusOne, ssl, ConnectionFactory.DefaultMaxMessageSize)
90+
{
7791
}
7892

7993
/// <summary>
@@ -119,7 +133,7 @@ public AmqpTcpEndpoint(Uri uri) : this(uri.Host, uri.Port)
119133
/// <returns>A copy with the same hostname, port, and TLS settings</returns>
120134
public object Clone()
121135
{
122-
return new AmqpTcpEndpoint(HostName, _port, Ssl);
136+
return new AmqpTcpEndpoint(HostName, _port, Ssl, _maxMessageSize);
123137
}
124138

125139
/// <summary>
@@ -129,7 +143,7 @@ public object Clone()
129143
/// <returns>A copy with the provided hostname and port/TLS settings of this endpoint</returns>
130144
public AmqpTcpEndpoint CloneWithHostname(string hostname)
131145
{
132-
return new AmqpTcpEndpoint(hostname, _port, Ssl);
146+
return new AmqpTcpEndpoint(hostname, _port, Ssl, _maxMessageSize);
133147
}
134148

135149
/// <summary>
@@ -179,22 +193,11 @@ public IProtocol Protocol
179193
public SslOption Ssl { get; set; }
180194

181195
/// <summary>
182-
/// Set the maximum size for a message in bytes. Setting it to 0 reverts to the default of 128MiB
196+
/// Get the maximum size for a message in bytes. The default value is 128MiB to match RabbitMQ's default
183197
/// </summary>
184198
public uint MaxMessageSize
185199
{
186200
get { return _maxMessageSize; }
187-
set
188-
{
189-
if (value == default(uint))
190-
{
191-
_maxMessageSize = Constants.DefaultMaxMessageSizeInBytes;
192-
}
193-
else
194-
{
195-
_maxMessageSize = value;
196-
}
197-
}
198201
}
199202

200203
/// <summary>

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

+16-1
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,7 @@ namespace RabbitMQ.Client
5757
/// factory.VirtualHost = ConnectionFactory.DefaultVHost;
5858
/// factory.HostName = hostName;
5959
/// factory.Port = AmqpTcpEndpoint.UseDefaultPort;
60+
/// factory.MaxMessageSize = 512 * 1024 * 1024;
6061
/// //
6162
/// IConnection conn = factory.CreateConnection();
6263
/// //
@@ -103,6 +104,13 @@ public sealed class ConnectionFactory : ConnectionFactoryBase, IConnectionFactor
103104
/// </summary>
104105
public const uint DefaultFrameMax = 0;
105106

107+
/// <summary>
108+
/// Default value for the maximum allowed message size, in bytes, from RabbitMQ.
109+
/// Corresponds to the <code>rabbit.max_message_size</code> setting.
110+
/// Note: the default is 0 which means "unlimited".
111+
/// </summary>
112+
public const uint DefaultMaxMessageSize = 134217728;
113+
106114
/// <summary>
107115
/// Default value for desired heartbeat interval. Default is 60 seconds,
108116
/// TimeSpan.Zero means "heartbeats are disabled".
@@ -264,12 +272,13 @@ public ConnectionFactory()
264272
/// </summary>
265273
public AmqpTcpEndpoint Endpoint
266274
{
267-
get { return new AmqpTcpEndpoint(HostName, Port, Ssl); }
275+
get { return new AmqpTcpEndpoint(HostName, Port, Ssl, MaxMessageSize); }
268276
set
269277
{
270278
Port = value.Port;
271279
HostName = value.HostName;
272280
Ssl = value.Ssl;
281+
MaxMessageSize = value.MaxMessageSize;
273282
}
274283
}
275284

@@ -317,6 +326,12 @@ public AmqpTcpEndpoint Endpoint
317326
/// </summary>
318327
public string VirtualHost { get; set; } = DefaultVHost;
319328

329+
/// <summary>
330+
/// Maximum allowed message size, in bytes, from RabbitMQ.
331+
/// Corresponds to the <code>rabbit.max_message_size</code> setting.
332+
/// </summary>
333+
public uint MaxMessageSize { get; set; } = DefaultMaxMessageSize;
334+
320335
/// <summary>
321336
/// The uri to use for the connection.
322337
/// </summary>

‎projects/RabbitMQ.Client/client/exceptions/HardProtocolException.cs

+12
Original file line numberDiff line numberDiff line change
@@ -35,8 +35,20 @@ namespace RabbitMQ.Client.Exceptions
3535
///requiring a connection.close.</summary>
3636
public abstract class HardProtocolException : ProtocolException
3737
{
38+
protected readonly bool _canShutdownCleanly = true;
39+
3840
protected HardProtocolException(string message) : base(message)
3941
{
4042
}
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+
}
4153
}
4254
}

‎projects/RabbitMQ.Client/client/exceptions/MalformedFrameException.cs

+5
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,11 @@ public MalformedFrameException(string message) : base(message)
4444
{
4545
}
4646

47+
public MalformedFrameException(string message, bool canShutdownCleanly) :
48+
base(message, canShutdownCleanly)
49+
{
50+
}
51+
4752
public override ushort ReplyCode
4853
{
4954
get { return Constants.FrameError; }

‎projects/RabbitMQ.Client/client/framing/Constants.cs

-3
Original file line numberDiff line numberDiff line change
@@ -81,8 +81,5 @@ public static class Constants
8181
public const int NotImplemented = 540;
8282
///<summary>(= 541)</summary>
8383
public const int InternalError = 541;
84-
85-
///<summary>(= 134217728)</summary>
86-
public const uint DefaultMaxMessageSizeInBytes = 134217728;
8784
}
8885
}

‎projects/RabbitMQ.Client/client/impl/Connection.Receive.cs

+4-1
Original file line numberDiff line numberDiff line change
@@ -149,7 +149,10 @@ private void HardProtocolExceptionHandler(HardProtocolException hpe)
149149
{
150150
var cmd = new ConnectionClose(hpe.ShutdownReason.ReplyCode, hpe.ShutdownReason.ReplyText, 0, 0);
151151
_session0.Transmit(ref cmd);
152-
ClosingLoop();
152+
if (hpe.CanShutdownCleanly)
153+
{
154+
ClosingLoop();
155+
}
153156
}
154157
catch (IOException ioe)
155158
{

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

+3-2
Original file line numberDiff line numberDiff line change
@@ -278,9 +278,10 @@ internal static InboundFrame ReadFrom(Stream reader, byte[] frameHeaderBuffer, u
278278
var frameHeaderSpan = new ReadOnlySpan<byte>(frameHeaderBuffer, 1, 6);
279279
int channel = NetworkOrderDeserializer.ReadUInt16(frameHeaderSpan);
280280
int payloadSize = NetworkOrderDeserializer.ReadInt32(frameHeaderSpan.Slice(2, 4));
281-
if (payloadSize > maxMessageSize)
281+
if ((maxMessageSize > 0) && (payloadSize > maxMessageSize))
282282
{
283-
throw new MalformedFrameException($"Frame payload size '{payloadSize}' exceeds maximum of '{maxMessageSize}' bytes");
283+
string msg = $"Frame payload size '{payloadSize}' exceeds maximum of '{maxMessageSize}' bytes";
284+
throw new MalformedFrameException(message: msg, canShutdownCleanly: false);
284285
}
285286

286287
const int EndMarkerLength = 1;

‎projects/Unit/APIApproval.Approve.verified.txt

+8-2
Original file line numberDiff line numberDiff line change
@@ -11,9 +11,10 @@ namespace RabbitMQ.Client
1111
public AmqpTcpEndpoint(string hostName, int portOrMinusOne = -1) { }
1212
public AmqpTcpEndpoint(System.Uri uri, RabbitMQ.Client.SslOption ssl) { }
1313
public AmqpTcpEndpoint(string hostName, int portOrMinusOne, RabbitMQ.Client.SslOption ssl) { }
14+
public AmqpTcpEndpoint(string hostName, int portOrMinusOne, RabbitMQ.Client.SslOption ssl, uint maxMessageSize) { }
1415
public System.Net.Sockets.AddressFamily AddressFamily { get; set; }
1516
public string HostName { get; set; }
16-
public uint MaxMessageSize { get; set; }
17+
public uint MaxMessageSize { get; }
1718
public int Port { get; set; }
1819
public RabbitMQ.Client.IProtocol Protocol { get; }
1920
public RabbitMQ.Client.SslOption Ssl { get; set; }
@@ -151,6 +152,7 @@ namespace RabbitMQ.Client
151152
{
152153
public const ushort DefaultChannelMax = 2047;
153154
public const uint DefaultFrameMax = 0u;
155+
public const uint DefaultMaxMessageSize = 134217728u;
154156
public const string DefaultPass = "guest";
155157
public const string DefaultUser = "guest";
156158
public const string DefaultVHost = "/";
@@ -170,6 +172,7 @@ namespace RabbitMQ.Client
170172
public System.Func<System.Collections.Generic.IEnumerable<RabbitMQ.Client.AmqpTcpEndpoint>, RabbitMQ.Client.IEndpointResolver> EndpointResolverFactory { get; set; }
171173
public System.TimeSpan HandshakeContinuationTimeout { get; set; }
172174
public string HostName { get; set; }
175+
public uint MaxMessageSize { get; set; }
173176
public System.TimeSpan NetworkRecoveryInterval { get; set; }
174177
public string Password { get; set; }
175178
public int Port { get; set; }
@@ -208,7 +211,6 @@ namespace RabbitMQ.Client
208211
public const int CommandInvalid = 503;
209212
public const int ConnectionForced = 320;
210213
public const int ContentTooLarge = 311;
211-
public const uint DefaultMaxMessageSizeInBytes = 134217728u;
212214
public const int FrameBody = 3;
213215
public const int FrameEnd = 206;
214216
public const int FrameError = 501;
@@ -825,11 +827,15 @@ namespace RabbitMQ.Client.Exceptions
825827
}
826828
public abstract class HardProtocolException : RabbitMQ.Client.Exceptions.ProtocolException
827829
{
830+
protected readonly bool _canShutdownCleanly;
828831
protected HardProtocolException(string message) { }
832+
protected HardProtocolException(string message, bool canShutdownCleanly) { }
833+
public bool CanShutdownCleanly { get; }
829834
}
830835
public class MalformedFrameException : RabbitMQ.Client.Exceptions.HardProtocolException
831836
{
832837
public MalformedFrameException(string message) { }
838+
public MalformedFrameException(string message, bool canShutdownCleanly) { }
833839
public override ushort ReplyCode { get; }
834840
}
835841
[System.Serializable]

‎projects/Unit/TestBasicPublish.cs

+92-1
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,11 @@
11
using System;
2+
using System.Text;
23
using System.Threading;
34
using System.Threading.Tasks;
45

56
using RabbitMQ.Client.Events;
6-
using RabbitMQ.Client.Framing;
77
using Xunit;
8+
using Xunit.Sdk;
89

910
namespace RabbitMQ.Client.Unit
1011
{
@@ -130,5 +131,95 @@ public void CanNotModifyPayloadAfterPublish()
130131
m.BasicCancel(tag);
131132
}
132133
}
134+
135+
[Fact]
136+
public void TestMaxMessageSize()
137+
{
138+
var re = new ManualResetEventSlim();
139+
const ushort maxMsgSize = 1024;
140+
141+
int count = 0;
142+
byte[] msg0 = Encoding.UTF8.GetBytes("hi");
143+
144+
var r = new System.Random();
145+
byte[] msg1 = new byte[maxMsgSize * 2];
146+
r.NextBytes(msg1);
147+
148+
var cf = new ConnectionFactory();
149+
cf.AutomaticRecoveryEnabled = false;
150+
cf.TopologyRecoveryEnabled = false;
151+
cf.MaxMessageSize = maxMsgSize;
152+
153+
bool sawConnectionShutdown = false;
154+
bool sawModelShutdown = false;
155+
bool sawConsumerRegistered = false;
156+
bool sawConsumerCancelled = false;
157+
158+
using (IConnection c = cf.CreateConnection())
159+
{
160+
c.ConnectionShutdown += (o, a) =>
161+
{
162+
sawConnectionShutdown = true;
163+
};
164+
165+
Assert.Equal(maxMsgSize, cf.MaxMessageSize);
166+
Assert.Equal(maxMsgSize, cf.Endpoint.MaxMessageSize);
167+
Assert.Equal(maxMsgSize, c.Endpoint.MaxMessageSize);
168+
169+
using (IModel m = c.CreateModel())
170+
{
171+
m.ModelShutdown += (o, a) =>
172+
{
173+
sawModelShutdown = true;
174+
};
175+
176+
m.CallbackException += (o, a) =>
177+
{
178+
throw new XunitException("Unexpected m.CallbackException");
179+
};
180+
181+
QueueDeclareOk q = m.QueueDeclare();
182+
183+
var consumer = new EventingBasicConsumer(m);
184+
185+
consumer.Shutdown += (o, a) =>
186+
{
187+
re.Set();
188+
};
189+
190+
consumer.Registered += (o, a) =>
191+
{
192+
sawConsumerRegistered = true;
193+
};
194+
195+
consumer.Unregistered += (o, a) =>
196+
{
197+
throw new XunitException("Unexpected consumer.Unregistered");
198+
};
199+
200+
consumer.ConsumerCancelled += (o, a) =>
201+
{
202+
sawConsumerCancelled = true;
203+
};
204+
205+
consumer.Received += (o, a) =>
206+
{
207+
Interlocked.Increment(ref count);
208+
};
209+
210+
string tag = m.BasicConsume(q.QueueName, true, consumer);
211+
212+
m.BasicPublish("", q.QueueName, msg0);
213+
m.BasicPublish("", q.QueueName, msg1);
214+
Assert.True(re.Wait(TimeSpan.FromSeconds(5)));
215+
216+
Assert.Equal(1, count);
217+
Assert.True(sawConnectionShutdown);
218+
Assert.True(sawModelShutdown);
219+
Assert.True(sawConsumerRegistered);
220+
Assert.True(sawConsumerCancelled);
221+
}
222+
}
223+
}
133224
}
134225
}

0 commit comments

Comments
 (0)