Compression and JustSaying 8.0 Changes

src/JustSaying/AwsTools/MessageHandling/ReceivedMessageConverter.cs

@@ -30,7 +30,7 @@ public ValueTask<ReceivedMessage> ConvertForReceiveAsync(Amazon.SQS.Model.Messag
             var jsonNode = JsonNode.Parse(body);
             if (jsonNode is JsonObject jsonObject && jsonObject.TryGetPropertyValue("Message", out var messageNode))
             {
-                body = messageNode?.ToString();
+                body = messageNode?.GetValue<string>();
             }
         }
         body = ApplyBodyDecompression(body, attributes);

src/JustSaying/AwsTools/MessageHandling/SnsMessagePublisher.cs

@@ -3,36 +3,32 @@
 using Amazon.SimpleNotificationService.Model;
 using JustSaying.Messaging;
 using JustSaying.Messaging.Interrogation;
-using JustSaying.Messaging.MessageSerialization;
 using JustSaying.Models;
 using Microsoft.Extensions.Logging;
 using MessageAttributeValue = Amazon.SimpleNotificationService.Model.MessageAttributeValue;
 
 namespace JustSaying.AwsTools.MessageHandling;
 
-internal class SnsMessagePublisher(
+internal sealed class SnsMessagePublisher(
     IAmazonSimpleNotificationService client,
     IPublishMessageConverter messageConverter,
     ILoggerFactory loggerFactory,
-    IMessageSubjectProvider messageSubjectProvider,
     Func<Exception, Message, bool> handleException = null) : IMessagePublisher, IInterrogable
 {
     private readonly IPublishMessageConverter _messageConverter = messageConverter;
-    private readonly IMessageSubjectProvider _messageSubjectProvider = messageSubjectProvider;
     private readonly Func<Exception, Message, bool> _handleException = handleException;
+    private readonly IAmazonSimpleNotificationService _client = client;
     public Action<MessageResponse, Message> MessageResponseLogger { get; set; }
     public string Arn { get; internal set; }
-    protected IAmazonSimpleNotificationService Client { get; } = client;
     private readonly ILogger _logger = loggerFactory.CreateLogger("JustSaying.Publish");
 
     public SnsMessagePublisher(
         string topicArn,
         IAmazonSimpleNotificationService client,
         IPublishMessageConverter messageConverter,
         ILoggerFactory loggerFactory,
-        IMessageSubjectProvider messageSubjectProvider,
         Func<Exception, Message, bool> handleException = null)
-        : this(client, messageConverter, loggerFactory, messageSubjectProvider, handleException)
+        : this(client, messageConverter, loggerFactory,handleException)
     {
         Arn = topicArn;
     }
@@ -51,7 +47,7 @@ public async Task PublishAsync(Message message, PublishMetadata metadata, Cancel
         PublishResponse response = null;
         try
         {
-            response = await Client.PublishAsync(request, cancellationToken).ConfigureAwait(false);
+            response = await _client.PublishAsync(request, cancellationToken).ConfigureAwait(false);
         }
         catch (AmazonServiceException ex)
         {
@@ -138,7 +134,7 @@ private static MessageAttributeValue BuildMessageAttributeValue(Messaging.Messag
         };
     }
 
-    public virtual InterrogationResult Interrogate()
+    public InterrogationResult Interrogate()
     {
         return new InterrogationResult(new
         {

src/JustSaying/Fluent/PublishConfig/StaticPublicationConfiguration.cs

@@ -41,8 +41,7 @@ public static StaticPublicationConfiguration Build<T>(
         var eventPublisher = new SnsMessagePublisher(
             snsClient,
             new PublishMessageConverter(PublishDestinationType.Topic, serializer, new MessageCompressionRegistry([new GzipMessageBodyCompression()]), compressionOptions, subject, writeConfiguration.IsRawMessage),
-            loggerFactory,
-            bus.Config.MessageSubjectProvider)
+            loggerFactory)
         {
             MessageResponseLogger = bus.Config.MessageResponseLogger,
         };

src/JustSaying/Fluent/TopicAddressPublicationBuilder`1.cs

@@ -71,7 +71,6 @@ public void Configure(JustSayingBus bus, IAwsClientFactoryProxy proxy, ILoggerFa
 
         logger.LogInformation("Adding SNS publisher for message type '{MessageType}'", typeof(T));
 
-        var config = bus.Config;
         var arn = Arn.Parse(_topicAddress.TopicArn);
 
         var compressionRegistry = bus.CompressionRegistry;
@@ -85,7 +84,6 @@ public void Configure(JustSayingBus bus, IAwsClientFactoryProxy proxy, ILoggerFa
             proxy.GetAwsClientFactory().GetSnsClient(RegionEndpoint.GetBySystemName(arn.Region)),
             new PublishMessageConverter(PublishDestinationType.Topic, serializer, compressionRegistry, compressionOptions, subject, true),
             loggerFactory,
-            subjectProvider,
             _exceptionHandler);
 
         CompressionEncodingValidator.ValidateEncoding(bus.CompressionRegistry, compressionOptions);

tests/JustSaying.IntegrationTests/Fluent/AwsTools/WhenUsingABasicThrottle.cs

@@ -1,4 +1,5 @@
 using System.Diagnostics;
+using System.Text.Json;
 using Amazon.SQS.Model;
 using JustSaying.AwsTools;
 using JustSaying.AwsTools.MessageHandling;
@@ -72,7 +73,7 @@ public async Task Messages_Are_Throttled_But_Still_Delivered(int throttleMessage
             {
                 var batchEntry = new SendMessageBatchRequestEntry
                 {
-                    MessageBody = $$"""{"Subject":"SimpleMessage", "Message": { "Content": "{{entriesAdded}}"} }""",
+                    MessageBody = $$"""{"Subject":"SimpleMessage", "Message": "{{JsonEncodedText.Encode($$"""{ "Content": "{{entriesAdded}}" }""")}}" }""",
                     Id = Guid.NewGuid().ToString()
                 };
 

tests/JustSaying.UnitTests/AwsTools/MessageHandling/Sns/TopicByName/WhenPublishing.cs

@@ -18,7 +18,7 @@ public class WhenPublishing : WhenPublishingTestBase
     private protected override Task<SnsMessagePublisher> CreateSystemUnderTestAsync()
     {
         var messageConverter = new PublishMessageConverter(PublishDestinationType.Topic, new FakeBodySerializer(Message), new MessageCompressionRegistry(), new PublishCompressionOptions(), nameof(SimpleMessage), false);
-        var topic = new SnsMessagePublisher(TopicArn, Sns, messageConverter, NullLoggerFactory.Instance, new NonGenericMessageSubjectProvider());
+        var topic = new SnsMessagePublisher(TopicArn, Sns, messageConverter, NullLoggerFactory.Instance);
         return Task.FromResult(topic);
     }
 

tests/JustSaying.UnitTests/AwsTools/MessageHandling/Sns/TopicByName/WhenPublishingAsync.cs

@@ -22,7 +22,7 @@ public class WhenPublishingAsync : WhenPublishingTestBase
     private protected override Task<SnsMessagePublisher> CreateSystemUnderTestAsync()
     {
         var messageConverter = new PublishMessageConverter(PublishDestinationType.Topic, new FakeBodySerializer(Message), new MessageCompressionRegistry(), new PublishCompressionOptions(), nameof(SimpleMessage), false);
-        var topic = new SnsMessagePublisher(TopicArn, Sns, messageConverter, NullLoggerFactory.Instance, new NonGenericMessageSubjectProvider());
+        var topic = new SnsMessagePublisher(TopicArn, Sns, messageConverter, NullLoggerFactory.Instance);
         return Task.FromResult(topic);
     }
 

tests/JustSaying.UnitTests/AwsTools/MessageHandling/Sns/TopicByName/WhenPublishingAsyncExceptionCanBeHandled.cs

@@ -17,7 +17,7 @@ public class WhenPublishingAsyncExceptionCanBeHandled : WhenPublishingTestBase
     private protected override Task<SnsMessagePublisher> CreateSystemUnderTestAsync()
     {
         var messageConverter = new PublishMessageConverter(PublishDestinationType.Topic, new NewtonsoftMessageBodySerializer<SimpleMessage>(), new MessageCompressionRegistry(), new PublishCompressionOptions(), "Subject", false);
-        var topic = new SnsMessagePublisher(TopicArn, Sns, messageConverter, NullLoggerFactory.Instance, Substitute.For<IMessageSubjectProvider>(), (_, _) => true);
+        var topic = new SnsMessagePublisher(TopicArn, Sns, messageConverter, NullLoggerFactory.Instance, (_, _) => true);
 
         return Task.FromResult(topic);
     }

tests/JustSaying.UnitTests/AwsTools/MessageHandling/Sns/TopicByName/WhenPublishingAsyncExceptionCanBeThrown.cs

@@ -18,7 +18,7 @@ public class WhenPublishingAsyncExceptionCanBeThrown : WhenPublishingTestBase
     private protected override Task<SnsMessagePublisher> CreateSystemUnderTestAsync()
     {
         var messageConverter = new PublishMessageConverter(PublishDestinationType.Topic, new NewtonsoftMessageBodySerializer<SimpleMessage>(), new MessageCompressionRegistry(), new PublishCompressionOptions(), "Subject", false);
-        var topic = new SnsMessagePublisher(TopicArn, Sns, messageConverter, NullLoggerFactory.Instance, Substitute.For<IMessageSubjectProvider>(), (_, _) => false);
+        var topic = new SnsMessagePublisher(TopicArn, Sns, messageConverter, NullLoggerFactory.Instance, (_, _) => false);
         return Task.FromResult(topic);
     }
 

tests/JustSaying.UnitTests/AwsTools/MessageHandling/Sns/TopicByName/WhenPublishingAsyncResponseLoggerIsCalled.cs

@@ -26,7 +26,7 @@ public class WhenPublishingAsyncResultLoggerIsCalled : WhenPublishingTestBase
     private protected override Task<SnsMessagePublisher> CreateSystemUnderTestAsync()
     {
         var messageConverter = new PublishMessageConverter(PublishDestinationType.Topic, new NewtonsoftMessageBodySerializer<SimpleMessage>(), new MessageCompressionRegistry(), new PublishCompressionOptions(), "Subject", false);
-        var topic = new SnsMessagePublisher(TopicArn, Sns, messageConverter, NullLoggerFactory.Instance, Substitute.For<IMessageSubjectProvider>())
+        var topic = new SnsMessagePublisher(TopicArn, Sns, messageConverter, NullLoggerFactory.Instance)
         {
             MessageResponseLogger = (r, m) =>
             {

tests/JustSaying.UnitTests/AwsTools/MessageHandling/Sns/TopicByName/WhenPublishingAsyncWithGenericMessageSubjectProvider.cs

@@ -21,7 +21,7 @@ private protected override Task<SnsMessagePublisher> CreateSystemUnderTestAsync(
     {
         var subject = new GenericMessageSubjectProvider().GetSubjectForType(typeof(MessageWithTypeParameters<int, string>));
         var messageConverter = new PublishMessageConverter(PublishDestinationType.Topic, new FakeBodySerializer(Message), new MessageCompressionRegistry(), new PublishCompressionOptions(), subject, false);
-        var topic = new SnsMessagePublisher(TopicArn, Sns, messageConverter, NullLoggerFactory.Instance, new GenericMessageSubjectProvider());
+        var topic = new SnsMessagePublisher(TopicArn, Sns, messageConverter, NullLoggerFactory.Instance);
         return Task.FromResult(topic);
     }
 

tests/JustSaying.UnitTests/Messaging/Channels/SubscriptionGroupTests/WhenExactlyOnceIsAppliedToHandler.cs

@@ -1,3 +1,4 @@
+using System.Text.Json;
 using JustSaying.AwsTools.MessageHandling;
 using JustSaying.Messaging.Channels.SubscriptionGroups;
 using JustSaying.Messaging.MessageHandling;
@@ -18,10 +19,11 @@ public class WhenExactlyOnceIsAppliedToHandler(ITestOutputHelper testOutputHelpe
 
     protected override void Given()
     {
-        _queue = CreateSuccessfulTestQueue("TestQueue", new TestMessage
-        {
-            Body = """{"Subject":"SimpleMessage", "Message": { "Content": "Hi"} }"""
-        });
+        _queue = CreateSuccessfulTestQueue("TestQueue",
+            new TestMessage
+            {
+                Body = $$"""{"Subject":"SimpleMessage", "Message": "{{JsonEncodedText.Encode("""{ "Content": "Hi"} }""")}}"}"""
+            });
 
         Queues.Add(_queue);
 

tests/JustSaying.UnitTests/Messaging/Channels/SubscriptionGroupTests/WhenExactlyOnceIsAppliedToHandlerWithoutExplicitTimeout.cs

@@ -1,3 +1,4 @@
+using System.Text.Json;
 using JustSaying.AwsTools.MessageHandling;
 using JustSaying.Messaging.Channels.SubscriptionGroups;
 using JustSaying.Messaging.MessageHandling;
@@ -18,10 +19,11 @@ public class WhenExactlyOnceIsAppliedWithoutSpecificTimeout(ITestOutputHelper te
 
     protected override void Given()
     {
-        _queue = CreateSuccessfulTestQueue(Guid.NewGuid().ToString(), new TestMessage
-        {
-            Body = """{"Subject":"SimpleMessage", "Message": { "Content": "Hi"} }"""
-        });
+        _queue = CreateSuccessfulTestQueue(Guid.NewGuid().ToString(),
+            new TestMessage
+            {
+                Body = $$"""{"Subject":"SimpleMessage", "Message": "{{JsonEncodedText.Encode("""{ "Content": "Hi" }""")}}"}"""
+            });
         Queues.Add(_queue);
         _messageLock = new FakeMessageLock();