Emit

https://emitdotnet.github.io/Emit/
Admin
Emit is a .NET library providing reliable, ordered messaging with transactional outbox patterns,...
Tokens:30,348
Snippets:432
Trust Score:8.3
Update:5 months ago
Show doc for...
Tokens:
Raw
### Install Emit.FluentValidation Package

Source: https://emitdotnet.github.io/Emit/error-handling/validation

Install the integration package to enable FluentValidation support within Emit.

```bash
dotnet add package Emit.FluentValidation
```

--------------------------------

### Install Emit Packages (PackageReference)

Source: https://emitdotnet.github.io/Emit/getting-started/quickstart

Install the necessary Emit packages using PackageReference in your project file.

```xml
<PackageReference Include="Emit" />
<PackageReference Include="Emit.MongoDB" />
<PackageReference Include="Emit.Kafka" />
<PackageReference Include="Emit.Kafka.JsonSerializer" />
```

--------------------------------

### Install Emit Packages (.NET CLI)

Source: https://emitdotnet.github.io/Emit/getting-started/quickstart

Install the necessary Emit packages using the .NET CLI.

```bash
dotnet add package Emit
dotnet add package Emit.MongoDB
dotnet add package Emit.Kafka
dotnet add package Emit.Kafka.JsonSerializer
```

--------------------------------

### Add Emit.Persistence.EntityFrameworkCore Package

Source: https://emitdotnet.github.io/Emit/persistence/efcore

Install the necessary NuGet package for EF Core integration.

```bash
dotnet add package Emit.Persistence.EntityFrameworkCore

```

--------------------------------

### Adding Emit.Kafka.ProtobufSerializer Package

Source: https://emitdotnet.github.io/Emit/kafka/serialization

Install the Emit.Kafka.ProtobufSerializer NuGet package to enable Protobuf serialization.

```bash
dotnet add package Emit.Kafka.ProtobufSerializer
```

--------------------------------

### Add Emit OpenTelemetry Package

Source: https://emitdotnet.github.io/Emit/observability/tracing

Install the Emit.OpenTelemetry NuGet package to enable tracing instrumentation.

```bash
dotnet add package Emit.OpenTelemetry

```

--------------------------------

### Add Emit.Testing Package

Source: https://emitdotnet.github.io/Emit/advanced/testing

Install the Emit.Testing package to enable testing functionalities.

```bash
dotnet add package Emit.Testing

```

--------------------------------

### Add Emit.Kafka Package

Source: https://emitdotnet.github.io/Emit/kafka/setup

Install the Emit.Kafka NuGet package to enable Kafka integration.

```bash
dotnet add package Emit.Kafka
```

--------------------------------

### Adding Emit.Kafka.AvroSerializer Package

Source: https://emitdotnet.github.io/Emit/kafka/serialization

Install the Emit.Kafka.AvroSerializer NuGet package to enable Avro serialization.

```bash
dotnet add package Emit.Kafka.AvroSerializer
```

--------------------------------

### Selector Examples for Route Keys

Source: https://emitdotnet.github.io/Emit/kafka/routing

Examples of how to define a selector function to extract a route key from a message context. The route key type is inferred from the selector's return type and must be non-nullable.

```csharp
// String discriminator field
selector: ctx => ctx.Message.EventType
```

```csharp
// Enum value
selector: ctx => ctx.Message.OrderKind
```

```csharp
// Derived value
selector: ctx => ctx.Message.CorrelationId.StartsWith("B2B") ? "b2b" : "b2c"
```

--------------------------------

### Install Emit Health Check Packages

Source: https://emitdotnet.github.io/Emit/observability/health-checks

Install the health check packages that match your registered Emit providers.

```bash
dotnet add package Emit.Kafka.HealthChecks
dotnet add package Emit.MongoDB.HealthChecks
dotnet add package Emit.EntityFrameworkCore.HealthChecks
```

--------------------------------

### Structured log message examples

Source: https://emitdotnet.github.io/Emit/observability/logging

Examples of log messages with named placeholders that structured logging backends capture as queryable properties.

```plaintext
Retry {Attempt}/{MaxAttempts} failed for message from {Source}
Dead-lettered message from {Source} to {DlqDestination}
Validation failed for message from {Source}: {Errors}. Discarding.
Circuit breaker opened, pausing consumer group. Pause duration: {PauseDuration}
```

--------------------------------

### Add Emit Packages via NuGet

Source: https://emitdotnet.github.io/Emit/getting-started/installation

Install the core Emit library and additional packages for Kafka and MongoDB integration using the .NET CLI.

```bash
dotnet add package Emit
dotnet add package Emit.Kafka
dotnet add package Emit.MongoDB
```

--------------------------------

### Install Emit MongoDB Package

Source: https://emitdotnet.github.io/Emit/persistence/mongodb

Add the Emit.Persistence.MongoDB NuGet package to your project using the .NET CLI.

```bash
dotnet add package Emit.Persistence.MongoDB
```

--------------------------------

### Adding Emit.Kafka.JsonSerializer Package

Source: https://emitdotnet.github.io/Emit/kafka/serialization

Install the Emit.Kafka.JsonSerializer NuGet package to enable JSON Schema serialization.

```bash
dotnet add package Emit.Kafka.JsonSerializer
```

--------------------------------

### Implement Timing Middleware

Source: https://emitdotnet.github.io/Emit/advanced/custom-middleware

Example of a generic middleware that logs the time taken to handle a message. Replace ConsumeContext<T> with SendContext<T> for outbound middleware.

```csharp
public class TimingMiddleware<T> : IMiddleware<ConsumeContext<T>>
{
    private readonly ILogger<TimingMiddleware<T>> _logger;


    public TimingMiddleware(ILogger<TimingMiddleware<T>> logger)
    {
        _logger = logger;
    }


    public async Task InvokeAsync(ConsumeContext<T> context, IMiddlewarePipeline<ConsumeContext<T>> next)
    {
        var sw = Stopwatch.StartNew();
        try
        {
            await next.InvokeAsync(context);
        }
        finally
        {
            _logger.LogDebug("Message handled in {ElapsedMs}ms", sw.ElapsedMilliseconds);
        }
    }
}
```

--------------------------------

### Minimal Kafka Registration

Source: https://emitdotnet.github.io/Emit/kafka/setup

Basic setup for Kafka integration, including broker connection and a single topic registration with producer and consumer group configuration.

```csharp
builder.Services.AddEmit(emit =>
{
    emit.AddKafka(kafka =>
    {
        kafka.ConfigureClient(cfg =>
        {
            cfg.BootstrapServers = "localhost:9092";
        });


        kafka.Topic<string, PizzaOrdered>("pizzas", topic =>
        {
            topic.SetKeySerializer(Confluent.Kafka.Serializers.Utf8);
            topic.SetValueSerializer(/* your serializer */);
            topic.SetKeyDeserializer(Confluent.Kafka.Deserializers.Utf8);
            topic.SetValueDeserializer(/* your deserializer */);


            topic.Producer();


            topic.ConsumerGroup("pizza-kitchen", group =>
            {
                group.AddConsumer<PizzaOrderedConsumer>();
            });
        });
    });
});
```

--------------------------------

### Configure Worker Count and Batching

Source: https://emitdotnet.github.io/Emit/kafka/batch-consuming

Set the WorkerCount for the consumer group and configure batch sizing with MaxSize and Timeout. This example sets WorkerCount to 4, MaxSize to 100, and Timeout to 1 second.

```csharp
group.WorkerCount = 4;
group.AddBatchConsumer<ScanBatchConsumer>(batch =>
{
    batch.MaxSize = 100;
    batch.Timeout = TimeSpan.FromSeconds(1);
});

```

--------------------------------

### Register DbContextFactory and Emit Services

Source: https://emitdotnet.github.io/Emit/persistence/efcore

Configure DbContextFactory for your application's DbContext and register Emit's EF Core services. UseNpgsql() is required for PostgreSQL-specific mappings. UseOutbox() starts the background worker and routes producers through the outbox by default.

```csharp
builder.Services.AddDbContextFactory<AppDbContext>(options =>
    options.UseNpgsql(connectionString));


builder.Services.AddEmit(emit =>
{
    emit.AddEntityFrameworkCore<AppDbContext>(ef =>
    {
        ef.UseNpgsql()
          .UseOutbox()
          .UseDistributedLock();
    });
});

```

--------------------------------

### Register Emit with MongoDB and Kafka

Source: https://emitdotnet.github.io/Emit/getting-started/quickstart

Register Emit services, configuring MongoDB for the outbox and distributed locking, and Kafka for message production and consumption. This setup includes configuring Kafka client, schema registry, and defining the 'pizzas' topic with serializers and a consumer group.

```csharp
using MongoDB.Driver;


builder.Services.AddSingleton<IMongoClient>(
    new MongoClient("mongodb://localhost:27017"));


builder.Services.AddEmit(emit =>
{
    emit.AddMongoDb(mongo =>
    {
        mongo.Configure((sp, ctx) =>
        {
            ctx.Client = sp.GetRequiredService<IMongoClient>();
            ctx.Database = ctx.Client.GetDatabase("pizzeria-db");
        })
        .UseOutbox()
        .UseDistributedLock();
    });


    emit.AddKafka(kafka =>
    {
        kafka.ConfigureClient(cfg =>
        {
            cfg.BootstrapServers = "localhost:9092";
        });


        kafka.ConfigureSchemaRegistry(cfg =>
        {
            cfg.Url = "localhost:8081";
        });


        kafka.AutoProvision();


        kafka.Topic<string, PizzaOrdered>("pizzas", topic =>
        {
            topic.SetKeySerializer(Confluent.Kafka.Serializers.Utf8);
            topic.SetJsonSchemaValueSerializer();
            topic.SetKeyDeserializer(Confluent.Kafka.Deserializers.Utf8);
            topic.SetJsonSchemaValueDeserializer();


            topic.Producer();


            topic.ConsumerGroup("pizza-kitchen", group =>
            {
                group.AddConsumer<PizzaOrderedConsumer>();
            });
        });
    });
});
```

--------------------------------

### Setting Explicit Serializers and Deserializers

Source: https://emitdotnet.github.io/Emit/kafka/serialization

Assign serializers and deserializers per topic using the Topic builder. This example shows explicit assignment using Confluent.Kafka serializers.

```csharp
kafka.Topic<string, OrderPlaced>("orders", topic =>
{
    topic.SetKeySerializer(Confluent.Kafka.Serializers.Utf8);
    topic.SetValueSerializer(myValueSerializer);


    topic.SetKeyDeserializer(Confluent.Kafka.Deserializers.Utf8);
    topic.SetValueDeserializer(myValueDeserializer);
});
```

--------------------------------

### Register Mediator and Handlers

Source: https://emitdotnet.github.io/Emit/mediator

Register the mediator and its handlers using `AddEmit` and `AddMediator` extension methods. This setup is independent of other Emit services like Kafka.

```csharp
builder.Services.AddEmit(emit =>
{
    emit.AddMediator(mediator =>
    {
        mediator.AddHandler<PlacePizzaOrderHandler>();
        mediator.AddHandler<GetPizzaStatusHandler>();
    });
});
```

--------------------------------

### Enable Auto-Provisioning

Source: https://emitdotnet.github.io/Emit/advanced/configuration

Enable automatic topic creation at startup. Per-topic creation options like partition count, replication factor, and message retention can be configured.

```csharp
kafka.AutoProvision();
```

--------------------------------

### Configure Outbox with MongoDB and Kafka

Source: https://emitdotnet.github.io/Emit/outbox

Enable the outbox infrastructure by calling `UseOutbox()` on your persistence provider. Producers route through the outbox by default, but `UseDirect()` can be used to bypass it for specific producers.

```csharp
emit.AddMongoDb(mongo =>
{
    mongo.Configure((sp, ctx) => { ... })
         .UseOutbox(options =>
         {
             options.PollingInterval = TimeSpan.FromSeconds(5);  // default
             options.BatchSize = 100;                            // default
         });
});


emit.AddKafka(kafka =>
{
    kafka.ConfigureClient(cfg => cfg.BootstrapServers = "localhost:9092");


    kafka.Topic<string, PizzaOrdered>("pizzas", topic =>
    {
        topic.Producer();  // routes through outbox by default
    });


    kafka.Topic<string, AnalyticsEvent>("analytics", topic =>
    {
        topic.Producer(p => p.UseDirect());  // bypasses the outbox
    });
});
```

--------------------------------

### Auto-provisioning Topics with Custom Settings

Source: https://emitdotnet.github.io/Emit/kafka/setup

Enable auto-provisioning for Kafka topics, allowing Emit to create missing topics. Configure specific provisioning options like number of partitions, replication factor, retention, and compression.

```csharp
emit.AddKafka(kafka =>
{
    kafka.AutoProvision();


    kafka.Topic<string, PizzaOrdered>("pizzas", topic =>
    {
        topic.Provisioning(options =>
        {
            options.NumPartitions = 6;
            options.ReplicationFactor = 3;
            options.Retention = TimeSpan.FromDays(14);
            options.CompressionType = TopicCompressionType.Zstd;
        });


        // ... serializers, producer, consumer groups ...
    });
});
```

--------------------------------

### Map Health Checks with Tag Filtering

Source: https://emitdotnet.github.io/Emit/observability/health-checks

Filter health checks by tags to separate liveness from readiness probes, for example, in Kubernetes.

```csharp
// Liveness: is the process alive?
app.MapHealthChecks("/health/live");


// Readiness: are downstream services reachable?
app.MapHealthChecks("/health/ready", new HealthCheckOptions
{
    Predicate = check => check.Tags.Contains("emit")
});
```

--------------------------------

### Configure DLQ Topic Provisioning

Source: https://emitdotnet.github.io/Emit/error-handling/dead-letter-queue

Configure provisioning options for the dead-letter topic, such as the number of partitions and retention period. This allows the DLQ topic to participate in topic verification and auto-provisioning.

```csharp
kafka.DeadLetter("orders.dlt", dlq =>
{
    dlq.Provisioning(options =>
    {
        options.NumPartitions = 3;
        options.Retention = TimeSpan.FromDays(30);
    });
});
```

--------------------------------

### Extracting Transport Scheme from Destination Address

Source: https://emitdotnet.github.io/Emit/advanced/feature-collection

Gets the transport scheme (e.g., "kafka") from a destination address URI.

```csharp
// Get the transport scheme (e.g. "kafka")
var provider = EmitEndpointAddress.GetScheme(context.DestinationAddress);
```

--------------------------------

### Class-Based Validator Implementation

Source: https://emitdotnet.github.io/Emit/error-handling/validation

Implement IMessageValidator<T> for reusable validation logic that can be unit tested independently. This example shows validation for CustomerId and Amount.

```csharp
public class OrderValidator : IMessageValidator<OrderCreated>
{
    public Task<MessageValidationResult> ValidateAsync(
        OrderCreated message, CancellationToken cancellationToken)
    {
        if (string.IsNullOrWhiteSpace(message.CustomerId))
            return Task.FromResult(MessageValidationResult.Fail("CustomerId is required"));


        if (message.Amount <= 0)
            return Task.FromResult(MessageValidationResult.Fail("Amount must be positive"));


        return Task.FromResult(MessageValidationResult.Success);
    }
}
```

--------------------------------

### Implement Kafka Consumer Lifecycle Events

Source: https://emitdotnet.github.io/Emit/kafka/consumers

Register IKafkaConsumerObserver to receive Kafka-specific lifecycle events such as consumer start/stop, partition rebalances, and offset commit errors. Implement only the methods required for your specific needs.

```csharp
kafka.AddConsumerObserver<MyKafkaObserver>();
```

--------------------------------

### Selectively Enable/Disable Emit Meters

Source: https://emitdotnet.github.io/Emit/observability/metrics

Customize which Emit meters are enabled. By default, all meters are enabled. This example disables the Kafka broker statistics meter.

```csharp
metrics.AddEmitInstrumentation(options =>
{
    options.EnableKafkaBrokerMeter = false;  // skip librdkafka statistics
});
```

--------------------------------

### Extracting Entity Name from Destination Address

Source: https://emitdotnet.github.io/Emit/advanced/feature-collection

Gets the entity name (e.g., topic name "orders") from a destination address URI.

```csharp
// Get the entity name (e.g. topic name "orders")
var destination = EmitEndpointAddress.GetEntityName(context.DestinationAddress);
```

--------------------------------

### Register a Consumer with a Topic

Source: https://emitdotnet.github.io/Emit/kafka/consumers

Register a consumer implementation for a specific topic within a consumer group. This configuration sets up the message handling pipeline.

```csharp
kafka.Topic<string, PizzaOrdered>("pizzas", topic =>
{
    topic.ConsumerGroup("pizza-kitchen", group =>
    {
        group.AddConsumer<PizzaOrderedConsumer>();
    });
});
```

--------------------------------

### Test Message Sink and Consumer

Source: https://emitdotnet.github.io/Emit/advanced/testing

Use MessageSink and SinkConsumer to test message consumption logic in memory. This setup allows you to verify that messages are correctly processed by your consumer.

```csharp
[Fact]
public async Task Consumer_writes_to_sink()
{
    // Arrange
    var sink = new MessageSink<PizzaOrdered>();
    var services = new ServiceCollection();


    services.AddEmit(emit =>
    {
        emit.AddKafka(kafka =>
        {
            kafka.ConfigureClient(cfg => cfg.BootstrapServers = "localhost:9092");


            kafka.Topic<string, PizzaOrdered>("pizzas", topic =>
            {
                topic.SetKeyDeserializer(Confluent.Kafka.Deserializers.Utf8);
                topic.SetValueDeserializer(myDeserializer);


                topic.ConsumerGroup("test-group", group =>
                {
                    group.AddConsumer<SinkConsumer<PizzaOrdered>>();
                });
            });
        });
    });


    services.AddSingleton(sink);


    // Act: produce a message using your real producer or inject one directly


    // Assert
    var context = await sink.WaitForMessageAsync(TimeSpan.FromSeconds(10));
    Assert.Equal(expectedPizzaId, context.Message.PizzaId);
}

```

--------------------------------

### Implement IBatchConsumer

Source: https://emitdotnet.github.io/Emit/kafka/batch-consuming

Implement IBatchConsumer<TValue> to receive and process a batch of messages in a single invocation. Useful for reducing per-message overhead or when domain logic benefits from batch processing.

```csharp
public class ScanBatchConsumer(ILogger<ScanBatchConsumer> logger)
    : IBatchConsumer<PackageScan>
{
    public async Task ConsumeAsync(
        ConsumeContext<MessageBatch<PackageScan>> context,
        CancellationToken cancellationToken)
    {
        var batch = context.Message;
        logger.LogInformation("Processing {Count} scans", batch.Count);


        foreach (var item in batch.Items)
        {
            var scan = item.Message;       // deserialized payload
            var tc = item.TransportContext; // raw bytes, headers, partition, offset
            // process...
        }
    }
}

```

--------------------------------

### Match Specific Exception Types and Subclasses

Source: https://emitdotnet.github.io/Emit/error-handling/error-policies

Demonstrates matching specific exception types and their subclasses using `When<TException>`. Place more specific types before broader ones.

```csharp
group.OnError(err =>
{
    // Matches HttpRequestException and any derived types
    err.When<HttpRequestException>(action => action
        .Retry(3, Backoff.Exponential(TimeSpan.FromSeconds(1)))
        .DeadLetter());


    // Matches InvalidOperationException and subclasses
    err.When<InvalidOperationException>(action => action.Discard());


    err.Default(action => action.DeadLetter());
});
```

--------------------------------

### Implement Transactional Consumer

Source: https://emitdotnet.github.io/Emit/kafka/consumers

Decorates a consumer class with [Transactional] to wrap invocations in a unit-of-work transaction. Outbox entries commit atomically with business data writes. Each retry gets a fresh transaction.

```csharp
[Transactional]
public sealed class PizzaOrderedConsumer(
    IEventProducer<string, PizzaDispatched> producer) : IConsumer<PizzaOrdered>
{
    public async Task ConsumeAsync(ConsumeContext<PizzaOrdered> context, CancellationToken ct)
    {
        await producer.ProduceAsync(..., ct);
        // Transaction commits on success, rolls back on failure.
    }
}
```

--------------------------------

### Produce Message within a Transaction (MongoDB)

Source: https://emitdotnet.github.io/Emit/getting-started/quickstart

Produce a message within a transaction using IUnitOfWork to ensure atomicity between business writes and outbox writes. This example inserts an order into MongoDB and produces a PizzaOrdered event.

```csharp
app.MapPost("/orders", async (
    PlacePizzaOrderRequest request,
    IUnitOfWork unitOfWork,
    IMongoSessionAccessor sessionAccessor,
    IMongoDatabase database,
    IEventProducer<string, PizzaOrdered> producer,
    CancellationToken ct) =>
{
    var pizza = new PizzaOrdered(
        Guid.NewGuid(), request.CustomerId,
        request.Toppings, request.DeliveryAddress);


    await using var tx = await unitOfWork.BeginAsync(ct);
    var session = sessionAccessor.Session !;
    await database.GetCollection<PizzaOrdered>("orders")
        .InsertOneAsync(session, pizza, cancellationToken: ct);
    await producer.ProduceAsync(
        new EventMessage<string, PizzaOrdered>(pizza.PizzaId.ToString(), pizza), ct);
    await tx.CommitAsync(ct);


    return Results.Accepted();
});
```

--------------------------------

### Topic Registration with Serializers and Consumer

Source: https://emitdotnet.github.io/Emit/kafka/setup

Define a Kafka topic with specific key and value serializers/deserializers, and configure it for both producing messages and consuming them with a specified consumer group.

```csharp
kafka.Topic<string, PizzaOrdered>("pizzas", topic =>
{
    topic.SetKeySerializer(Confluent.Kafka.Serializers.Utf8);
    topic.SetValueSerializer(mySerializer);


    topic.SetKeyDeserializer(Confluent.Kafka.Deserializers.Utf8);
    topic.SetValueDeserializer(myDeserializer);


    topic.Producer();


    topic.ConsumerGroup("my-group", group =>
    {
        group.AddConsumer<MyConsumer>();
    });
});
```

--------------------------------

### Register Global Outbox Observer

Source: https://emitdotnet.github.io/Emit/observability/observers

Register an IOutboxObserver implementation globally to observe outbox entry lifecycle events.

```csharp
emit.AddOutboxObserver<MyOutboxObserver>();
```

--------------------------------

### Configure Batch Rate Limiting and Max Size

Source: https://emitdotnet.github.io/Emit/kafka/batch-consuming

Sets up rate limiting for a batch consumer using a token bucket strategy and defines the maximum batch size. Ensure burst size is at least as large as MaxSize to avoid rejections.

```csharp
group.RateLimit(rl => rl.TokenBucket(permitsPerSecond: 500, burstSize: 200));
group.AddBatchConsumer<ScanBatchConsumer>(batch => batch.MaxSize = 100);
```

--------------------------------

### Configure Consumer Worker Pool and Buffer Size

Source: https://emitdotnet.github.io/Emit/kafka/consumers

Configure the number of concurrent workers and the buffer size for each worker within a consumer group. Matching WorkerCount to the number of partitions assigned to the instance is recommended for optimal parallelism.

```csharp
topic.ConsumerGroup("pizza-kitchen", group =>
{
    group.WorkerCount = 3;
    group.BufferSize = 32;  // per-worker channel capacity


    group.AddConsumer<PizzaOrderedConsumer>();
});
```

--------------------------------

### Implement Produce Observer

Source: https://emitdotnet.github.io/Emit/kafka/producers

Implement the IProduceObserver interface to hook into the produce lifecycle. The OnProducingAsync, OnProducedAsync, and OnProduceErrorAsync methods provide callbacks for different stages of the produce operation.

```csharp
public class ProduceAuditObserver : IProduceObserver
{
    public Task OnProducingAsync<T>(SendContext<T> context) { ... }
    public Task OnProducedAsync<T>(SendContext<T> context) { ... }
    public Task OnProduceErrorAsync<T>(SendContext<T> context, Exception exception) { ... }
}
```

--------------------------------

### Add cross-cutting log context with middleware

Source: https://emitdotnet.github.io/Emit/observability/logging

Implement `IMiddleware` to open a scope once for cross-cutting log context, such as adding a TenantId to all log messages.

```csharp
public class TenantLoggingMiddleware<T>(ILogger<TenantLoggingMiddleware<T>> logger, ITenantResolver tenant)
    : IMiddleware<ConsumeContext<T>>
{
    public async Task InvokeAsync(ConsumeContext<T> context, IMiddlewarePipeline<ConsumeContext<T>> next)
    {
        var tenantId = await tenant.ResolveAsync(context, context.CancellationToken);
        using var scope = logger.BeginScope(new { TenantId = tenantId });
        await next.InvokeAsync(context);
    }
}
```

--------------------------------

### Implement PizzaOrderedConsumer

Source: https://emitdotnet.github.io/Emit/getting-started/quickstart

Implement the IConsumer interface to handle incoming PizzaOrdered messages. This consumer logs the details of a new pizza order.

```csharp
public class PizzaOrderedConsumer(ILogger<PizzaOrderedConsumer> logger)
    : IConsumer<PizzaOrdered>
{
    public Task ConsumeAsync(
        ConsumeContext<PizzaOrdered> context,
        CancellationToken cancellationToken)
    {
        var pizza = context.Message;
        logger.LogInformation(
            "New order {PizzaId} for {CustomerId}: {Toppings}",
            pizza.PizzaId,
            pizza.CustomerId,
            string.Join(", ", pizza.Toppings));


        return Task.CompletedTask;
    }
}
```

--------------------------------

### Configure Batch Sizing

Source: https://emitdotnet.github.io/Emit/kafka/batch-consuming

Control batch sizing by passing a configuration delegate to AddBatchConsumer<T>(). Configure MaxSize for the maximum number of messages per batch and Timeout for the maximum wait time before dispatching a partial batch.

```csharp
group.AddBatchConsumer<ScanBatchConsumer>(batch =>
{
    batch.MaxSize = 500;
    batch.Timeout = TimeSpan.FromSeconds(2);
});

```

--------------------------------

### Configure Tracer Provider with Emit

Source: https://emitdotnet.github.io/Emit/observability/tracing

Register Emit's tracing instrumentation and an OTLP exporter with the OpenTelemetry tracer provider in your application's services.

```csharp
builder.Services.AddOpenTelemetry()
    .WithTracing(tracing =>
    {
        tracing.AddEmitInstrumentation();
        tracing.AddOtlpExporter();  // or your preferred exporter
    });

```

--------------------------------

### Configure DLQ Alerting Consumer

Source: https://emitdotnet.github.io/Emit/error-handling/dead-letter-queue

Register a consumer group directly on the DLQ topic to automate inspection or alerting. This is useful for feeding DLQ entries into a monitoring system, but the replay decision should remain a human one.

```csharp
kafka.DeadLetter("orders.dlt", dlq =>
{
    dlq.ConsumerGroup("dlq-alerting", group =>
    {
        group.AddConsumer<DlqAlertConsumer>();
    });
});

```

--------------------------------

### Register Emit Services with Kafka and MongoDB

Source: https://emitdotnet.github.io/Emit/getting-started/installation

Configure and register Emit services, including MongoDB outbox and distributed lock, and Kafka producers/consumers with specific topic and group configurations.

```csharp
builder.Services.AddEmit(emit =>
{
    emit.AddMongoDb(mongo =>
    {
        mongo.Configure((sp, ctx) =>
        {
            ctx.Client = sp.GetRequiredService<IMongoClient>();
            ctx.Database = ctx.Client.GetDatabase("my-db");
        })
        .UseOutbox()
        .UseDistributedLock();
    });


    emit.AddKafka(kafka =>
    {
        kafka.ConfigureClient(cfg =>
        {
            cfg.BootstrapServers = "localhost:9092";
        });


        kafka.Topic<string, OrderPlaced>("orders", topic =>
        {
            topic.Producer();
            topic.ConsumerGroup("order-processor", group =>
            {
                group.AddConsumer<OrderPlacedConsumer>();
            });
        });
    });
});

```

--------------------------------

### Implement a Custom Distributed Lock Provider

Source: https://emitdotnet.github.io/Emit/persistence/distributed-locks

Implement a custom distributed lock provider by inheriting from `DistributedLockProviderBase`. You only need to implement `TryAcquireCoreAsync`, `ReleaseCoreAsync`, and `ExtendCoreAsync` for atomic operations.

```csharp
public class RedisDistributedLockProvider : DistributedLockProviderBase
{
    protected override Task<IDistributedLockHandle?> TryAcquireCoreAsync(
        string key, TimeSpan ttl, CancellationToken ct)
    {
        // Single atomic acquire attempt. Return a handle on success, null on failure.
    }


    protected override Task ReleaseCoreAsync(string key, CancellationToken ct)
    {
        // Release the lock for the given key.
    }


    protected override Task<bool> ExtendCoreAsync(string key, TimeSpan ttl, CancellationToken ct)
    {
        // Extend the lock's TTL. Return true if still held, false if lost.
    }
}
```

--------------------------------

### Factory-Based Middleware Registration

Source: https://emitdotnet.github.io/Emit/advanced/custom-middleware

Manually construct middleware using a factory function that receives the IServiceProvider.

```csharp
group.Use<OrderPlaced>(sp => new CustomMiddleware<OrderPlaced>(
    sp.GetRequiredService<ILogger<CustomMiddleware<OrderPlaced>>>(),
    sp.GetRequiredService<IFeatureFlags>()));
```

--------------------------------

### Default Batch Configuration

Source: https://emitdotnet.github.io/Emit/kafka/batch-consuming

When batch sizing configuration is omitted, MaxSize defaults to 100 messages and Timeout defaults to 5 seconds.

```csharp
group.AddBatchConsumer<ScanBatchConsumer>(); // MaxSize = 100, Timeout = 5s

```

--------------------------------

### Configure Emit Health Checks with Options

Source: https://emitdotnet.github.io/Emit/observability/health-checks

Customize health check behavior using optional parameters like name, failureStatus, tags, and timeout.

```csharp
builder.Services.AddHealthChecks()
    .AddEmitKafka(
        name: "emit-kafka",
        failureStatus: HealthStatus.Degraded,
        tags: ["messaging"],
        timeout: TimeSpan.FromSeconds(10))
    .AddEmitMongoDB(
        name: "emit-mongodb",
        failureStatus: HealthStatus.Degraded,
        tags: ["database"]);
```

--------------------------------

### Setting Shorthand Serializers and Deserializers

Source: https://emitdotnet.github.io/Emit/kafka/serialization

Use shorthand methods to set common serializers and deserializers for string keys and values, avoiding direct reference to Confluent.Kafka namespaces.

```csharp
kafka.Topic<string, OrderPlaced>("orders", topic =>
{
    topic.SetUtf8KeySerializer();
    topic.SetUtf8KeyDeserializer();
});
```

--------------------------------

### Configuring Async Serializers and Deserializers

Source: https://emitdotnet.github.io/Emit/kafka/serialization

Use the corresponding overloads for async serializers (IAsyncSerializer<T>) and deserializers (IAsyncDeserializer<T>).

```csharp
topic.SetValueSerializer(myAsyncSerializer);
topic.SetValueDeserializer(myAsyncDeserializer);
```

--------------------------------

### Configure Outbox Worker Options

Source: https://emitdotnet.github.io/Emit/persistence/mongodb

Tune the behavior of the outbox background worker by providing a configuration action to `UseOutbox()`. Options include setting the polling interval and batch size for processing outbox entries.

```csharp
.UseOutbox(options =>
{
    options.PollingInterval = TimeSpan.FromSeconds(3);
    options.BatchSize = 200;
})
```

--------------------------------

### Configure Kafka Client Additional Properties

Source: https://emitdotnet.github.io/Emit/advanced/configuration

Use AdditionalProperties to set librdkafka settings not exposed as typed properties. Typed properties take precedence. Invalid keys will cause an exception at build time.

```csharp
kafka.ClientConfig(client =>
{
    client.AdditionalProperties = new Dictionary<string, string>
    {
        ["socket.nagle.disable"] = "true"
    };
});

```

--------------------------------

### Apply Provider-Level Middleware for Kafka

Source: https://emitdotnet.github.io/Emit/kafka/setup

Register middleware on `kafka.InboundPipeline` or `kafka.OutboundPipeline` to apply it to all Kafka consumers or producers, respectively. This configures the Provider level of the pipeline model.

```csharp
emit.AddKafka(kafka =>
{
    kafka.InboundPipeline.Use<TracingMiddleware>();
    kafka.OutboundPipeline.Use<SigningMiddleware>();
});
```

--------------------------------

### Constructing EventMessage with Headers

Source: https://emitdotnet.github.io/Emit/kafka/producers

Build an `EventMessage` with a key, value, and custom headers. Headers are forwarded to Kafka as-is, and Emit automatically appends W3C trace context headers when OpenTelemetry is enabled.

```csharp
// With headers
var msg = new EventMessage<string, PizzaOrdered>(
    pizza.PizzaId.ToString(),
    pizza,
    [new KeyValuePair<string, string>("x-tenant-id", tenantId)]);
```

--------------------------------

### Register Per-Consumer Middleware

Source: https://emitdotnet.github.io/Emit/advanced/custom-middleware

Register middleware specifically for a single consumer within a consumer group.

```csharp
group.AddConsumer<OrderPlacedConsumer>(consumer =>
{
    consumer.Use<ConsumerSpecificMiddleware<OrderPlaced>>();
});
```

--------------------------------

### Setting Protobuf Value Serializer and Deserializer

Source: https://emitdotnet.github.io/Emit/kafka/serialization

Configure the topic to use Protobuf serialization and deserialization for message values. The TValue must implement IMessage<TValue>.

```csharp
kafka.Topic<string, MyProtoMessage>("events", topic =>
{
    topic.SetProtobufValueSerializer();
    topic.SetProtobufValueDeserializer();
});
```

--------------------------------

### Configure Per-Route Middleware

Source: https://emitdotnet.github.io/Emit/kafka/routing

Apply specific middleware to individual routes within a router. This middleware will only wrap the handler for the configured route.

```csharp
router.Route<DeepDishHandler>("deep-dish", route =>
{
    route.Use<PriorityTracingMiddleware>();
});
```

--------------------------------

### Generate EF Core Migrations for Emit Tables

Source: https://emitdotnet.github.io/Emit/persistence/efcore

Manually add and apply migrations for Emit's schema changes. Run these commands after adding AddEmitModel to your DbContext.

```bash
dotnet ef migrations add AddEmitTables
dotnet ef database update

```

--------------------------------

### Add MongoDB Persistence to Emit

Source: https://emitdotnet.github.io/Emit/persistence/mongodb

Register MongoDB persistence for outbox and distributed locks. Ensure your IMongoClient is registered before calling AddMongoDb. Configure the client and database within the provided callback.

```csharp
// Register your IMongoClient however you normally would.
builder.Services.AddSingleton<IMongoClient>(new MongoClient("mongodb://localhost:27017"));


builder.Services.AddEmit(emit =>
{
    emit.AddMongoDb(mongo =>
    {
        mongo.Configure((sp, ctx) =>
        {
            ctx.Client = sp.GetRequiredService<IMongoClient>();
            ctx.Database = ctx.Client.GetDatabase("pizzeria-db");
        })
        .UseOutbox()
        .UseDistributedLock();
    });
});
```

--------------------------------

### Registering and Using FluentValidation

Source: https://emitdotnet.github.io/Emit/error-handling/validation

Register FluentValidation validators in DI and configure the consumer group to use ValidateWithFluentValidation. Invalid messages will be dead-lettered.

```csharp
services.AddValidatorsFromAssemblyContaining<OrderCreatedValidator>();


services.AddEmit(emit =>
{
    emit.AddKafka(kafka =>
    {
        kafka.Topic<string, OrderCreated>("orders", topic =>
        {
            topic.ConsumerGroup("order-processors", group =>
            {
                group.ValidateWithFluentValidation(onFailure => onFailure.DeadLetter());
                group.AddConsumer<OrderCreatedHandler>();
            });
        });
    });
});
```

--------------------------------

### Add log context in consumer using scopes

Source: https://emitdotnet.github.io/Emit/observability/logging

Wrap consumer logic in a scope using `ILogger.BeginScope` to add message-specific context like OrderId and CustomerId to logs.

```csharp
public class OrderPlacedConsumer(ILogger<OrderPlacedConsumer> logger) : IConsumer<OrderPlaced>
{
    public async Task ConsumeAsync(ConsumeContext<OrderPlaced> context, CancellationToken ct)
    {
        using var scope = logger.BeginScope(new Dictionary<string, object>
        {
            ["OrderId"] = context.Message.OrderId,
            ["CustomerId"] = context.Message.CustomerId,
        });


        logger.LogInformation("Processing order");
        // ...
    }
}
```

--------------------------------

### Configure Emit log level

Source: https://emitdotnet.github.io/Emit/observability/logging

Set the default log level for all .NET logging and a specific level for Emit. Use 'Warning' to suppress verbose logs in development.

```json
{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Emit": "Warning"
    }
  }
}
```

--------------------------------

### Implement Mediator Observer

Source: https://emitdotnet.github.io/Emit/mediator

Implement `IMediatorObserver` to receive lifecycle hooks (`OnHandlingAsync`, `OnHandledAsync`, `OnHandleErrorAsync`) without writing middleware. Observer exceptions are logged and swallowed.

```csharp
public class AuditObserver : IMediatorObserver
{
    public Task OnHandlingAsync<T>(MediatorContext<T> context)
    {
        return Task.CompletedTask;
    }


    public Task OnHandledAsync<T>(MediatorContext<T> context)
    {
        return Task.CompletedTask;
    }


    public Task OnHandleErrorAsync<T>(MediatorContext<T> context, Exception exception)
    {
        return Task.CompletedTask;
    }
}
```

```csharp
mediator.AddObserver<AuditObserver>();
```

--------------------------------

### Kafka Client Configuration

Source: https://emitdotnet.github.io/Emit/kafka/setup

Configure client-wide settings for Kafka producers and consumers, including broker addresses, client ID, and security protocols like SASL/SSL.

```csharp
kafka.ConfigureClient(cfg =>
{
    cfg.BootstrapServers = "broker1:9092,broker2:9092";
    cfg.ClientId = "my-service";


    // SASL/SSL
    cfg.SecurityProtocol = SecurityProtocol.SaslSsl;
    cfg.SaslMechanism = SaslMechanism.Plain;
    cfg.SaslUsername = "user";
    cfg.SaslPassword = "pass";
    cfg.SslCaLocation = "/etc/ssl/certs/ca-bundle.crt";
});
```

--------------------------------

### Configure Emit Tracing Options

Source: https://emitdotnet.github.io/Emit/observability/tracing

Customize Emit's tracing behavior, such as enabling spans, controlling root activity creation, and propagating baggage, using the ConfigureTracing method.

```csharp
builder.Services.AddEmit(emit =>
{
    emit.ConfigureTracing(tracing =>
    {
        tracing.Configure(options =>
        {
            options.Enabled = true;               // set false to suppress all Emit spans
            options.CreateRootActivities = true;  // false = only trace within an existing span
            options.PropagateBaggage = true;      // propagate W3C baggage across message boundaries
            options.MaxBaggageSizeBytes = 8192;   // excess items are dropped with a warning
        });
    });
});

```

--------------------------------

### Schema Registry Configuration

Source: https://emitdotnet.github.io/Emit/kafka/setup

Configure the Schema Registry connection details, including the URL and authentication information, for use with Avro, JSON Schema, or Protobuf serializers.

```csharp
kafka.ConfigureSchemaRegistry(cfg =>
{
    cfg.Url = "http://schema-registry:8081";
    cfg.BasicAuthUserInfo = "user:password";
});
```

--------------------------------

### Configure Dead Letter Queue

Source: https://emitdotnet.github.io/Emit/advanced/configuration

Set up a dead-letter queue for failed messages. This allows for later inspection or reprocessing of messages that could not be handled successfully. Options include specifying the DLQ topic name and provisioning settings.

```csharp
kafka.DeadLetter("orders.dlt");
```

```csharp
// With provisioning options:
kafka.DeadLetter("orders.dlt", dlq =>
{
    dlq.Provisioning(options =>
    {
        options.Retention = TimeSpan.FromDays(30);
    });
});
```

--------------------------------

### Configure Rate Limiter for Consumer Group

Source: https://emitdotnet.github.io/Emit/error-handling/resilience

Implement rate limiting for a consumer group using the token bucket algorithm. This throttles message processing to protect downstream services from being overwhelmed.

```csharp
topic.ConsumerGroup("order-processor", group =>
{
    group.RateLimit(rl =>
    {
        rl.TokenBucket(permitsPerSecond: 100, burstSize: 100);
    });


    group.AddConsumer<OrderPlacedConsumer>();
});

```

--------------------------------

### Registering Per-Producer Middleware

Source: https://emitdotnet.github.io/Emit/kafka/producers

Add custom middleware that will execute only for a specific topic's producer pipeline. This middleware operates at the Leaf level of the Emit pipeline model.

```csharp
kafka.Topic<string, PizzaOrdered>("pizzas", topic =>
{
    topic.Producer(producer =>
    {
        producer.Use<EnrichmentMiddleware>();
    });
});
```

--------------------------------

### Constructing EventMessage with Key and Value

Source: https://emitdotnet.github.io/Emit/kafka/producers

Create an `EventMessage` object containing the key and value. This is a sealed record that can also carry optional headers.

```csharp
// Key + value only
var msg = new EventMessage<string, PizzaOrdered>(pizza.PizzaId.ToString(), pizza);
```

--------------------------------

### Producing Messages with Convenience Overloads

Source: https://emitdotnet.github.io/Emit/kafka/producers

Utilize extension methods from `EventProducerExtensions` to directly produce messages with key, value, and optional headers, skipping the explicit `EventMessage` wrapper.

```csharp
// Key + value
await producer.ProduceAsync(pizza.PizzaId.ToString(), pizza, ct);
```

```csharp
// Key + value + headers
await producer.ProduceAsync(pizza.PizzaId.ToString(), pizza,
    [new KeyValuePair<string, string>("x-tenant-id", tenantId)], ct);
```

--------------------------------

### Add Emit Instrumentation to OpenTelemetry

Source: https://emitdotnet.github.io/Emit/observability/metrics

Configure OpenTelemetry to include Emit's instrumentation. This registers all default meters and allows for further customization.

```csharp
builder.Services.AddOpenTelemetry()
    .WithMetrics(metrics =>
    {
        metrics.AddEmitInstrumentation();
        metrics.AddOtlpExporter();
    });
```

--------------------------------

### Transactional Outbox with IUnitOfWork

Source: https://emitdotnet.github.io/Emit/outbox

Use `IUnitOfWork` for explicit control over transactions in hosted services, background jobs, or scenarios requiring conditional commits. The `await using` statement ensures automatic transaction rollback if exceptions occur before `CommitAsync`.

```csharp
await using var tx = await unitOfWork.BeginAsync(ct);
// ... business writes and produce calls ...
await tx.CommitAsync(ct);
```

--------------------------------

### Implement IConsumer for Message Handling

Source: https://emitdotnet.github.io/Emit/kafka/consumers

Implement the IConsumer<TValue> interface to define how messages of a specific type are handled. Constructor injection works normally as consumers are resolved from the DI container per message.

```csharp
public class PizzaOrderedConsumer(ILogger<PizzaOrderedConsumer> logger)
    : IConsumer<PizzaOrdered>
{
    public async Task ConsumeAsync(ConsumeContext<PizzaOrdered> context, CancellationToken cancellationToken)
    {
        var pizza = context.Message;
        logger.LogInformation("Preparing pizza {PizzaId} for {CustomerId}", pizza.PizzaId, pizza.CustomerId);
    }
}
```

--------------------------------

### Register DLQ Topic

Source: https://emitdotnet.github.io/Emit/error-handling/dead-letter-queue

Register a dead-letter topic for your Kafka builder. This topic will receive all messages that fail processing.

```csharp
emit.AddKafka(kafka =>
{
    kafka.DeadLetter("orders.dlt");
});
```

--------------------------------

### Configuring Direct Delivery Mode for Kafka Producer

Source: https://emitdotnet.github.io/Emit/kafka/producers

Opt into direct delivery mode for a specific Kafka topic producer. This mode bypasses the outbox and calls the Kafka producer immediately, waiting for broker acknowledgement.

```csharp
kafka.Topic<string, AnalyticsEvent>("analytics", topic =>
{
    topic.Producer(p => p.UseDirect());
});
```

--------------------------------

### Configure Circuit Breaker for Consumer Group

Source: https://emitdotnet.github.io/Emit/error-handling/resilience

Set up a circuit breaker for a consumer group, defining failure thresholds, sampling windows, and pause durations. This prevents excessive requests to a failing downstream system.

```csharp
topic.ConsumerGroup("order-processor", group =>
{
    group.CircuitBreaker(cb =>
    {
        cb.FailureThreshold(5);
        cb.SamplingWindow(TimeSpan.FromSeconds(60));
        cb.PauseDuration(TimeSpan.FromSeconds(30));
    });


    group.AddConsumer<OrderPlacedConsumer>();
});

```

--------------------------------

### Implement Audit Consume Observer

Source: https://emitdotnet.github.io/Emit/observability/observers

Implement IConsumeObserver to receive notifications for inbound messages. Default implementations are provided for all methods.

```csharp
public class AuditConsumeObserver(IAuditLog audit) : IConsumeObserver
{
    public Task OnConsumingAsync<T>(ConsumeContext<T> context)
    {
        audit.Record($"Consuming {typeof(T).Name} at {context.Timestamp}");
        return Task.CompletedTask;
    }


    public Task OnConsumedAsync<T>(ConsumeContext<T> context)
    {
        audit.Record($"Consumed {typeof(T).Name}: success");
        return Task.CompletedTask;
    }


    public Task OnConsumeErrorAsync<T>(ConsumeContext<T> context, Exception exception)
    {
        audit.Record($"Consume error: {exception.Message}");
        return Task.CompletedTask;
    }
}
```

--------------------------------

### Register Per-Consumer Middleware

Source: https://emitdotnet.github.io/Emit/kafka/consumers

Registers middleware that specifically wraps a single consumer within a group. This middleware runs after any group-level middleware.

```csharp
group.AddConsumer<PizzaOrderedConsumer>(consumer =>
{
    consumer.Use<RetryLoggingMiddleware>();
});
```

--------------------------------

### Register Delegate for Inline Activity Enrichment

Source: https://emitdotnet.github.io/Emit/observability/tracing

Register a delegate directly with Emit.ConfigureTracing for inline enrichment without a dedicated class. Enrichers are invoked after Emit sets its own standard tags.

```csharp
emit.ConfigureTracing(tracing =>
{
    tracing.EnrichActivity((activity, context) =>
    {
        activity.SetTag("env", "production");
    });
});
```

--------------------------------

### Register Global Middleware

Source: https://emitdotnet.github.io/Emit/advanced/custom-middleware

Register middleware that applies to all messages and all providers globally.

```csharp
builder.Services.AddEmit(emit =>
{
    emit.InboundPipeline.Use<TimingMiddleware<>>();    // open generic
    emit.OutboundPipeline.Use<SigningMiddleware<>>();
});
```

--------------------------------

### Implement Mediator Handlers

Source: https://emitdotnet.github.io/Emit/mediator

Implement handlers by creating classes that implement `IRequestHandler<TRequest>` or `IRequestHandler<TRequest, TResponse>`. Handlers are resolved from the DI container.

```csharp
public class PlacePizzaOrderHandler(IPizzaRepository pizzas)
    : IRequestHandler<PlacePizzaOrder>
{
    public async Task HandleAsync(
        PlacePizzaOrder request,
        CancellationToken cancellationToken)
    {
        await pizzas.SaveAsync(
            new PizzaOrder(Guid.NewGuid(), request.CustomerId, request.Toppings, request.Address),
            cancellationToken);
    }
}
```

```csharp
public class GetPizzaStatusHandler(IPizzaRepository pizzas)
    : IRequestHandler<GetPizzaStatus, PizzaStatusDto>
{
    public async Task<PizzaStatusDto> HandleAsync(
        GetPizzaStatus request,
        CancellationToken cancellationToken)
    {
        var order = await pizzas.FindAsync(request.PizzaId, cancellationToken);
        return new PizzaStatusDto(order.PizzaId, order.Status, order.EstimatedDelivery);
    }
}
```
Tokens:
Raw