Skip to content

Commit ee73a5c

Browse files
ngallegosDharmishtha Patelbartekwasielak
authored
Add support for SQS messages up to 1MB (#2982)
Co-authored-by: Dharmishtha Patel <dharmishthapatel@Dharmishthas-MacBook-Pro.local> Co-authored-by: Bartek Wasielak <bartek.wasielak@particular.net>
1 parent 7a8c505 commit ee73a5c

17 files changed

Lines changed: 85 additions & 50 deletions

‎src/NServiceBus.Transport.SQS.AcceptanceTests/Sending/When_sending_not_wrapped_message.cs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,13 +5,14 @@
55
using AcceptanceTesting.Customization;
66
using EndpointTemplates;
77
using NUnit.Framework;
8+
using Transport.SQS;
89

910
public class When_sending_not_wrapped_message : NServiceBusAcceptanceTest
1011
{
1112
public static object[] Payload =
1213
{
1314
new object[] { new byte[4] },
14-
new object[] { new byte[500 * 1024] }
15+
new object[] { new byte[TransportConstraints.SqsMaximumMessageSize - TransportTestsConstraints.SqsHeadersBuffer] }
1516
};
1617

1718
[Test]

‎src/NServiceBus.Transport.SQS.AcceptanceTests/Sending/When_using_large_message_with_customer_provided_aes.cs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ namespace NServiceBus.AcceptanceTests.Sending;
88
using Amazon.S3.Model;
99
using EndpointTemplates;
1010
using NUnit.Framework;
11+
using Transport.SQS;
1112
using Transport.SQS.Tests;
1213

1314
public class When_using_large_message_with_customer_provided_aes : NServiceBusAcceptanceTest
@@ -43,7 +44,7 @@ public async Task Should_receive_message()
4344
});
4445
}
4546

46-
const int PayloadSize = 150 * 1024;
47+
const int PayloadSize = TransportConstraints.SqsMaximumMessageSize + 1;
4748

4849
static string BucketName;
4950
static string Base64Key;

‎src/NServiceBus.Transport.SQS.AcceptanceTests/Sending/When_using_large_message_with_kms_encrypted_bucket.cs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
using AcceptanceTesting;
55
using EndpointTemplates;
66
using NUnit.Framework;
7+
using Transport.SQS;
78
using Transport.SQS.Tests;
89

910
public class When_using_large_message_with_kms_encrypted_bucket : NServiceBusAcceptanceTest
@@ -28,7 +29,7 @@ public async Task Should_receive_message()
2829
Assert.DoesNotThrowAsync(async () => await s3Client.GetObjectAsync(BucketName, $"{ConfigureEndpointSqsTransport.S3Prefix}/{context.MessageId}"));
2930
}
3031

31-
const int PayloadSize = 150 * 1024;
32+
const int PayloadSize = TransportConstraints.SqsMaximumMessageSize + 1;
3233

3334
static string BucketName;
3435

‎src/NServiceBus.Transport.SQS.AcceptanceTests/Sending/When_using_large_message_with_serverside_aes.cs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ namespace NServiceBus.AcceptanceTests.Sending;
55
using Amazon.S3;
66
using EndpointTemplates;
77
using NUnit.Framework;
8+
using Transport.SQS;
89
using Transport.SQS.Tests;
910

1011
public class When_using_large_message_with_serverside_aes : NServiceBusAcceptanceTest
@@ -34,7 +35,7 @@ public async Task Should_receive_message()
3435
});
3536
}
3637

37-
const int PayloadSize = 150 * 1024;
38+
const int PayloadSize = TransportConstraints.SqsMaximumMessageSize + 1;
3839

3940
static string BucketName;
4041

‎src/NServiceBus.Transport.SQS.AcceptanceTests/Sending/When_using_large_message_with_unencrypted_bucket.cs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
using AcceptanceTesting;
55
using EndpointTemplates;
66
using NUnit.Framework;
7+
using Transport.SQS;
78
using Transport.SQS.Tests;
89

910
public class When_using_large_message_with_unencrypted_bucket : NServiceBusAcceptanceTest
@@ -28,7 +29,7 @@ public async Task Should_receive_message()
2829
Assert.DoesNotThrowAsync(async () => await s3Client.GetObjectAsync(ConfigureEndpointSqsTransport.S3BucketName, $"{ConfigureEndpointSqsTransport.S3Prefix}/{context.MessageId}"));
2930
}
3031

31-
const int PayloadSize = 150 * 1024;
32+
const int PayloadSize = TransportConstraints.SqsMaximumMessageSize + 1;
3233

3334
public class Context : ScenarioContext
3435
{

‎src/NServiceBus.Transport.SQS.AcceptanceTests/Sending/When_using_oversized_message_without_bucket_configured.cs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
using AcceptanceTesting;
55
using EndpointTemplates;
66
using NUnit.Framework;
7+
using Transport.SQS;
78

89
public class When_using_oversized_message_without_bucket_configured : NServiceBusAcceptanceTest
910
{
@@ -24,7 +25,7 @@ await Scenario.Define<Context>()
2425
Assert.That(exception.Message, Is.EqualTo("Cannot send large message because no S3 bucket was configured. Add an S3 bucket name to your configuration."));
2526
}
2627

27-
const int PayloadSize = 150 * 1024;
28+
const int PayloadSize = TransportConstraints.SqsMaximumMessageSize + 1;
2829

2930
public class Context : ScenarioContext
3031
{

‎src/NServiceBus.Transport.SQS.AcceptanceTests/Sending/When_using_small_message_with_no_bucket_configured.cs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,13 +4,14 @@
44
using AcceptanceTesting;
55
using EndpointTemplates;
66
using NUnit.Framework;
7+
using Transport.SQS;
78

89
public class When_using_small_message_with_no_bucket_configured : NServiceBusAcceptanceTest
910
{
1011
[Test]
1112
public async Task Should_receive_message()
1213
{
13-
var payloadToSend = new byte[10];
14+
var payloadToSend = new byte[TransportConstraints.SqsMaximumMessageSize - TransportTestsConstraints.SqsHeadersBuffer];
1415
var context = await Scenario.Define<Context>()
1516
.WithEndpoint<Endpoint>(b => b.When(session => session.SendLocal(new MyMessage
1617
{

‎src/NServiceBus.Transport.SQS.AcceptanceTests/TransportTestsConstraints.cs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
namespace NServiceBus.AcceptanceTests;
2+
3+
class TransportTestsConstraints
4+
{
5+
public const int SqsHeadersBuffer = 500;
6+
}

‎src/NServiceBus.Transport.SQS.Tests/Receiving/DelayedMessagesPumpTests.cs‎

Lines changed: 16 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -165,7 +165,7 @@ public async Task Consume_sends_due_messages_in_batches_to_the_input_queue()
165165
mockSqsClient.ReceiveMessagesRequestResponse = (req, token) =>
166166
{
167167
var receivedMessages = new List<Message>();
168-
for (var i = 0; i < 10; i++)
168+
for (var i = 0; i < TransportConstraints.MaximumItemsInBatch; i++)
169169
{
170170
receivedMessages.Add(new Message
171171
{
@@ -180,7 +180,7 @@ public async Task Consume_sends_due_messages_in_batches_to_the_input_queue()
180180
{ TransportHeaders.DelaySeconds, new MessageAttributeValue { StringValue = "20" }},
181181
{ Headers.MessageId, new MessageAttributeValue { StringValue = Guid.NewGuid().ToString() }}
182182
},
183-
Body = new string('a', 50 * 1024)
183+
Body = new string('a', GetSqsMessageSizeToEnsureNumberOfBatches(2))
184184
});
185185
}
186186

@@ -230,7 +230,7 @@ public async Task Consume_deletes_due_messages_in_batches()
230230
{ TransportHeaders.DelaySeconds, new MessageAttributeValue { StringValue = "20" }},
231231
{ Headers.MessageId, new MessageAttributeValue { StringValue = messageId }}
232232
},
233-
Body = new string('a', 50 * 1024),
233+
Body = new string('a', GetSqsMessageSizeToEnsureNumberOfBatches(2)),
234234
ReceiptHandle = $"Message-{i}"
235235
});
236236
}
@@ -285,7 +285,7 @@ public async Task Consume_sends_not_yet_due_messages_in_batches_back_to_delayed_
285285
{ TransportHeaders.DelaySeconds, new MessageAttributeValue { StringValue = "1500" }},
286286
{ Headers.MessageId, new MessageAttributeValue { StringValue = messageId }}
287287
},
288-
Body = new string('a', 50 * 1024)
288+
Body = new string('a', GetSqsMessageSizeToEnsureNumberOfBatches(2))
289289
});
290290
}
291291

@@ -335,7 +335,7 @@ public async Task Consume_deletes_not_yet_due_messages_in_batches()
335335
{ TransportHeaders.DelaySeconds, new MessageAttributeValue { StringValue = "1500" }},
336336
{ Headers.MessageId, new MessageAttributeValue { StringValue = messageId }}
337337
},
338-
Body = new string('a', 50 * 1024),
338+
Body = new string('a', GetSqsMessageSizeToEnsureNumberOfBatches(2)),
339339
ReceiptHandle = $"Message-{i}"
340340
});
341341
}
@@ -777,6 +777,17 @@ public async Task Consume_if_change_visibility_fails_does_not_rethrow() // chang
777777
});
778778
}
779779

780+
int GetSqsMessageSizeToEnsureNumberOfBatches(int desiredBatchCount, int? numberOfMessages = null)
781+
{
782+
var maxSqsBatchSize = TransportConstraints.SqsMaximumMessageSize;
783+
desiredBatchCount = Math.Max(desiredBatchCount, 1);
784+
numberOfMessages ??= TransportConstraints.MaximumItemsInBatch;
785+
786+
var totalSize = maxSqsBatchSize * desiredBatchCount;
787+
var messageSize = (int)Math.Floor((double)totalSize / numberOfMessages.Value) - TransportTestsConstraints.SqsHeadersBuffer;
788+
return messageSize;
789+
}
790+
780791
DelayedMessagesPump pump;
781792
MockSqsClient mockSqsClient;
782793
CancellationTokenSource cancellationTokenSource;

‎src/NServiceBus.Transport.SQS.Tests/Sending/MessageDispatcherTests.cs‎

Lines changed: 22 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -325,11 +325,11 @@ public async Task Should_upload_large_multicast_operations_request_to_s3(bool wr
325325

326326
var longBodyMessageId = Guid.NewGuid().ToString();
327327
/* Crazy long message id will cause the message to go over limits because attributes count as well */
328-
var crazyLongMessageId = new string('x', 256 * 1024);
328+
var crazyLongMessageId = new string('x', TransportConstraints.SnsMaximumMessageSize);
329329

330330
var transportOperations = new TransportOperations(
331331
new TransportOperation(
332-
new OutgoingMessage(longBodyMessageId, [], Encoding.UTF8.GetBytes(new string('x', 256 * 1024))),
332+
new OutgoingMessage(longBodyMessageId, [], Encoding.UTF8.GetBytes(new string('x', TransportConstraints.SnsMaximumMessageSize))),
333333
new MulticastAddressTag(typeof(Event)),
334334
[],
335335
DispatchConsistency.Isolated),
@@ -704,22 +704,21 @@ public async Task Should_upload_large_non_isolated_operations_to_s3(bool wrapMes
704704
var mockSqsClient = new MockSqsClient();
705705

706706
var dispatcher = new MessageDispatcher(new SettingsHolder(), mockSqsClient, null, new QueueCache(mockSqsClient,
707-
dest => QueueCache.GetSqsQueueName(dest, "")), null,
708-
new S3Settings("someBucket", "somePrefix", mockS3Client), 15 * 60, 0, wrapOutgoingMessages: wrapMessage);
707+
dest => QueueCache.GetSqsQueueName(dest, "")), null, new S3Settings("someBucket", "somePrefix", mockS3Client), 15 * 60, 0, wrapOutgoingMessages: wrapMessage);
709708

710709
var transportOperations = new TransportOperations(
711710
new TransportOperation(
712-
new OutgoingMessage(Guid.NewGuid().ToString(), [], Encoding.UTF8.GetBytes(new string('x', 256 * 1024))),
711+
new OutgoingMessage(Guid.NewGuid().ToString(), [], Encoding.UTF8.GetBytes(new string('x', TransportConstraints.SqsMaximumMessageSize))),
713712
new UnicastAddressTag("address1"),
714713
[],
715714
DispatchConsistency.Default),
716715
new TransportOperation(
717-
new OutgoingMessage(Guid.NewGuid().ToString(), [], Encoding.UTF8.GetBytes(new string('x', 256 * 1024))),
716+
new OutgoingMessage(Guid.NewGuid().ToString(), [], Encoding.UTF8.GetBytes(new string('x', TransportConstraints.SqsMaximumMessageSize))),
718717
new UnicastAddressTag("address2"),
719718
[],
720719
DispatchConsistency.Default),
721720
new TransportOperation( /* Crazy long message id will cause the message to go over limits because attributes count as well */
722-
new OutgoingMessage(new string('x', 256 * 1024), [], Encoding.UTF8.GetBytes("{}")),
721+
new OutgoingMessage(new string('x', TransportConstraints.SqsMaximumMessageSize), [], Encoding.UTF8.GetBytes("{}")),
723722
new UnicastAddressTag("address2"),
724723
[],
725724
DispatchConsistency.Default));
@@ -767,22 +766,21 @@ public async Task Should_upload_large_isolated_operations_request_to_s3(bool wra
767766
var mockSqsClient = new MockSqsClient();
768767

769768
var dispatcher = new MessageDispatcher(new SettingsHolder(), mockSqsClient, null, new QueueCache(mockSqsClient,
770-
dest => QueueCache.GetSqsQueueName(dest, "")), null,
771-
new S3Settings("someBucket", "somePrefix", mockS3Client), 15 * 60, 0, wrapOutgoingMessages: wrapMessage);
769+
dest => QueueCache.GetSqsQueueName(dest, "")), null, new S3Settings("someBucket", "somePrefix", mockS3Client), 15 * 60, 0, wrapMessage);
772770

773771
var transportOperations = new TransportOperations(
774772
new TransportOperation(
775-
new OutgoingMessage(Guid.NewGuid().ToString(), [], Encoding.UTF8.GetBytes(new string('x', 256 * 1024))),
773+
new OutgoingMessage(Guid.NewGuid().ToString(), [], Encoding.UTF8.GetBytes(new string('x', TransportConstraints.SqsMaximumMessageSize))),
776774
new UnicastAddressTag("address1"),
777775
[],
778776
DispatchConsistency.Isolated),
779777
new TransportOperation(
780-
new OutgoingMessage(Guid.NewGuid().ToString(), [], Encoding.UTF8.GetBytes(new string('x', 256 * 1024))),
778+
new OutgoingMessage(Guid.NewGuid().ToString(), [], Encoding.UTF8.GetBytes(new string('x', TransportConstraints.SqsMaximumMessageSize))),
781779
new UnicastAddressTag("address2"),
782780
[],
783781
DispatchConsistency.Isolated),
784782
new TransportOperation( /* Crazy long message id will cause the message to go over limits because attributes count as well */
785-
new OutgoingMessage(new string('x', 256 * 1024), [], Encoding.UTF8.GetBytes("{}")),
783+
new OutgoingMessage(new string('x', TransportConstraints.SqsMaximumMessageSize), [], Encoding.UTF8.GetBytes("{}")),
786784
new UnicastAddressTag("address2"),
787785
[],
788786
DispatchConsistency.Isolated));
@@ -930,10 +928,11 @@ public async Task Should_handle_empty_body_gracefully(bool wrapMessage)
930928
public async Task Does_upload_to_S3_when_payload_reserved_bytes_are_set()
931929
{
932930
var mockSqsClient = new MockSqsClient();
931+
var mockSnsClient = new MockSnsClient();
933932
var mockS3Client = new MockS3Client();
934933

935-
var dispatcher = new MessageDispatcher(new SettingsHolder(), mockSqsClient, null, new QueueCache(mockSqsClient,
936-
dest => QueueCache.GetSqsQueueName(dest, "")), null, new S3Settings("bucketname", "keyprefix", mockS3Client), 15 * 60, 50);
934+
var dispatcher = new MessageDispatcher(new SettingsHolder(), mockSqsClient, mockSnsClient, new QueueCache(mockSqsClient,
935+
dest => QueueCache.GetSqsQueueName(dest, "")), new TopicCache(mockSnsClient, new SettingsHolder(), new EventToTopicsMappings(), new EventToEventsMappings(), (type, s) => TopicNameHelper.GetSnsTopicName(type, ""), ""), new S3Settings("bucketname", "keyprefix", mockS3Client), 15 * 60, 50);
937936

938937
var transportOperations = new TransportOperations(
939938
new TransportOperation(
@@ -942,20 +941,23 @@ public async Task Does_upload_to_S3_when_payload_reserved_bytes_are_set()
942941
{TransportHeaders.TimeToBeReceived, ExpectedTtbr.ToString()},
943942
{Headers.ReplyToAddress, ExpectedReplyToAddress},
944943
{Headers.MessageId, "093C17C6-D32E-44FE-9134-65C10C1287EB"}
945-
}, Encoding.Default.GetBytes(new string('x', 256 * 1024))),
944+
}, Encoding.Default.GetBytes(new string('x', TransportConstraints.SqsMaximumMessageSize))),
946945
new UnicastAddressTag("address"),
947946
[],
947+
DispatchConsistency.Isolated),
948+
new TransportOperation(
949+
new OutgoingMessage("5678", new Dictionary<string, string> { { TransportHeaders.TimeToBeReceived, ExpectedTtbr.ToString() }, { Headers.ReplyToAddress, ExpectedReplyToAddress }, { Headers.MessageId, "093C17C6-D32E-44FE-9134-65C10C1287EC" } }, Encoding.Default.GetBytes(new string('x', TransportConstraints.SnsMaximumMessageSize))),
950+
new MulticastAddressTag(typeof(Event)),
951+
[],
948952
DispatchConsistency.Isolated));
949953

950954
var transportTransaction = new TransportTransaction();
951955

952956
await dispatcher.Dispatch(transportOperations, transportTransaction);
953957

954-
Assert.Multiple(() =>
955-
{
956-
Assert.That(mockSqsClient.RequestsSent, Is.Not.Empty, "No requests sent");
957-
Assert.That(mockS3Client.PutObjectRequestsSent, Is.Not.Empty, "No S3 requests sent");
958-
});
958+
Assert.That(mockSqsClient.RequestsSent, Is.Not.Empty, "No SQS requests sent");
959+
Assert.That(mockSnsClient.PublishedEvents, Is.Not.Empty, "No SNS events published");
960+
Assert.That(mockS3Client.PutObjectRequestsSent, Is.Not.Empty, "No S3 requests sent");
959961

960962
var s3Request = mockS3Client.PutObjectRequestsSent.First();
961963
Assert.That(s3Request.InputStream.Length, Is.GreaterThan(0));

0 commit comments

Comments
 (0)