# 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

```bash
dotnet add package Kanject.Core.Channels
dotnet 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:

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

> **Delete the `IncludeAssets` line:** `dotnet add package` writes an `IncludeAssets` line for the Annotations package that omits `compile`. Delete that line and keep `PrivateAssets="all"`, or the attribute types can be stripped from compilation and the generator emits nothing.

## 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:

```csharp
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

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

```csharp
// 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:

```csharp
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

> **Channels or Queue?:** They solve adjacent problems and coexist in one service. [Queue](https://www.kanject.com/docs/core-queue/) is the full SQS messaging layer — producer, consumers, queue provisioning, dead-letter handling. The Channels SQS adapter is receive-only and Lambda-first: it has no producer and owns no queues. Send with Queue; reach for Channels when you only need to drain a batch inside a Lambda, or want one handler to move between a hosted consumer and a function.

> **Free engine, licensed AWS adapters:** `Kanject.Core.Channels` and its generator are published on public NuGet and free under the Community licence — individuals, open source, education, and companies under $250K. The SQS and Kinesis adapters (`Kanject.Core.Channels.Provider.AwsSqs` / `.AwsKinesis`) ship from the private feed with a commercial Core licence.

---
_Source: https://www.kanject.com/docs/core-channels/ · Kanject Docs_