Kanject.Core.Queue.Abstractions 3.7.1

Kanject.Core.Queue.Abstractions

Provider-neutral queue contracts for Kanject services. The package contains the IQueueManager / IQueueManagerService surface for sending, receiving and acknowledging messages, the consumer and router interfaces, the MessageContext<T> envelope, queue-naming helpers, and the scheduling contract IQueueSchedulerProvider with its request and response models.

Reference it from code that sends or handles queue messages but shouldn't depend on a specific broker. The Amazon SQS implementation ships in Kanject.Core.Queue.Provider.AwsSqs.

Installation

dotnet add package Kanject.Core.Queue.Abstractions

Targets .NET 8, .NET 9 and .NET 10. Depends on Kanject.Core.

Quick start

Write application services against IQueueManagerService. The provider package registers the implementation.

using Kanject.Core.Queue.Abstractions.Interfaces;

public sealed class OrderPlaced
{
    public string OrderId { get; set; } = string.Empty;
    public string TenantId { get; set; } = string.Empty;
}

public sealed class CheckoutService(IQueueManagerService queues)
{
    public async Task CompleteAsync(OrderPlaced order)
    {
        var metadata = new Dictionary<string, string> { ["tenant"] = order.TenantId };

        // "orders" identifies the registered queue.
        if (!await queues.EnqueueAsync("orders", order, metadata))
            throw new InvalidOperationException($"Order {order.OrderId} could not be queued.");
    }
}

Message types must have a public parameterless constructor, because every generic member is constrained to where TQueueMessage : new(). The SQS provider sends the metadata dictionary as SQS message attributes.

Sending, receiving and acknowledging

IQueueManager: one queue

Every enqueue method returns Task<bool>: true means the provider accepted the message.

IQueueManagerService: every registered queue

IQueueManagerService exposes one IQueueManager per queue through its indexer, along with shortcuts that take the queue key as their first argument:

// Shortcut form: routeKey is the queue key.
await queues.EnqueueAsync("orders", order);

// Indexer form: reach one queue's IQueueManager, e.g. to tag a route.
await queues["order-events"].EnqueueAsync("order.created", order);

The service also defines Initialize(), which a provider uses to resolve its registered queues and create any that are missing. It also offers DequeueAsync<T>(routeKey) and AcknowledgeAsync<T>(routeKey, context).

MessageContext<T>

Consumers

Extension methods in Kanject.Core.Queue.Abstractions.Extensions:

• serviceProvider.GetQueueConsumer<T>() resolves a registered consumer. The first instance it resolves is cached per type for the lifetime of the process.
• serviceProvider.GetRouteQueueConsumer<T>() resolves a route consumer. This one isn't cached, because route consumers are scoped.
• serviceProvider.UseQueueConsumerRouterWatcher() is defined on ServiceProvider and starts WatchAsync on every hosted IQueueConsumerRouter.

Queue names

QueueNameHelper (in Kanject.Core.Queue.Abstractions.Helpers) holds the naming convention that the SQS provider applies:

[QueueName("orders", CreateIfNotExist = true)] marks a consumer class with its queue name, plus an optional namespace as the second argument. QueueNameHelper.GetQueueName, GetQueueAttributeSettings and GetQueueAttribute read that attribute. QueueConstants holds the shared names: QueueRoute, DefaultRouteKey (_), DefaultQueueSuffix (queue) and DeadLetterQueueNamePrefix (dlq).

Scheduling contract

IQueueSchedulerProvider describes delivering a message to a queue later, either once or on a recurring schedule:

The companion attributes live in Kanject.Core.Queue.Abstractions.Attributes:

• [SchedulableQueue] marks a message type as schedulable. It has the optional properties GroupName, Name, Description and TimeZone (a SystemTimezone from Kanject.Core).
• [CronSchedule("0 9 * * ? *")] sets a six-field EventBridge cron expression. You can pass it as one string or as separate fields: minutes, hours, dayOfMonth, month, dayOfWeek, year.
• [RateSchedule(5, RateUnit.Minutes)] sets a fixed-rate expression. RateUnit is Minutes, Hours or Days, and the value must be positive.

The implementation, and the code generator that reads these attributes, ship in the commercial Kanject.Core.Queue.Provider.AwsSqs.Extensions.EventBridge packages.

Trimming and Native AOT

The assembly is marked AOT-compatible. The generic IQueueManager / AbstractQueueManager members serialize TQueueMessage with reflection-based System.Text.Json, so they carry [RequiresUnreferencedCode] and [RequiresDynamicCode]. The trim and AOT analyzers flag those call sites in a trimmed or Native AOT app.

Related packages

License

Licensed under the Kanject Code Libraries License Agreement (KCLLA); the full text ships in this package as LICENSE.md. Organizations whose trailing-twelve-month gross revenue and total funding raised are each below US$250,000 may use it at no cost under the Free Tier. At or above either threshold a commercial license is required — contact commercial@kanjectbusiness.com.