Completed messaging refactoring. Test coverage to be completed.

Author: MarkGentHe1

compose/start-localstack.sh

@@ -19,4 +19,58 @@ queue_url=$(awslocal sqs create-queue  \
 
 echo "SQS Queue created: $queue_url"
 
+# Get the SQS Queue ARN
+queue_arn=$(awslocal sqs get-queue-attributes \
+  --queue-url "$queue_url" \
+  --attribute-name QueueArn \
+  --output text \
+  --query 'Attributes.QueueArn')
+
+echo "SQS Queue ARN: $queue_arn"
+
+# Create SNS Topics
+topic_arn=$(awslocal sns create-topic \
+  --name ls-keeper-data-bridge-events \
+  --endpoint-url=http://localhost:4566 \
+  --output text \
+  --query 'TopicArn')
+
+echo "SNS Topic created: $topic_arn"
+
+# Construct the policy JSON inline with escaped quotes
+policy_json=$(cat <<EOF
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Principal": "*",
+      "Action": "sqs:SendMessage",
+      "Resource": "$queue_arn",
+      "Condition": {
+        "ArnEquals": {
+          "aws:SourceArn": "$topic_arn"
+        }
+      }
+    }
+  ]
+}
+EOF
+)
+
+# Set SQS policy
+awslocal sqs set-queue-attributes \
+  --queue-url "$queue_url" \
+  --attributes "{\"Policy\": \"$(
+    echo "$policy_json" | jq -c
+  )\"}"
+
+# Subscribe the Queue to the Topic
+awslocal sns subscribe \
+  --topic-arn "$topic_arn" \
+  --protocol sqs \
+  --notification-endpoint "$queue_arn"
+
+echo "SNS Topic subscription complete"
+
 echo "Bootstrapping Complete"

docker-compose.override.yml

@@ -6,7 +6,9 @@ services:
       - ASPNETCORE_ENVIRONMENT=Development
       - ASPNETCORE_HTTP_PORTS=8080
       - AWS__ServiceURL=http://localstack-emulator:4566
+      #- AWS__ServiceURL=http://sqs.eu-west-2.localhost.localstack.cloud:4566/
       - Mongo__DatabaseUri=mongodb://mongodb:27017
+      - QueueConsumerOptions__IntakeEventQueueOptions__QueueUrl=http://sqs.eu-west-2.127.0.0.1:4566/000000000000/ls_keeper_data_intake_queue
     ports:
       - "8080"
     volumes:

src/KeeperData.Api/appsettings.IntegrationTest.json

@@ -10,9 +10,9 @@
   },
   "QueueConsumerOptions": {
     "IntakeEventQueueOptions": {
-      "QueueUrl": "http://sqs.eu-west-2.127.0.0.1:4566/000000000000/ls_keeper_data_intake_queue",
+      "QueueUrl": "Set in cdp-app-config (or for local development using docker-compose.override.yml)",
       "MaxNumberOfMessages": 10,
-      "WaitTimeSeconds": 1,
+      "WaitTimeSeconds": 5,
       "Disabled": false
     }
   },

src/KeeperData.Api/appsettings.json

@@ -0,0 +1,22 @@
+using KeeperData.Core.Messaging.Contracts;
+using KeeperData.Core.Messaging.Contracts.V1;
+using KeeperData.Core.Messaging.MessageHandlers;
+using KeeperData.Core.Messaging.Serializers;
+
+namespace KeeperData.Application.MessageHandlers;
+
+public class PlaceholderMessageHandler(IUnwrappedMessageSerializer<PlaceholderMessage> serializer) : IMessageHandler<PlaceholderMessage>
+{
+    private readonly IUnwrappedMessageSerializer<PlaceholderMessage> _serializer = serializer;
+
+    public async Task<MessageType> Handle(UnwrappedMessage message, CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(message, nameof(message));
+
+        var messagePayload = _serializer.Deserialize(message);
+
+        // Do something with the messagePayload
+
+        return await Task.FromResult(messagePayload!);
+    }
+}

src/KeeperData.Application/MessageHandlers/PlaceholderMessageHandler.cs

@@ -0,0 +1,11 @@
+namespace KeeperData.Core.Exceptions;
+
+public class NonRetryableException : Exception
+{
+    public NonRetryableException(string message, Exception inner) : base(message, inner)
+    {
+    }
+    public NonRetryableException(string message) : base(message)
+    {
+    }
+}

src/KeeperData.Core/Exceptions/NonRetryableException.cs

@@ -0,0 +1,11 @@
+namespace KeeperData.Core.Exceptions;
+
+public class RetryableException : Exception
+{
+    public RetryableException(string message, Exception inner) : base(message, inner)
+    {
+    }
+    public RetryableException(string message) : base(message)
+    {
+    }
+}

src/KeeperData.Core/Exceptions/RetryableException.cs

@@ -7,6 +7,9 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="AWSSDK.Core" Version="4.0.0.26" />
+    <PackageReference Include="AWSSDK.SQS" Version="4.0.0.15" />
+    <PackageReference Include="AWSSDK.SecurityToken" Version="4.0.2.1" />
     <PackageReference Include="MongoDB.Driver" Version="2.28.0" />
   </ItemGroup>
 

src/KeeperData.Core/KeeperData.Core.csproj

@@ -0,0 +1,7 @@
+namespace KeeperData.Core.Messaging.Consumers;
+
+public interface IQueuePoller
+{
+    Task StartAsync(CancellationToken token);
+    Task StopAsync(CancellationToken token);
+}