MinimalLambda.Envelopes.Kafka 2.6.0

MinimalLambda.Envelopes.Kafka

Strongly-typed Kafka event handling for the MinimalLambda framework.

Overview

This package provides KafkaEnvelope<T>, which extends the base KafkaEvent class with a generic Records collection that deserializes base64-encoded Kafka message values into strongly-typed objects. Instead of manually decoding and parsing base64 data from record.Value, you access deserialized payloads directly via record.ValueContent.

Quick Start

Define your message type and handler:

using Amazon.Lambda.KafkaEvents;
using MinimalLambda.Builder;
using MinimalLambda.Envelopes.Kafka;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

var builder = LambdaApplication.CreateBuilder();
var lambda = builder.Build();

// KafkaEnvelope<OrderEvent> provides access to the Kafka event and deserialized OrderEvent payloads
lambda.MapHandler(
 ([FromEvent] KafkaEnvelope<OrderEvent> envelope, ILogger<Program> logger) =>
 {
 foreach (var topic in envelope.Records)
 {
 logger.LogInformation("Processing {Count} records from topic: {Topic}", topic.Value.Count, topic.Key);

 foreach (var record in topic.Value)
 {
 logger.LogInformation(
 "Order ID: {OrderId}, Amount: {Amount}",
 record.ValueContent?.OrderId,
 record.ValueContent?.Amount
 );
 }
 }
 }
);

await lambda.RunAsync();

// Your message payload - will be deserialized from base64-encoded Kafka message value
internal record OrderEvent(string OrderId, decimal Amount, DateTime Timestamp);

Custom Envelopes

To implement custom deserialization logic, extend KafkaEnvelopeBase<T> and override the ExtractPayload method:

// Example: Custom XML deserialization
public sealed class KafkaXmlEnvelope<T> : KafkaEnvelopeBase<T>
{
 private static readonly XmlSerializer Serializer = new(typeof(T));

 public override void ExtractPayload(EnvelopeOptions options)
 {
 foreach (var topic in Records)
 {
 foreach (var record in topic.Value)
 {
 using var reader = new StreamReader(
 record.Value,
 Encoding.UTF8,
 leaveOpen: true
 );
 var base64String = reader.ReadToEnd();
 var xmlBytes = Convert.FromBase64String(base64String);
 using var xmlReader = XmlReader.Create(
 new MemoryStream(xmlBytes),
 options.XmlReaderSettings
 );
 record.ValueContent = (T)Serializer.Deserialize(xmlReader)!;
 }
 }
 }
}

This pattern allows you to support multiple serialization formats while maintaining the same envelope interface.

AOT Support

When using .NET Native AOT, register both the envelope and payload types in your JsonSerializerContext:

[JsonSerializable(typeof(KafkaEnvelope<OrderEvent>))]
[JsonSerializable(typeof(OrderEvent))]
internal partial class SerializerContext : JsonSerializerContext;

Register the serializer and configure envelope options to use the context:

builder.Services.AddLambdaSerializerWithContext<SerializerContext>();

builder.Services.ConfigureEnvelopeOptions(options =>
{
 options.JsonOptions.TypeInfoResolver = SerializerContext.Default;
});

Other Packages

Additional packages in the minimal-lambda framework for abstractions, observability, and event source handling.

License

This project is licensed under the MIT License. See for details.