NexJob.Trigger.AwsSqs 6.7.0

NexJob.Trigger.AwsSqs

AWS SQS trigger for NexJob. Receives messages from an SQS queue and automatically enqueues them as NexJob jobs.

Installation

dotnet add package NexJob.Trigger.AwsSqs

Usage

Basic Setup

using NexJob.Trigger.AwsSqs;

builder.Services.AddNexJobAwsSqsTrigger(options =>
{
    options.QueueUrl = "https://sqs.us-east-1.amazonaws.com/123456789/my-queue";
    options.JobName = typeof(MyJob).AssemblyQualifiedName!;
});

With Custom SQS Client

var sqsClient = new AmazonSQSClient(RegionEndpoint.USEast1);

builder.Services.AddNexJobAwsSqsTrigger(sqsClient, options =>
{
    options.QueueUrl = "https://sqs.us-east-1.amazonaws.com/123456789/my-queue";
    options.JobName = typeof(MyJob).AssemblyQualifiedName!;
    options.MaxMessages = 5;
    options.WaitTimeSeconds = 20;
});

You can also bind the trigger to one job type with the generic overload, which sets JobName for you:

builder.Services.AddNexJobAwsSqsTrigger<ProcessOrderJob>(sqsClient, options =>
{
    options.QueueUrl = "https://sqs.us-east-1.amazonaws.com/123456789/my-queue";
});

SQS does not read a job type from the message attributes: every message on the queue runs the configured job. The job idempotency key is the SQS MessageId.

Configuration Options

Broker Guarantees

This trigger satisfies all 5 NexJob trigger guarantees:

1. At-least-once delivery — messages are never lost before enqueue
2. Visibility extension — messages stay invisible while processing via ChangeMessageVisibilityAsync
3. Explicit ack — messages deleted only after EnqueueAsync succeeds
4. Dead-letter on poison — failed messages reappear and eventually route to DLQ via SQS redrive policy
5. Graceful shutdown — CancellationToken respected, in-flight messages not abandoned

Trace Propagation

The trigger extracts W3C traceparent from SQS message attributes and propagates it to the JobRecord. Set the traceparent message attribute when sending to SQS:

await sqsClient.SendMessageAsync(new SendMessageRequest
{
    QueueUrl = queueUrl,
    MessageBody = payload,
    MessageAttributes = new Dictionary<string, MessageAttributeValue>
    {
        ["traceparent"] = new MessageAttributeValue
        {
            DataType = "String",
            StringValue = Activity.Current?.Id,
        },
    },
});

Strongly-Typed Consumer Registration

You can bind a strongly-typed job consumer directly using the generic overload:

builder.Services.AddNexJobAwsSqsTrigger<ProcessOrderSqsJob>(options =>
{
    options.QueueUrl = "https://sqs.us-east-1.amazonaws.com/.../orders";
    options.TargetQueue = "orders";
});

Operational Visibility

This trigger registers with IListenerRegistry to report live connection states (Starting, Listening, Reconnecting, Faulted, Stopped) directly to the Dashboard /listeners page and the Cluster Pipeline Topology Map.

Known Limitations

• SQS FIFO queues: Deduplication relies on SQS MessageId. For FIFO queues, MessageDeduplicationId can also be used — this is not yet exposed as an option.
• No batching: Messages are processed one at a time to ensure visibility extension and proper ordering.