Kanject.Core.Queue

View .md

A typed SQS layer: publish a message with IQueueManager.EnqueueAsync, consume a batch by marking a partial class [QueueConsumer] — the source generator makes it an AbstractQueueConsumer<T> with the receive loop, acknowledgement, retries, and dead-letter handling wired in. Free under the Community licence.

Install

bash 
dotnet add package Kanject.Core.Queue.Provider.AwsSqs

Define the message

csharp 
using Kanject.Core.Queue.Provider.AwsSqs.Annotations.Attributes;​// The message. [QueueMessage] names the queue it belongs to.[QueueMessage("order-events")]public record struct OrderPlaced(Guid OrderId, string CustomerId);

Produce

Inject IQueueManager and enqueue the typed message — serialization and the target queue are resolved from the [QueueMessage] attribute:

csharp 
// Produce: inject IQueueManager and enqueue a typed message.public class CheckoutService(IQueueManager queue){    public Task PlaceAsync(Guid orderId, string customerId)        => queue.EnqueueAsync(new OrderPlaced(orderId, customerId));}

Consume

A consumer is a partial class with [QueueConsumer]. The generator makes it inherit AbstractQueueConsumer<T>; you override the batched ConsumeAsync, AcknowledgeAsync the messages you handle, and push the rest onto Response.BatchItemFailures so SQS redelivers only those:

csharp 
using Kanject.Core.Queue.Provider.AwsSqs.Annotations.Attributes;​// Consume: a partial class the generator makes inherit// AbstractQueueConsumer<OrderPlaced>. You override ConsumeAsync (batched)// and acknowledge each message you successfully handle.[QueueConsumer(    QueueName      = "order-events",    QueueNamespace = "beanandbark",    Message        = typeof(OrderPlaced))]public partial class OrderPlacedConsumer{    protected override async Task ConsumeAsync(        List<MessageContext<OrderPlaced>> messages)    {        foreach (var ctx in messages)        {            try            {                await Fulfil(ctx.Message);                await AcknowledgeAsync(ctx);   // handled → delete from the queue            }            catch (Exception ex)            {                ex.PrintInConsole(tag: nameof(ConsumeAsync));                // Leave it un-acked → SQS redelivers, then DLQs after N tries.                Response.BatchItemFailures.Add(new SQSBatchResponse.BatchItemFailure                    { ItemIdentifier = ctx.MessageId });            }        }    }}

Register

csharp 
// Global SQS config, then register each consumer.builder.Services    .AddAwsSqsGlobalQueueConfiguration(options =>    {        options.AWSRegion         = appSettings.AwsRegion;        options.Namespace         = appSettings.Stage;   // per-stage queue isolation        options.UseDeadLetterQueue = true;               // sibling DLQ auto-created    })    .AddQueueConsumer<OrderPlacedConsumer>(options =>    {        options.QueueName                 = "order-events";        options.MaximumReceiveMessageCount = 5;        options.MaximumMessageRetry        = 5;          // then → DLQ    });

One global SQS configuration, then a fluent AddQueueConsumer<T> per consumer. UseDeadLetterQueue auto-provisions a sibling DLQ; MaximumMessageRetry sets how many redeliveries a message gets before it lands there.

Dispatch from Lambda

In an SQS-triggered function, hand the event batch to the consumer and return the partial-batch response:

csharp 
// SQS-triggered Lambda: dispatch the event batch to the consumer and return// the partial-batch response SQS expects.public Task<SQSBatchResponse> Handle(SQSEvent sqsEvent)    => ServiceProvider.ProcessSqsEventWithQueueConsumerAsync<OrderPlacedConsumer>(sqsEvent);

What ships with it

  • IQueueManager.EnqueueAsync<T>(message) — typed producer, JSON serialized
  • [QueueMessage("queue")] on the message; [QueueConsumer(QueueName, QueueNamespace, Message)] on the consumer
  • Generated AbstractQueueConsumer<T> with a batched ConsumeAsync + AcknowledgeAsync — no receive-loop boilerplate
  • Partial-batch failures via Response.BatchItemFailures — only un-acked messages redeliver
  • Automatic DLQ provisioning (UseDeadLetterQueue) + retry budget (MaximumMessageRetry)
  • ProcessSqsEventWithQueueConsumerAsync<T> for the Lambda entry point; Namespace for per-stage isolation
Was this page helpful?