redb.Route.S3 2.0.2

redb.Route.S3

S3/MinIO transport for the redb.Route ESB framework.

Polling consumer with prefix filtering, glob patterns, idempotency, done-file markers, move/delete after read. Multi-operation producer with multipart upload, server-side encryption (SSE-S3, SSE-KMS, SSE-C), presigned URLs, tagging, versioning, ACL management, streaming upload, and Glacier restore.

Compatible with AWS S3 and MinIO (via ForcePathStyle).

Quick Start

services.AddRedbRoute(route =>
{
    route.Services.AddRedbRouteS3();

    // Poll MinIO bucket for new CSV files
    route.AddRoute("s3-import", r => r
        .From(S3.Bucket("incoming")
            .ServiceUrl("http://localhost:9000").ForcePathStyle()
            .AccessKey("minioadmin").SecretKey("minioadmin")
            .Prefix("data/").Include("*.csv")
            .DeleteAfterRead().Delay(30_000))
        .To("log:received"));

    // Upload to AWS S3 with SSE-KMS
    route.AddRoute("s3-export", r => r
        .From("direct:upload")
        .To(S3.Bucket("reports")
            .Region("eu-west-1")
            .AccessKey("AKIA...").SecretKey("...")
            .KeyName(Expression.Header("fileName"))
            .UseKmsEncryption("arn:aws:kms:eu-west-1:123456:key/abc")
            .StorageClass("INTELLIGENT_TIERING")));
});

URI Format

s3://bucket-name?option=value&...
s3:OPERATION:bucket-name?option=value&...

Consumer — polls the bucket with ListObjectsV2, downloads matching objects:

s3://my-bucket?accessKey=KEY&secretKey=SECRET&prefix=data/&include=*.json&delay=30000

Producer — default operation PutObject (upload):

s3://my-bucket?accessKey=KEY&secretKey=SECRET&keyName=output/${header.id}.json

Producer with operation — specify operation before bucket name:

s3:CopyObject:my-bucket?accessKey=KEY&secretKey=SECRET
s3:CreateDownloadLink:my-bucket?accessKey=KEY&secretKey=SECRET&presignedUrlExpiration=7200000
s3:ListObjects:my-bucket?accessKey=KEY&secretKey=SECRET&prefix=archive/

Producer Operations

22 operations available, set via URI (s3:OPERATION:bucket) or redbS3.Operation header at runtime.

Object Operations

Presigned URLs

Expiration controlled by presignedUrlExpiration (default: 3600000 ms = 1 hour).

Bucket Operations

Tagging

ACL

Versioning

Glacier

Consumer

Polls S3 bucket using ListObjectsV2 with automatic pagination.

Pipeline

1. List — ListObjectsV2 with Prefix, paginated via ContinuationToken
2. Filter — glob Include/Exclude patterns, MinAge/MaxAge, folder exclusion
3. Sort — SortBy (Key, LastModified, Size + Desc variants)
4. Limit — MaxMessagesPerPoll (default: 10, 0 = unlimited)
5. Per object:
   • Done-file check (DoneFileName pattern)
   • Idempotency check (in-memory, key = Key + ETag + Size)
   • Download body (or stream, or metadata-only)
   • Process exchange
   • Post-process: DeleteAfterRead (default) or MoveAfterRead

Move After Read

.From(S3.Bucket("incoming")
    .AccessKey("KEY").SecretKey("SECRET")
    .MoveAfterRead("archive", prefix: "processed/")
    .RemovePrefixOnMove())

Objects are copied to the destination bucket then deleted from source.

Done File Markers

.From(S3.Bucket("data")
    .AccessKey("KEY").SecretKey("SECRET")
    .DoneFileName("${file:name}.done"))

Consumer only processes report.csv when report.csv.done exists in the same prefix.

Idempotency

.From(S3.Bucket("data")
    .AccessKey("KEY").SecretKey("SECRET")
    .Idempotent())

Uses an in-memory ConcurrentDictionary keyed by Key + ETag + Size. Custom key expression supported via Idempotent(Expression.Header("myKey")).

Connection & Credentials

Credential resolution order:

1. UseDefaultCredentialsProvider → AWS default chain (env vars, instance profile, ECS task role)
2. ProfileName → named profile from ~/.aws/credentials
3. SessionToken set → SessionCredentials (STS temporary credentials)
4. AccessKey + SecretKey → BasicAWSCredentials

MinIO / S3-Compatible

S3.Bucket("my-bucket")
    .ServiceUrl("http://localhost:9000")
    .ForcePathStyle()
    .AccessKey("minioadmin")
    .SecretKey("minioadmin")

ForcePathStyle is required for MinIO, Ceph, Dell ECS, and most S3-compatible stores.

Connection Factory

Register a named S3ConnectionFactory in DI, then reference by name:

S3.Bucket("data").ConnectionFactory("myS3Factory")

Proxy

S3.Bucket("data")
    .AccessKey("KEY").SecretKey("SECRET")
    .Proxy("proxy.corp.local", 8080)

Multipart Upload

For large files, enable multipart upload:

.To(S3.Bucket("large-files")
    .AccessKey("KEY").SecretKey("SECRET")
    .KeyName("uploads/${header.name}")
    .MultiPartUpload()
    .PartSize(10_485_760))  // 10 MB parts (min 5 MB)

Automatically splits files larger than PartSize into parts uploaded sequentially.

Server-Side Encryption

Three SSE modes:

// SSE-S3: AWS-managed AES-256
.UseAes256Encryption()

// SSE-KMS: AWS KMS managed key
.UseKmsEncryption("arn:aws:kms:region:account:key/id")

// SSE-C: Customer-provided key
.UseCustomerEncryption("AES256", "base64-key", "base64-key-md5")

Streaming Upload

Accumulate multiple exchanges into a single S3 object:

.To(S3.Bucket("logs")
    .AccessKey("KEY").SecretKey("SECRET")
    .KeyName("batch/events.jsonl")
    .StreamingUpload(batchMessages: 100, batchSize: 5_242_880)
    .StreamingUploadTimeout(30_000)
    .NamingStrategy(S3NamingStrategy.Progressive))

Conditional Write

.To(S3.Bucket("data")
    .AccessKey("KEY").SecretKey("SECRET")
    .KeyName("unique-file.json")
    .ConditionalWrite())

Uses x-amz-if-none-match — upload fails if the object key already exists.

Configuration Reference

Connection / Credentials

Common

Consumer

Producer

Server-Side Encryption

Presigned URLs

Streaming Upload

Metadata

Headers Reference

Common (set by consumer, used by producer)

Consumer Output

Producer Input

Producer Output

Metadata Mapping

Headers prefixed with redbS3.Meta. are mapped to/from S3 user metadata:

redbS3.Meta.x-amz-meta-author → S3 metadata "x-amz-meta-author"

DSL

using redb.Route.S3;

// Basic consumer
S3.Bucket("my-bucket")
    .ServiceUrl("http://localhost:9000").ForcePathStyle()
    .AccessKey("minioadmin").SecretKey("minioadmin")

// Producer with operation
S3.Bucket("my-bucket", S3OperationType.CopyObject)
    .Region("eu-west-1")
    .AccessKey("KEY").SecretKey("SECRET")

// Presigned download link
S3.Bucket("my-bucket", S3OperationType.CreateDownloadLink)
    .AccessKey("KEY").SecretKey("SECRET")
    .PresignedUrlExpiration(7_200_000)

// Full producer example
S3.Bucket("archive")
    .Region("eu-west-1")
    .AccessKey("KEY").SecretKey("SECRET")
    .KeyName(Expression.Header("fileName"))
    .MultiPartUpload().PartSize(10_485_760)
    .UseKmsEncryption("my-key-id")
    .StorageClass("INTELLIGENT_TIERING")
    .Metadata("env=prod,team=data")

The builder implements implicit operator string — pass directly to .From() / .To().

DI Registration

services.AddRedbRoute(route =>
{
    route.Services.AddRedbRouteS3();
    // ...
});

Registers the S3Component (scheme s3).