Skip to content
Prague
Improve this page

Host with DI

Goal: a host that consumes the right topics with the right settings, in which your own services get the caches injected and start only once the data is there.

Prerequisite: Prague.Kafka installed (Installation). The code is in HostWithDi.cs. The Program-shaped blocks are compiled-only; options binding, placeholder resolution and the registry run as tests, because building the container contacts no broker.

1. Generic host

public static IHost BuildWorker(string[] args) {
    var builder = Host.CreateApplicationBuilder(args);

    // Options come from the "kafkaCaches" section of IConfiguration (appsettings, env vars, ...).
    builder.Services.AddKafkaCaches(b => {
        b.AddCache<OrderCache, string, Order>();
        b.AddCache<CustomerCache, int, Customer>();
    });
    builder.Services.AddHostedService<OrderReportWorker>();  // starts after the caches are loaded: see below

    return builder.Build();                                  // host.Run(): StartAsync blocks until every topic hit EOF
}

AddKafkaCaches registers each AddCached cache as a singleton, the IDataCacheRegistry, KafkaCachesStatistics, a keyed KafkaCacheProducer and one hosted KafkaCachesBackgroundWorker. Its StartAsync returns after the initial load, so hosted services registered after it start on full caches when the host starts services in order.

2. ASP.NET Core

The overload with an IServiceProvider lets the options read other services; MapPragueEndpoints is covered in Inspect over HTTP.

public static WebApplication BuildApi(string[] args) {
    var builder = WebApplication.CreateBuilder(args);

    builder.Services.AddKafkaCaches("kafkaCaches",          // explicit section name; same default as above
        (o, sp) => o.ClientSettings["client.id"] = sp.GetRequiredService<IHostEnvironment>().ApplicationName,
        b => b.AddCache<OrderCache, string, Order>());

    var app = builder.Build();
    app.MapPragueEndpoints();                                // GET /prague, /prague/OrderCache?customerId=42
    app.MapGet("/orders/{id}", (string id, OrderCache orders)
        => orders.Cache.TryGet(id, out var order) ? Results.Ok(order) : Results.NotFound());
    return app;                                              // app.Run() serves only after the initial load
}

3. Options from configuration

The named KafkaCachesOptions are bound from the configuration section first ("kafkaCaches" unless you pass a name), then your delegate runs on top. Vars fills [v:name] placeholders in topic templates; [e:NAME] reads an environment variable.

// appsettings.json:  { "kafkaCaches": { "BootstrapServers": "kafka:9092",
//                      "ClientSettings": { "group.id": "order-api" }, "Vars": { "env": "staging" } } }
services.AddSingleton(Config(
    ("kafkaCaches:BootstrapServers", "kafka:9092"),
    ("kafkaCaches:ClientSettings:group.id", "order-api"),
    ("kafkaCaches:Vars:env", "staging")));

services.AddKafkaCaches(b => b.AddCache<OrderCache, string, Order>());   // binds "kafkaCaches" first ...

using var provider = services.BuildServiceProvider();
var options = provider.GetRequiredService<IOptionsMonitor<KafkaCachesOptions>>().Get("kafkaCaches");
var orders = provider.GetRequiredService<OrderCache>();

// ... and the [v:env] placeholder in [DataCacheTopic("orders.[v:env]")] resolved from Vars.
Console.WriteLine(orders.Topic);                         // orders.staging

services.AddKafkaCaches(
    o => {
        o.BootstrapServers = "from-code";                   // runs after the IConfiguration bind: code wins
        o.Vars["env"] = "prod";
    },
    b => b.AddCache<OrderCache, string, Order>("orders.[v:env].[e:HOSTNAME]"));   // [e:NAME] = environment variable

An unresolved placeholder is left in the string: the test AnUnresolvedPlaceholderIsLeftAsIs gets a topic literally named orders.[v:env]. Check cache.Topic at startup if your templates carry placeholders.

4. Several clusters

Each AddKafkaCaches(sectionName, …) call is a cluster: its own options, consumer and keyed producer. One hosted worker drives all of them. AddInternalCache puts a cache in the registry (so joins resolve and it can be written locally) without consuming a topic.

public static void TwoClusters(IServiceCollection services) {
    services.AddKafkaCaches("ordersCluster", b => b.AddCache<OrderCache, string, Order>());
    services.AddKafkaCaches("catalogCluster", b => {
        b.AddCache<ProductCache, int, Product>();
        b.AddInternalCache<ShipmentCache, string, Shipment>();   // in the registry and joinable, never consumed
    });
}

// One hosted worker drives every cluster; each section owns a producer, a KEYED singleton under the section name.
public sealed class OrderPublisher([FromKeyedServices("ordersCluster")] KafkaCacheProducer producer, OrderCache orders) {
    public void Place(Order order) => orders.AddOrUpdateAndProduce(order);   // local write + produce (only if changed)

    public void Cancel(string orderId) => orders.RemoveAndProduce(orderId);   // local remove + tombstone

    public void Mirror(Order order) => producer.Produce("orders.archive", order.OrderId, order);   // any topic
}

AddOrUpdateAndProduce writes locally and produces to the cache's topic only when the value actually changed; RemoveAndProduce removes locally and produces a tombstone.

5. Start work after the load

Derive from KafkaCachesBackgroundService instead of BackgroundService: its ExecuteAsync is called after the registry's LoadingCompletion, whatever the start order.

// Derive from KafkaCachesBackgroundService instead of BackgroundService: ExecuteAsync starts after the load.
public sealed class OrderReportWorker(IDataCacheRegistry registry, OrderCache orders) : KafkaCachesBackgroundService(registry) {
    protected override async Task ExecuteAsync(CancellationToken stoppingToken) {
        while (!stoppingToken.IsCancellationRequested) {
            Console.WriteLine($"{orders.Query().WithStatus(OrderStatus.Pending).Count()} pending orders");
            await Task.Delay(TimeSpan.FromMinutes(1), stoppingToken);
        }
    }
}

6. The registry

Only the caches you AddCache are DI services. Caches pulled in through foreign keys, and global indices, are reachable through IDataCacheRegistry; register a factory if you want them injected.

var registry = provider.GetRequiredService<IDataCacheRegistry>();
var orders = registry.GetCache<OrderCache>();           // same instance as GetRequiredService<OrderCache>()
var customers = registry.GetCache<CustomerCache>();     // pulled in by the FK: in the registry, NOT a DI service
var activity = registry.GetGlobalIndex<CustomerActivityIndex>();   // likewise

foreach (var info in registry.GetCaches())
    Console.WriteLine($"{info.Name}: {info.QueryableFields.Count} queryable fields");

7. Shutdown

public static async Task StopGracefully(IHost host) {
    // StopAsync cancels the consume loops and waits for each cache's live worker to drain its queue,
    // bounded by the token you pass (the host passes HostOptions.ShutdownTimeout, 30 s by default).
    using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
    await host.StopAsync(cts.Token);                         // anything still draining is logged by name, not thrown
}

Pitfalls

  • AddHostedService<MyWorker>() registered before AddKafkaCaches. It starts first, on empty caches. Register it after, or derive from KafkaCachesBackgroundService.
  • No IConfiguration in a hand-built ServiceCollection. The options binder needs one; add services.AddSingleton<IConfiguration>(new ConfigurationBuilder().Build()) as the tests do.
  • Expecting GetRequiredService<CustomerCache>() to work because Order references it. It throws unless you AddCache it or register a factory; see step 6.
  • Resolving the producer unkeyed with several clusters. It is a keyed singleton: [FromKeyedServices("sectionName")].
  • A group.id shared across replicas. Each replica must replay the whole topic; give each its own group or none. Kafka Integration explains the consumer model.
Prague documentation