Pervaxis.Genesis.Messaging.AWS 2.0.0

Pervaxis.Genesis.Messaging.AWS

AWS messaging provider for the Pervaxis Genesis platform — SQS for queue-based messaging and SNS for pub/sub topic delivery.

Overview

Pervaxis.Genesis.Messaging.AWS implements the IMessaging abstraction from Pervaxis.Core.Abstractions using the AWS SDK for SQS and SNS. Both providers share a single MessagingOptions and can be registered independently or together as keyed services.

Installation

<PackageReference Include="Pervaxis.Genesis.Messaging.AWS" Version="1.0.0" />

Configuration

appsettings.json

{
  "Messaging": {
    "Region": "ap-south-1",
    "Sqs": {
      "DefaultQueueUrl": "https://sqs.ap-south-1.amazonaws.com/123456789012/default-queue",
      "QueueUrlMappings": {
        "orders": "https://sqs.ap-south-1.amazonaws.com/123456789012/orders-queue",
        "payments": "https://sqs.ap-south-1.amazonaws.com/123456789012/payments-queue"
      },
      "MaxNumberOfMessages": 10,
      "WaitTimeSeconds": 20,
      "VisibilityTimeoutSeconds": 30
    },
    "Sns": {
      "DefaultTopicArn": "arn:aws:sns:ap-south-1:123456789012:default-topic",
      "TopicArnMappings": {
        "order-events": "arn:aws:sns:ap-south-1:123456789012:order-events-topic"
      }
    }
  }
}

Option properties

Root (MessagingOptions)

SQS (SqsOptions)

SNS (SnsOptions)

Registration

SQS only

// From IConfiguration
builder.Services.AddGenesisSqsMessaging(
    builder.Configuration.GetSection("Messaging"));

// From action
builder.Services.AddGenesisSqsMessaging(options =>
{
    options.Region = "ap-south-1";
    options.Sqs.DefaultQueueUrl = "https://sqs.ap-south-1.amazonaws.com/123/my-queue";
});

SNS only

builder.Services.AddGenesisSnsMess aging(
    builder.Configuration.GetSection("Messaging"));

Both (keyed services)

builder.Services.AddGenesisMessaging(
    builder.Configuration.GetSection("Messaging"));

Inject by key:

public class OrderService(
    [FromKeyedServices("sqs")] IMessaging queue,
    [FromKeyedServices("sns")] IMessaging topic)

All providers are registered as Singleton. AWS clients use Lazy<T> and are created on first use.

Usage

Publishing to SQS

public class OrderService(IMessaging messaging)
{
    public async Task PlaceOrderAsync(Order order, CancellationToken ct = default)
    {
        var messageId = await messaging.PublishAsync("orders", order, ct);
        // messageId = SQS MessageId
    }

    public async Task PlaceOrdersBatchAsync(IEnumerable<Order> orders, CancellationToken ct = default)
    {
        var messageIds = await messaging.PublishBatchAsync("orders", orders, ct);
        // Automatically chunked into batches of 10
    }
}

Receiving and deleting from SQS

public class OrderConsumer(IMessaging messaging)
{
    public async Task ProcessAsync(CancellationToken ct = default)
    {
        var orders = await messaging.ReceiveAsync<Order>("orders", maxMessages: 5, ct);
        foreach (var order in orders)
        {
            await ProcessOrderAsync(order, ct);
            // Note: receipt handle is not returned through IMessaging.
            // For full consume-then-delete, inject SqsMessagingProvider directly
            // or use a background worker with the AWS SDK.
        }
    }
}

Receipt handle note: IMessaging.ReceiveAsync<T> returns deserialized message bodies only. To delete a message after processing via DeleteAsync, you need its receipt handle — which requires direct access to the SQS response. For worker/consumer patterns, consider injecting SqsMessagingProvider directly, or using a hosted background service.

Publishing to SNS

public class EventPublisher([FromKeyedServices("sns")] IMessaging topic)
{
    public async Task PublishOrderEventAsync(OrderPlacedEvent evt, CancellationToken ct = default)
    {
        var messageId = await topic.PublishAsync("order-events", evt, ct);
    }
}

Subscribing to SNS

var subscriptionArn = await topic.SubscribeAsync(
    "order-events",
    "https://my-service.example.com/webhooks/orders");

// Protocol is inferred automatically:
// https:// → "https"
// http://  → "http"
// arn:aws:sqs:... → "sqs"
// user@domain → "email"

IAM Permissions

SQS

{
  "Effect": "Allow",
  "Action": [
    "sqs:SendMessage",
    "sqs:SendMessageBatch",
    "sqs:ReceiveMessage",
    "sqs:DeleteMessage",
    "sqs:DeleteMessageBatch",
    "sqs:GetQueueAttributes"
  ],
  "Resource": "arn:aws:sqs:ap-south-1:123456789012:*"
}

SNS

{
  "Effect": "Allow",
  "Action": [
    "sns:Publish",
    "sns:PublishBatch",
    "sns:Subscribe"
  ],
  "Resource": "arn:aws:sns:ap-south-1:123456789012:*"
}

Local Development with LocalStack

{
  "Messaging": {
    "Region": "us-east-1",
    "UseLocalEmulator": true,
    "LocalEmulatorUrl": "http://localhost:4566",
    "Sqs": {
      "DefaultQueueUrl": "http://localhost:4566/000000000000/my-queue"
    },
    "Sns": {
      "DefaultTopicArn": "arn:aws:sns:us-east-1:000000000000:my-topic"
    }
  }
}

# Start LocalStack
docker run -d -p 4566:4566 localstack/localstack

# Create queue and topic
awslocal sqs create-queue --queue-name my-queue
awslocal sns create-topic --name my-topic

Troubleshooting