AwsLambda.Host.Envelopes.Kinesis 1.3.1

AwsLambda.Host.Envelopes.Kinesis

Strongly-typed Kinesis event handling for the AwsLambda.Host framework.

Overview

This package provides KinesisEnvelope<T>, which extends the base KinesisEvent class with a generic Records collection that deserializes Kinesis data streams into strongly-typed objects. Instead of manually decoding and parsing base64 data from record.Kinesis.Data, you access deserialized payloads directly via record.Kinesis.DataContent.

Quick Start

Define your record type and handler:

using Amazon.Lambda.KinesisEvents;
using AwsLambda.Host.Builder;
using AwsLambda.Host.Envelopes.Kinesis;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

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

// KinesisEnvelope<StreamRecord> provides access to the Kinesis event and deserialized StreamRecord payloads
lambda.MapHandler(
 ([Event] KinesisEnvelope<StreamRecord> envelope, ILogger<Program> logger) =>
 {
 foreach (var record in envelope.Records)
 {
 logger.LogInformation(
 "Stream Record: {EventName} at {Timestamp}",
 record.Kinesis.DataContent?.EventName,
 record.Kinesis.DataContent?.Timestamp
 );
 }
 }
);

await lambda.RunAsync();

// Your record payload - will be deserialized from base64-encoded Kinesis data stream
internal record StreamRecord(string EventName, DateTime Timestamp);

Custom Envelopes

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

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

 public override void ExtractPayload(EnvelopeOptions options)
 {
 foreach (var record in Records)
 {
 using var reader = new StreamReader(
 record.Kinesis.Data,
 Encoding.UTF8,
 leaveOpen: true
 );
 var base64String = reader.ReadToEnd();
 var xmlBytes = Convert.FromBase64String(base64String);
 using var xmlReader = XmlReader.Create(
 new MemoryStream(xmlBytes),
 options.XmlReaderSettings
 );
 record.Kinesis.DataContent = (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(KinesisEnvelope<StreamRecord>))]
[JsonSerializable(typeof(StreamRecord))]
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 aws-lambda-host framework for abstractions, observability, and event source handling.

License

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