Kanject.Core.Channels

One handler signature, more than one hosting mode. Channels is a bounded, batching drain loop over System.Threading.Channels: mark a method [Channel], and the source generator emits the typed writer, the DI registration, and a consumer that fits the host — a long-lived BackgroundService, or a drain inside a single Lambda invocation. The engine and generator are free under the Community licence.

Install

dotnet add package Kanject.Core.Channelsdotnet add package Kanject.Core.Channels.Annotations

The runtime carries the engine and the [Channel] attribute; the Annotations package is the generator that turns the attribute into code. Your project file should end up like this:

<ItemGroup>  <PackageReference Include="Kanject.Core.Channels" />  <PackageReference Include="Kanject.Core.Channels.Annotations" PrivateAssets="all" /></ItemGroup>

Write the handler

A handler is an instance method on a partial type that returns Task or ValueTask. Take a single item when BatchSize is 1, or a ReadOnlyMemory<T> batch when it is larger — the batch is passed through without copying:

using Kanject.Core.Channels.Annotations.Attributes;​namespace Shop.Orders;​public sealed record OrderPlaced(string OrderId, decimal Total);​// The containing type must be partial. The handler is an ordinary method.public sealed partial class OrderProjector(IOrderReadModel readModel){    [Channel(Capacity = 5000, BatchSize = 25, FlushIntervalMs = 200)]    public async ValueTask ProjectAsync(        ReadOnlyMemory<OrderPlaced> batch, CancellationToken ct)    {        await readModel.UpsertAsync(batch.ToArray(), ct);    }}

Items reach ProjectAsync in batches of up to 25, and a partial batch is flushed after 200 ms. With the default scoped lifetime, each batch runs in a fresh DI scope.

Register and produce

// The generator names the extension after the type and method.builder.Services.AddSingleton<IOrderReadModel, SqlOrderReadModel>();builder.Services.AddOrderProjectorProjectAsyncChannel();

// Produce through the generated writer interface.public sealed class CheckoutService(    IOrderProjectorProjectAsyncChannelWriter orders){    public ValueTask OnOrderPlacedAsync(OrderPlaced order, CancellationToken ct)        => orders.WriteAsync(order, ct);}

The channel, its reader and writer, and its options are registered as keyed services under the handler's fully qualified name, so any number of channels — several over the same item type included — coexist in one container.

Hosting modes

The mode is resolved when the project compiles, not at runtime, so nothing changes behaviour at deploy time:

• InProcess — a generated BackgroundService drains the channel for the life of the host. Containers, VMs, worker services.
• InvocationLocal — a generated DrainAsync helper your Lambda entry point awaits, with a soft deadline taken from the remaining invocation time.
• Distributed — only the writer interface and DI helper are emitted; register your own transport-backed writer for a consumer that runs elsewhere.
• Default — InvocationLocal when the project references Kanject.Core.CloudFunction.Provider.AwsLambda, otherwise InProcess.

Drain an SQS batch in Lambda

The SQS adapter turns an SQS-triggered Lambda into a channel pipeline and projects the messages your handler did not acknowledge into SQSBatchResponse.BatchItemFailures, so SQS redelivers only the failures:

using Amazon.Lambda.Core;using Amazon.Lambda.SQSEvents;using Kanject.Core.Channels.Engine.Models;using Kanject.Core.Channels.Provider.AwsSqs;​public sealed class OrdersFunction{    private static OrderPlaced Deserialize(SQSEvent.SQSMessage m) =>        System.Text.Json.JsonSerializer.Deserialize<OrderPlaced>(m.Body)!;​    // One call drains the event and returns the partial-batch response.    public Task<SQSBatchResponse> HandleAsync(SQSEvent evt, ILambdaContext ctx) =>        SqsChannelDrainer.DrainAsync<OrderPlaced>(            evt,            ctx,            deserializer: Deserialize,            handler: async (batch, ct) =>            {                for (var i = 0; i < batch.Length; i++)                    await ProcessAsync(batch.Span[i].Payload, ct);            },            options: new ChannelOptions            {                BatchSize      = 10,                HandlerTimeout = TimeSpan.FromSeconds(15),                SafetyMargin   = TimeSpan.FromSeconds(2),            });}

The drain stops at the remaining invocation time minus SafetyMargin. A batch is either executed or reported un-acknowledged as a whole — never torn — which is what makes the partial-batch response safe to rely on. A Kinesis sibling, KinesisChannelDrainer, does the same for stream records.

What ships with it

• [Channel(Capacity, BatchSize, FlushIntervalMs, …)] on a method — generated writer interface, DI extension, and host-appropriate consumer
• Batch-size and time-based flush, per-batch retry with backoff (RetryCount / RetryBackoff), and restart-on-fault for hosted consumers
• BatchResult for partial success — fail specific items without throwing away the whole batch
• Deadline-aware Lambda draining via InvocationLocalDrainer, with un-acknowledged message ids returned for the trigger's failure response
• ChannelDrainer and ChannelHostedServiceBase<T> for using the engine directly, without the generator
• IChannelMetrics hooks for batch success, faults, handler timeouts, and queue depth
• Compile-time KANCHN diagnostics — a handler shape the generator cannot emit is a build error, never silently nothing