From 07c6a04368b0dc24a070fa9701131ce4bc9e467f Mon Sep 17 00:00:00 2001 From: Rafael Lillo <7280959+lillo42@users.noreply.github.com> Date: Wed, 4 Mar 2026 13:06:04 +0000 Subject: [PATCH] Add gitbooks --- README.md | 46 ++--- docs/README.md | 42 +++++ docs/SUMMARY.md | 28 ++++ docs/advanced/distributed-locking.md | 94 +++++++++++ docs/advanced/luggage-store.md | 86 ++++++++++ docs/advanced/outbox-inbox.md | 121 ++++++++++++++ docs/advanced/sweeper-archiver.md | 123 ++++++++++++++ docs/getting-started.md | 102 ++++++++++++ docs/providers/aws.md | 240 +++++++++++++++++++++++++++ docs/providers/gcp.md | 220 ++++++++++++++++++++++++ docs/providers/kafka.md | 180 ++++++++++++++++++++ docs/providers/mongodb.md | 143 ++++++++++++++++ docs/providers/mysql.md | 85 ++++++++++ docs/providers/postgres.md | 146 ++++++++++++++++ docs/providers/rabbitmq.md | 205 +++++++++++++++++++++++ docs/providers/redis.md | 155 +++++++++++++++++ docs/providers/rocketmq.md | 144 ++++++++++++++++ docs/providers/sqlite.md | 82 +++++++++ docs/providers/sqlserver.md | 122 ++++++++++++++ 19 files changed, 2342 insertions(+), 22 deletions(-) create mode 100644 docs/README.md create mode 100644 docs/SUMMARY.md create mode 100644 docs/advanced/distributed-locking.md create mode 100644 docs/advanced/luggage-store.md create mode 100644 docs/advanced/outbox-inbox.md create mode 100644 docs/advanced/sweeper-archiver.md create mode 100644 docs/getting-started.md create mode 100644 docs/providers/aws.md create mode 100644 docs/providers/gcp.md create mode 100644 docs/providers/kafka.md create mode 100644 docs/providers/mongodb.md create mode 100644 docs/providers/mysql.md create mode 100644 docs/providers/postgres.md create mode 100644 docs/providers/rabbitmq.md create mode 100644 docs/providers/redis.md create mode 100644 docs/providers/rocketmq.md create mode 100644 docs/providers/sqlite.md create mode 100644 docs/providers/sqlserver.md diff --git a/README.md b/README.md index e18ebb1..5c17415 100644 --- a/README.md +++ b/README.md @@ -67,14 +67,14 @@ services .CreateTopicIfMissing())) .UseSubscriptions(sb => sb .AddSubscription(s => s - .SetSubscriptionName("paramore.example.greeting") + .SetSubscription("paramore.example.greeting") .SetQueue("greeting.event.queue") .SetTopic("greeting.event.topic") .SetTimeout(TimeSpan.FromSeconds(200)) .EnableDurable() .EnableHighAvailability()) .AddSubscription(s => s - .SetSubscriptionName("paramore.example.farewell") + .SetSubscription("paramore.example.farewell") .SetQueue("farewell.event.queue") .SetTopic("farewell.event.topic"))) )); @@ -111,28 +111,32 @@ services.AddFluentBrighter(brighter => brighter ```csharp services.AddFluentBrighter(brighter => brighter .UsingGcp(gcp => gcp - .SetProjectId("my-gcp-project") + .SetConnection(c => c + .SetProjectId("my-gcp-project")) .UsePubSubPublication(pb => pb .AddPublication(p => p - .SetTopicAttributes(t => t.SetName("user-registered-topic")) + .SetTopic("user-registered-topic") .SetSource("https://example.com/users")) .AddPublication(p => p - .SetTopicAttributes(t => t.SetName("user-deleted-topic")) + .SetTopic("user-deleted-topic") .SetSource("https://example.com/users"))) .UsePubSubSubscription(sb => sb .AddSubscription(s => s .SetSubscriptionName("user-registration-handler") - .SetTopicAttributes(t => t.SetName("user-registered-topic")) + .SetSubscription("user-registered-queue") + .SetTopic("user-registered-topic") .SetNoOfPerformers(5)) .AddSubscription(s => s .SetSubscriptionName("user-deletion-handler") - .SetTopicAttributes(t => t.SetName("user-deleted-topic")) + .SetSubscription("user-deleted-queue") + .SetTopic("user-deleted-topic") .SetNoOfPerformers(3))) + .SetFirestoreConfiguration("my-firestore-database") .UseFirestoreOutbox("outbox") .UseFirestoreInbox("inbox") .UseFirestoreOutboxArchive(cfg => cfg .SetMinimumAge(TimeSpan.FromDays(7)) - .SetBatchSize(100)))); + .SetArchiveBatchSize(100)))); ``` ### Apache Kafka @@ -140,17 +144,17 @@ services.AddFluentBrighter(brighter => brighter ```csharp services.AddFluentBrighter(brighter => brighter .UsingKafka(kafka => kafka - .SetBootstrapServers("localhost:9092") + .SetConnection(c => c + .SetBootstrapServers("localhost:9092")) .UsePublications(pb => pb .AddPublication(p => p - .SetTopic("payments-processed") - .SetMessageIdHeaderKey("message-id"))) + .SetTopic("payments-processed"))) .UseSubscriptions(sb => sb .AddSubscription(s => s - .SetSubscriptionName("payment-notification-service") + .SetSubscription("payment-notification-service") .SetTopic("payments-processed") - .SetGroupId("payment-consumers") - .SetNoOfPerformers(10))))); + .SetConsumerGroupId("payment-consumers") + .SetNumberOfPerformers(10))))); ``` ### PostgreSQL with Outbox Pattern @@ -161,13 +165,13 @@ services.AddFluentBrighter(brighter => brighter .Producers(producer => producer .UsePostgresOutbox(cfg => cfg .SetConnectionString("Host=localhost;Database=brighter;") - .SetTableName("outbox"))) + .SetOutboxTableName("outbox"))) .Subscriptions(sub => sub .UsePostgresInbox(cfg => cfg .SetConnectionString("Host=localhost;Database=brighter;") - .SetTableName("inbox"))) + .SetInboxTableName("inbox"))) .UseOutboxSweeper(cfg => cfg - .SetTimerInterval(TimeSpan.FromSeconds(30)) + .SetTimerInterval(30) .SetBatchSize(100))); ``` @@ -177,12 +181,10 @@ services.AddFluentBrighter(brighter => brighter services.AddFluentBrighter(brighter => brighter .UsingRabbitMq(rabbitmq => /* ... RabbitMQ config ... */) .Producers(producer => producer - .UseSqlServerOutbox(cfg => cfg - .SetConnectionString("Server=localhost;Database=Brighter;") - .SetTableName("Outbox")) - .UseSqlServerDistributedLock(cfg => cfg + .UseMicrosoftSqlServerOutbox(cfg => cfg .SetConnectionString("Server=localhost;Database=Brighter;") - .SetTableName("DistributedLock")))); + .SetOutboxTableName("Outbox")) + .UseMicrosoftSqlServerDistributedLock("Server=localhost;Database=Brighter;"))); ``` ## 🎯 Key Concepts diff --git a/docs/README.md b/docs/README.md new file mode 100644 index 0000000..e6ad75e --- /dev/null +++ b/docs/README.md @@ -0,0 +1,42 @@ +# Fluent Brighter + +A fluent, type-safe configuration API for [Paramore.Brighter](https://github.com/BrighterCommand/Brighter) — the command processor and message-oriented middleware framework for .NET. + +## Why Fluent Brighter? + +Paramore.Brighter is a powerful framework, but configuring its messaging infrastructure can involve significant boilerplate. Fluent Brighter provides an intuitive, strongly-typed builder pattern that makes complex configurations easier to read, write, and maintain. + +### Key Features + +- **Type-Safe Configuration** — IntelliSense-driven API with compile-time safety +- **Multiple Message Brokers** — RabbitMQ, Kafka, AWS SNS/SQS, GCP Pub/Sub, Redis, RocketMQ +- **Database Transports** — PostgreSQL, SQL Server, MySQL, SQLite, MongoDB +- **Transactional Outbox Pattern** — DynamoDB, Firestore, Spanner, MongoDB, SQL Server, PostgreSQL, MySQL, SQLite +- **Idempotent Inbox Pattern** — Prevent duplicate message processing +- **Distributed Locking** — Coordinate across multiple instances +- **Large Message Storage** — S3, GCS, MongoDB GridFS for oversized payloads +- **Fluent Syntax** — Readable, maintainable configuration code + +## Quick Example + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingRabbitMq(rabbitmq => rabbitmq + .SetConnection(conn => conn + .SetAmpq(amqp => amqp.SetUri("amqp://guest:guest@localhost:5672")) + .SetExchange(exchange => exchange.SetName("my.exchange"))) + .UsePublications(pb => pb + .AddPublication(p => p + .SetTopic("greeting.topic") + .CreateTopicIfMissing())) + .UseSubscriptions(sb => sb + .AddSubscription(s => s + .SetSubscription("greeting-handler") + .SetQueue("greeting.queue") + .SetTopic("greeting.topic"))))); +``` + +## Supported Platforms + +- .NET 8.0 +- .NET 9.0 diff --git a/docs/SUMMARY.md b/docs/SUMMARY.md new file mode 100644 index 0000000..6765a64 --- /dev/null +++ b/docs/SUMMARY.md @@ -0,0 +1,28 @@ +# Table of contents + +* [Introduction](README.md) +* [Getting Started](getting-started.md) + +## Messaging Providers + +* [RabbitMQ](providers/rabbitmq.md) +* [Apache Kafka](providers/kafka.md) +* [AWS SNS/SQS](providers/aws.md) +* [Google Cloud Pub/Sub](providers/gcp.md) +* [Redis](providers/redis.md) +* [Apache RocketMQ](providers/rocketmq.md) + +## Database Providers + +* [PostgreSQL](providers/postgres.md) +* [SQL Server](providers/sqlserver.md) +* [MySQL](providers/mysql.md) +* [SQLite](providers/sqlite.md) +* [MongoDB](providers/mongodb.md) + +## Advanced Topics + +* [Outbox & Inbox Patterns](advanced/outbox-inbox.md) +* [Distributed Locking](advanced/distributed-locking.md) +* [Large Message Storage](advanced/luggage-store.md) +* [Outbox Sweeper & Archiver](advanced/sweeper-archiver.md) diff --git a/docs/advanced/distributed-locking.md b/docs/advanced/distributed-locking.md new file mode 100644 index 0000000..d754555 --- /dev/null +++ b/docs/advanced/distributed-locking.md @@ -0,0 +1,94 @@ +# Distributed Locking + +Distributed locking coordinates message processing across multiple service instances to prevent concurrent execution of the outbox sweeper or other shared operations. + +## Supported Providers + +| Provider | Method | Package | +|----------|--------|---------| +| PostgreSQL | `UsePostgresDistributedLock(...)` | `Fluent.Brighter.Postgres` | +| SQL Server | `UseMicrosoftSqlServerDistributedLock(...)` | `Fluent.Brighter.SqlServer` | +| MySQL | `UseMySqlDistributedLock(...)` | `Fluent.Brighter.MySql` | +| MongoDB | `UseMongoDbDistributedLock(...)` | `Fluent.Brighter.MongoDb` | +| DynamoDB | `UseDynamoDbDistributedLock(...)` | `Fluent.Brighter.AWS.V4` | +| Firestore | `UseFirestoreLocking(...)` | `Fluent.Brighter.GoogleCloud` | + +## Configuration + +### Via Provider Configurator (Integrated) + +When using a database provider, enable distributed locking directly: + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingPostgres(cfg => + { + cfg.SetConnection(new RelationalDatabaseConfiguration("Host=localhost;Database=brighter;")); + cfg.UseDistributedLock(); + })); +``` + +### Via Producer Builder (Standalone) + +When using a separate messaging transport: + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingRabbitMq(rabbitmq => /* ... */) + .Producers(producer => producer + .UsePostgresOutbox(/* ... */) + .UsePostgresDistributedLock("Host=localhost;Database=brighter;"))); +``` + +### Provider-Specific Examples + +#### PostgreSQL + +```csharp +// With connection string +.UsePostgresDistributedLock("Host=localhost;Database=brighter;") + +// With configuration object +.UsePostgresDistributedLock(new RelationalDatabaseConfiguration("Host=localhost;Database=brighter;")) +``` + +#### SQL Server + +```csharp +.UseMicrosoftSqlServerDistributedLock("Server=localhost;Database=Brighter;") +``` + +#### MySQL + +```csharp +.UseMySqlDistributedLock("Server=localhost;Database=brighter;User=root;Password=password;") +``` + +#### MongoDB + +```csharp +// Via configurator +.UsingMongoDb(mongodb => mongodb + .SetConnection(c => c.SetConnectionString("mongodb://localhost:27017").SetDatabaseName("brighter")) + .UseDistributedLock("locking")) + +// Via producer builder +.Producers(producer => producer + .UseMongoDbDistributedLock(cfg => { /* MongoDbLockingBuilder config */ })) +``` + +#### DynamoDB + +```csharp +.UsingAws(aws => aws + // ... connection config ... + .UseDynamoDbDistributedLock(cfg => { /* DynamoDbLockingProviderOptionsBuilder config */ })) +``` + +#### Firestore + +```csharp +.UsingGcp(gcp => gcp + // ... connection config ... + .UseFirestoreLocking(cfg => { /* FirestoreLockingBuilder config */ })) +``` diff --git a/docs/advanced/luggage-store.md b/docs/advanced/luggage-store.md new file mode 100644 index 0000000..5ebba44 --- /dev/null +++ b/docs/advanced/luggage-store.md @@ -0,0 +1,86 @@ +# Large Message Storage (Luggage Store) + +When message payloads exceed the size limits of your message broker, Fluent Brighter can automatically store the payload in a cloud storage service and pass a reference through the broker instead. + +## Supported Providers + +| Provider | Method | Package | +|----------|--------|---------| +| AWS S3 | `UseS3LuggageStore(...)` | `Fluent.Brighter.AWS.V4` | +| Google Cloud Storage | `UseGcsLuggageStore(...)` | `Fluent.Brighter.GoogleCloud` | +| MongoDB GridFS | `UseLuggageStore(...)` | `Fluent.Brighter.MongoDb` | + +## Configuration + +### AWS S3 + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingAws(aws => aws + .SetConnection(/* ... */) + // ... publications and subscriptions ... + .UseS3LuggageStore("my-luggage-bucket"))); +``` + +Or with advanced configuration: + +```csharp +.UseS3LuggageStore(cfg => { /* S3LuggageStoreBuilder config */ }) +``` + +### Google Cloud Storage + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingGcp(gcp => gcp + .SetConnection(/* ... */) + // ... publications and subscriptions ... + .UseGcsLuggageStore("my-luggage-bucket"))); +``` + +Or with advanced configuration: + +```csharp +.UseGcsLuggageStore(cfg => { /* GcsLuggageStoreBuilder config */ }) +``` + +### MongoDB GridFS + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingMongoDb(mongodb => mongodb + .SetConnection(c => c + .SetConnectionString("mongodb://localhost:27017") + .SetDatabaseName("brighter")) + .UseLuggageStore("my-bucket"))); +``` + +Or with advanced configuration: + +```csharp +.UseLuggageStore(cfg => { /* MongoDbLuggageStoreBuilder config */ }) +``` + +### Generic Luggage Store + +You can also configure a luggage store using a custom `IAmAStorageProvider` implementation: + +```csharp +services.AddFluentBrighter(brighter => brighter + .SetLuggageStore(store => store + .UseLuggageStore())); +``` + +Or provide a factory: + +```csharp +.SetLuggageStore(store => store + .UseLuggageStore(sp => new MyCustomStorageProvider(sp.GetRequiredService()))) +``` + +## How It Works + +1. When publishing a message, if the payload exceeds the broker's size limit, the payload is uploaded to the configured storage provider +2. A reference (claim check) is placed in the message instead of the full payload +3. When consuming the message, the reference is used to retrieve the full payload from storage +4. The consumer receives the fully hydrated message transparently diff --git a/docs/advanced/outbox-inbox.md b/docs/advanced/outbox-inbox.md new file mode 100644 index 0000000..1c0f6d3 --- /dev/null +++ b/docs/advanced/outbox-inbox.md @@ -0,0 +1,121 @@ +# Outbox & Inbox Patterns + +## Outbox Pattern + +The Transactional Outbox pattern ensures reliable message publishing by storing messages in a database transaction alongside your business data, then publishing them asynchronously via the [Outbox Sweeper](sweeper-archiver.md). + +This guarantees that messages are published if and only if the business transaction succeeds — even if the message broker is temporarily unavailable. + +### How It Works + +1. Your application writes business data and an outbox message in the **same database transaction** +2. The outbox sweeper periodically reads unpublished messages from the outbox table +3. Messages are published to the message broker +4. Published messages are marked as dispatched + +### Supported Outbox Stores + +| Provider | Method | Package | +|----------|--------|---------| +| PostgreSQL | `UsePostgresOutbox(...)` | `Fluent.Brighter.Postgres` | +| SQL Server | `UseMicrosoftSqlServerOutbox(...)` | `Fluent.Brighter.SqlServer` | +| MySQL | `UseMySqlOutbox(...)` | `Fluent.Brighter.MySql` | +| SQLite | `UseSqliteOutbox(...)` | `Fluent.Brighter.Sqlite` | +| MongoDB | `UseMongoDbOutbox(...)` | `Fluent.Brighter.MongoDb` | +| DynamoDB | `UseDynamoDbOutbox(...)` | `Fluent.Brighter.AWS.V4` | +| Firestore | `UseFirestoreOutbox(...)` | `Fluent.Brighter.GoogleCloud` | +| Spanner | `UseSpannerOutbox(...)` | `Fluent.Brighter.GoogleCloud` | + +### Configuration Approaches + +There are two ways to configure the outbox: + +#### 1. Via the Provider Configurator (Integrated) + +When using a database as both transport and outbox: + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingPostgres(cfg => + { + cfg.SetConnection(new RelationalDatabaseConfiguration("Host=localhost;Database=brighter;")); + cfg.UseOutbox(); + })); +``` + +#### 2. Via the Producer Builder (Standalone) + +When using a separate messaging transport with a database outbox: + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingRabbitMq(rabbitmq => /* ... */) + .Producers(producer => producer + .UsePostgresOutbox(cfg => cfg + .SetConnectionString("Host=localhost;Database=brighter;") + .SetOutboxTableName("outbox")))); +``` + +--- + +## Inbox Pattern + +The Inbox pattern prevents duplicate message processing by tracking received messages in a database. When a message arrives, the inbox is checked to determine if it has already been processed. + +### Supported Inbox Stores + +| Provider | Method | Package | +|----------|--------|---------| +| PostgreSQL | `UsePostgresInbox(...)` | `Fluent.Brighter.Postgres` | +| SQL Server | `UseMicrosoftSqlServerInbox(...)` | `Fluent.Brighter.SqlServer` | +| MySQL | `UseMySqlInbox(...)` | `Fluent.Brighter.MySql` | +| SQLite | `UseSqliteInbox(...)` | `Fluent.Brighter.Sqlite` | +| MongoDB | `UseMongoDbInbox(...)` | `Fluent.Brighter.MongoDb` | +| DynamoDB | `UseDynamoDbInbox(...)` | `Fluent.Brighter.AWS.V4` | +| Firestore | `UseFirestoreInbox(...)` | `Fluent.Brighter.GoogleCloud` | +| Spanner | `UseSpannerInbox(...)` | `Fluent.Brighter.GoogleCloud` | + +### Configuration + +#### Via the Provider Configurator + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingPostgres(cfg => + { + cfg.SetConnection(new RelationalDatabaseConfiguration("Host=localhost;Database=brighter;")); + cfg.UseInbox(); + })); +``` + +#### Via the Consumer Builder + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingRabbitMq(rabbitmq => /* ... */) + .Subscriptions(sub => sub + .UsePostgresInbox(cfg => cfg + .SetConnectionString("Host=localhost;Database=brighter;") + .SetInboxTableName("inbox")))); +``` + +#### Inbox Configuration Options + +You can also configure inbox behavior directly on the consumer builder: + +```csharp +.Subscriptions(sub => sub + .SetInbox(inbox => inbox + .SetInbox(myInboxInstance) + .SetScope(InboxScope.Commands) + .SetOnceOnly(true) + .SetActionOnExists(OnceOnlyAction.Throw))) +``` + +| Method | Description | +|--------|-------------| +| `SetInbox(IAmAnInbox)` | Set the inbox implementation | +| `SetScope(InboxScope)` | Scope: `Commands`, `All` | +| `SetOnceOnly(bool)` | Enable once-only processing | +| `SetActionOnExists(OnceOnlyAction)` | Action when duplicate: `Throw`, `Warn` | +| `SetContext(Func)` | Context key function | diff --git a/docs/advanced/sweeper-archiver.md b/docs/advanced/sweeper-archiver.md new file mode 100644 index 0000000..7b311dc --- /dev/null +++ b/docs/advanced/sweeper-archiver.md @@ -0,0 +1,123 @@ +# Outbox Sweeper & Archiver + +## Outbox Sweeper + +The outbox sweeper is a background service that periodically reads unpublished messages from the outbox and publishes them to the message broker. This is a critical component of the [Transactional Outbox pattern](outbox-inbox.md). + +### Basic Setup + +```csharp +services.AddFluentBrighter(brighter => brighter + .UseOutboxSweeper() + // ... provider config ... +); +``` + +### With Configuration + +```csharp +services.AddFluentBrighter(brighter => brighter + .UseOutboxSweeper(cfg => cfg + .SetTimerInterval(30) // Run every 30 seconds + .SetBatchSize(100) // Process 100 messages per sweep + .SetMinimumMessageAge(TimeSpan.FromSeconds(5)) // Only sweep messages older than 5s + .SetBulk(true)) // Use bulk dispatch + // ... provider config ... +); +``` + +### Sweeper Options + +| Method | Description | +|--------|-------------| +| `SetTimerInterval(int seconds)` | Interval between sweeps in seconds | +| `SetBatchSize(int)` | Number of messages to process per sweep | +| `SetMinimumMessageAge(TimeSpan)` | Minimum age before a message is eligible for sweeping | +| `SetBulk(bool)` | Enable bulk dispatch mode | +| `SetArg(string key, object value)` | Set custom arguments | + +--- + +## Outbox Archiver + +The outbox archiver moves old dispatched messages from the outbox to an archive store. This keeps the outbox table small and performant. + +### Basic Setup + +```csharp +services.AddFluentBrighter(brighter => brighter + .UseDbTransactionOutboxArchive() + // ... provider config ... +); +``` + +### With Configuration + +```csharp +services.AddFluentBrighter(brighter => brighter + .UseDbTransactionOutboxArchive(cfg => cfg + .SetTimerInterval(60) // Run every 60 seconds + .SetMinimumAge(TimeSpan.FromDays(7)) // Archive messages older than 7 days + .SetArchiveBatchSize(500)) // Archive 500 messages per run + // ... provider config ... +); +``` + +### Archiver Options + +| Method | Description | +|--------|-------------| +| `SetTimerInterval(int seconds)` | Interval between archive runs in seconds | +| `SetMinimumAge(TimeSpan)` | Minimum age before a message is archived | +| `SetMinimumAgeInHours(int)` | Minimum age in hours (convenience) | +| `SetArchiveBatchSize(int)` | Number of messages to archive per run | +| `SetInstrumentation(InstrumentationOptions)` | Instrumentation level | + +### Provider-Specific Archivers + +Some providers have their own archive extensions: + +```csharp +// DynamoDB +.UseDynamoDbTransactionOutboxArchive() + +// Firestore +.UseFirestoreTransactionOutboxArchive() + +// Spanner +.UseSpannerTransactionOutboxArchive() +``` + +### Custom Archive Provider + +You can provide a custom archive provider: + +```csharp +services.AddFluentBrighter(brighter => brighter + .UseOutboxArchiver(myArchiveProvider, opts => + { + opts.TimerInterval = 60; + opts.MinimumAge = TimeSpan.FromDays(7); + })); +``` + +## Combining Sweeper and Archiver + +A typical production setup uses both: + +```csharp +services.AddFluentBrighter(brighter => brighter + .UseOutboxSweeper(cfg => cfg + .SetTimerInterval(30) + .SetBatchSize(100)) + .UseDbTransactionOutboxArchive(cfg => cfg + .SetTimerInterval(300) + .SetMinimumAge(TimeSpan.FromDays(7)) + .SetArchiveBatchSize(500)) + .UsingRabbitMq(/* ... */) + .UsingPostgres(cfg => + { + cfg.SetConnection(new RelationalDatabaseConfiguration("Host=localhost;Database=brighter;")); + cfg.UseOutbox(); + })); +``` diff --git a/docs/getting-started.md b/docs/getting-started.md new file mode 100644 index 0000000..55348f8 --- /dev/null +++ b/docs/getting-started.md @@ -0,0 +1,102 @@ +# Getting Started + +## Installation + +Install the core package and your chosen provider(s) via the .NET CLI: + +```bash +# Core package (required) +dotnet add package Fluent.Brighter +``` + +Then add the packages for your messaging provider and any database integrations you need: + +### Messaging Providers + +```bash +dotnet add package Fluent.Brighter.RMQ.Async # RabbitMQ (async) +dotnet add package Fluent.Brighter.RMQ.Sync # RabbitMQ (sync) +dotnet add package Fluent.Brighter.Kafka # Apache Kafka +dotnet add package Fluent.Brighter.AWS.V4 # AWS SNS/SQS +dotnet add package Fluent.Brighter.GoogleCloud # GCP Pub/Sub +dotnet add package Fluent.Brighter.Redis # Redis Streams +dotnet add package Fluent.Brighter.RocketMQ # Apache RocketMQ +``` + +### Database Providers + +These packages can be used as message transports and/or for outbox/inbox storage: + +```bash +dotnet add package Fluent.Brighter.Postgres # PostgreSQL +dotnet add package Fluent.Brighter.SqlServer # SQL Server +dotnet add package Fluent.Brighter.MySql # MySQL +dotnet add package Fluent.Brighter.Sqlite # SQLite +dotnet add package Fluent.Brighter.MongoDb # MongoDB +``` + +## Basic Setup + +All configuration starts with the `AddFluentBrighter` extension method on `IServiceCollection`: + +```csharp +using Fluent.Brighter; + +services.AddFluentBrighter(brighter => +{ + // Configure your messaging provider + brighter.UsingRabbitMq(rabbitmq => { /* ... */ }); + + // Optional: configure producers with outbox + brighter.Producers(producer => { /* ... */ }); + + // Optional: configure consumers with inbox + brighter.Subscriptions(sub => { /* ... */ }); + + // Optional: enable outbox sweeper + brighter.UseOutboxSweeper(); +}); +``` + +## Core Builder Options + +The `FluentBrighterBuilder` provides the following configuration methods: + +| Method | Description | +|--------|-------------| +| `Subscriptions(...)` | Configure message consumers, inbox, and subscription settings | +| `Producers(...)` | Configure message producers, outbox, and distributed locking | +| `RequestHandlers(...)` | Register request handlers from assemblies or individually | +| `Mappers(...)` | Register message mappers | +| `Transformers(...)` | Register message transformers | +| `UseOutboxSweeper(...)` | Enable the background outbox sweeper service | +| `UseOutboxArchive(...)` / `UseDbTransactionOutboxArchive(...)` | Enable outbox archiving | +| `SetLuggageStore(...)` | Configure large message (luggage) storage | +| `SetScheduler(...)` | Configure message scheduling | +| `RegisterServices(...)` | Register additional services in the DI container | + +## Message Pump Types + +Brighter supports two message pump types: + +- **Reactor** — Synchronous, single-threaded message processing +- **Proactor** — Asynchronous, concurrent message processing + +## Channel Creation Modes + +When configuring subscriptions and publications, you can control how missing infrastructure is handled: + +- **Create** — Automatically create topics/queues if they don't exist +- **Validate** — Check that infrastructure exists and throw if missing +- **Assume** — Assume infrastructure already exists (no checks) + +## Next Steps + +Choose your messaging provider to get started: + +- [RabbitMQ](providers/rabbitmq.md) +- [Apache Kafka](providers/kafka.md) +- [AWS SNS/SQS](providers/aws.md) +- [Google Cloud Pub/Sub](providers/gcp.md) +- [Redis](providers/redis.md) +- [Apache RocketMQ](providers/rocketmq.md) diff --git a/docs/providers/aws.md b/docs/providers/aws.md new file mode 100644 index 0000000..6bb79d1 --- /dev/null +++ b/docs/providers/aws.md @@ -0,0 +1,240 @@ +# AWS SNS/SQS + +Fluent Brighter provides full support for AWS SNS (publishing) and SQS (subscribing), along with DynamoDB for outbox/inbox, S3 for luggage storage, and EventBridge for scheduling. + +## Installation + +```bash +dotnet add package Fluent.Brighter.AWS.V4 +``` + +> There is also a legacy `Fluent.Brighter.AWS` package. New projects should use `AWS.V4`. + +## Basic Setup + +```csharp +using Fluent.Brighter; + +services.AddFluentBrighter(brighter => brighter + .UsingAws(aws => aws + .SetConnection(conn => conn + .SetCredentials(new BasicAWSCredentials("key", "secret")) + .SetRegion(RegionEndpoint.USEast1)) + .UseSnsPublication(pb => pb + .AddPublication(p => p + .SetTopic("greeting-event-topic") + .CreateTopicIfMissing())) + .UseSqsSubscription(sb => sb + .AddSubscription(s => s + .SetSubscriptionName("paramore.example.greeting") + .SetQueue("greeting-event-queue") + .SetTopic("greeting-event-topic") + .SetMessagePumpType(MessagePumpType.Proactor) + .SetMakeChannels(OnMissingChannel.Create))))); +``` + +## Connection Configuration + +```csharp +.SetConnection(conn => conn + .SetCredentials(new BasicAWSCredentials("key", "secret")) + .SetRegion(RegionEndpoint.USEast1) + .SetClientConfigAction(cfg => cfg.ServiceURL = "http://localhost:4566")) +``` + +### Connection Options + +| Method | Description | +|--------|-------------| +| `SetCredentials(AWSCredentials)` | AWS credentials | +| `SetRegion(RegionEndpoint)` | AWS region | +| `SetClientConfigAction(Action)` | Custom client configuration (e.g., LocalStack endpoint) | + +## SNS Publications + +Publish messages to SNS topics: + +```csharp +.UseSnsPublication(pb => pb + .AddPublication(p => p + .SetTopic("orders-created") + .CreateTopicIfMissing() + .FindTopicByName() + .SetTopicAttributes(attrs => attrs + .UseStandardSns()))) +``` + +### SNS Publication Options + +| Method | Description | +|--------|-------------| +| `SetTopic(RoutingKey)` | SNS topic name | +| `CreateTopicIfMissing()` | Auto-create topic | +| `ValidateIfNotExists()` | Validate topic exists | +| `AssumeIfNotExists()` | Assume topic exists | +| `FindTopicByName()` | Find topic by name | +| `FindTopicByArn()` | Find topic by ARN | +| `FindTopicByConvention()` | Find topic by convention | +| `SetTopicArn(string)` | Set explicit topic ARN | +| `SetTopicAttributes(Action)` | Configure SNS attributes | +| `SetSource(Uri)` / `SetSource(string)` | CloudEvents source | +| `SetDataSchema(Uri)` / `SetDataSchema(string)` | CloudEvents data schema | + +### SNS Attributes + +```csharp +.SetTopicAttributes(attrs => attrs + .UseStandardSns() // or .UseFifoSns() + .SetDeliveryPolicy("...") + .SetPolicy("...") + .EnableContentBasedDeduplication()) +``` + +## SQS Publications + +Publish messages directly to SQS queues (point-to-point): + +```csharp +.UseSqsPublication(pb => pb + .AddPublication(p => p + .SetQueue("farewell-event-queue") + .CreateIfMissing() + .UsePointToPoint())) +``` + +### SQS Publication Options + +| Method | Description | +|--------|-------------| +| `SetQueue(RoutingKey)` | SQS queue name | +| `SetQueueUrl(ChannelName)` | Explicit queue URL | +| `UsePointToPoint()` | Point-to-point channel type | +| `UsePubSub()` | Pub/sub channel type | +| `CreateIfMissing()` | Auto-create queue | +| `FindQueueByName()` | Find queue by name | +| `FindQueueByUrl()` | Find queue by URL | +| `SetQueueAttributes(Action)` | Configure SQS attributes | + +### SQS Attributes + +```csharp +.SetQueueAttributes(attrs => attrs + .UseStandardQueue() // or .UseFifoQueue() + .SetLockTimeout(TimeSpan.FromSeconds(30)) + .SetMessageRetentionPeriod(TimeSpan.FromDays(14)) + .EnableRawMessageDelivery() + .SetRedrivePolicy(redrive => redrive + .SetDeadLetterQueue("my-dlq") + .SetMaxReceiveCount(5))) +``` + +## SQS Subscriptions + +```csharp +.UseSqsSubscription(sb => sb + .AddSubscription(s => s + .SetSubscriptionName("order-processor") + .SetQueue("order-processing-queue") + .SetTopic("orders-created") + .SetMessagePumpType(MessagePumpType.Proactor) + .SetMakeChannels(OnMissingChannel.Create) + .SetNoOfPerformers(5))) +``` + +### SQS Subscription Options + +| Method | Description | +|--------|-------------| +| `SetSubscriptionName(SubscriptionName)` | Unique subscription name | +| `SetQueue(ChannelName)` | SQS queue name | +| `SetTopic(RoutingKey)` | SNS topic to subscribe to | +| `SetChannelType(ChannelType)` | Channel type | +| `SetNoOfPerformers(int)` | Number of concurrent handlers | +| `SetMessagePumpType(MessagePumpType)` | `Reactor` or `Proactor` | +| `SetMakeChannels(OnMissingChannel)` | Channel creation mode | +| `SetTimeOut(TimeSpan)` | Processing timeout | +| `SetBufferSize(int)` | Internal buffer size | +| `SetRequeueCount(int)` | Max requeue attempts | +| `SetRequeueDelay(TimeSpan)` | Requeue delay | +| `SetFindTopicBy(TopicFindBy)` | Topic discovery strategy | +| `SetFindQueueBy(QueueFindBy)` | Queue discovery strategy | +| `SetQueueAttributes(SqsAttributes)` | SQS queue attributes | +| `SetTopicAttributes(SnsAttributes)` | SNS topic attributes | +| `SetEmptyChannelDelay(TimeSpan)` | Delay when channel is empty | +| `SetChannelFailureDelay(TimeSpan)` | Delay after channel failure | + +## DynamoDB Outbox & Inbox + +```csharp +.UsingAws(aws => aws + .SetConnection(/* ... */) + .UseSnsPublication(/* ... */) + .UseSqsSubscription(/* ... */) + .UseDynamoDbOutbox("outbox-table") + .UseDynamoDbInbox("inbox-table") + .UseDynamoDbOutboxArchive()) +``` + +### DynamoDB Options + +| Method | Description | +|--------|-------------| +| `UseDynamoDbOutbox()` | Use DynamoDB outbox with default table | +| `UseDynamoDbOutbox(string tableName)` | Use DynamoDB outbox with specific table | +| `UseDynamoDbOutbox(Action)` | Advanced outbox configuration | +| `UseDynamoDbInbox()` | Use DynamoDB inbox with default table | +| `UseDynamoDbInbox(string tableName)` | Use DynamoDB inbox with specific table | +| `UseDynamoDbOutboxArchive()` | Enable outbox archiving | +| `UseDynamoDbOutboxArchive(Action)` | Configure archive options | +| `UseDynamoDbDistributedLock(Action)` | Enable distributed locking | + +## S3 Luggage Store + +Store large message payloads in S3: + +```csharp +.UsingAws(aws => aws + // ... other config ... + .UseS3LuggageStore("my-luggage-bucket")) +``` + +## EventBridge Scheduler + +```csharp +.UsingAws(aws => aws + // ... other config ... + .UseScheduler()) +``` + +## Full Example + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingAws(aws => aws + .SetConnection(conn => conn + .SetCredentials(new BasicAWSCredentials("key", "secret")) + .SetRegion(RegionEndpoint.USEast1)) + .UseSnsPublication(pb => pb + .AddPublication(p => p + .SetTopic("greeting-event-topic") + .CreateTopicIfMissing())) + .UseSqsPublication(pb => pb + .AddPublication(p => p + .SetQueue("farewell-event-queue") + .CreateIfMissing())) + .UseSqsSubscription(sb => sb + .AddSubscription(s => s + .SetSubscriptionName("paramore.example.greeting") + .SetQueue("greeting-event-queue") + .SetTopic("greeting-event-topic") + .SetMessagePumpType(MessagePumpType.Proactor) + .SetMakeChannels(OnMissingChannel.Create)) + .AddSubscription(s => s + .SetSubscriptionName("paramore.example.farewell") + .SetQueue("farewell-event-queue") + .SetMessagePumpType(MessagePumpType.Proactor) + .SetMakeChannels(OnMissingChannel.Create))) + .UseDynamoDbOutbox("outbox-table") + .UseDynamoDbInbox("inbox-table") + .UseDynamoDbOutboxArchive())); +``` diff --git a/docs/providers/gcp.md b/docs/providers/gcp.md new file mode 100644 index 0000000..5a37985 --- /dev/null +++ b/docs/providers/gcp.md @@ -0,0 +1,220 @@ +# Google Cloud Pub/Sub + +Fluent Brighter provides full support for Google Cloud Pub/Sub, along with Firestore and Spanner for outbox/inbox, GCS for luggage storage, and Firestore for distributed locking. + +## Installation + +```bash +dotnet add package Fluent.Brighter.GoogleCloud +``` + +## Basic Setup + +```csharp +using Fluent.Brighter; + +services.AddFluentBrighter(brighter => brighter + .UsingGcp(gcp => gcp + .SetConnection(c => c + .SetCredential(GoogleCredential.GetApplicationDefault()) + .SetProjectId("my-gcp-project")) + .UsePubSubPublication(pb => pb + .AddPublication(p => p + .SetTopic("greeting-event-topic") + .SetSource("https://example.com/greeting"))) + .UsePubSubSubscription(sb => sb + .AddSubscription(s => s + .SetSubscriptionName("greeting-handler") + .SetSubscription("greeting-event-queue") + .SetTopic("greeting-event-topic") + .SetNoOfPerformers(1))))); +``` + +## Connection Configuration + +```csharp +.SetConnection(c => c + .SetCredential(GoogleCredential.GetApplicationDefault()) + .SetProjectId("my-gcp-project") + .SetPublisherConfiguration(pub => { /* PublisherClientBuilder config */ }) + .SetSubscriptionManagerConfiguration(sub => { /* SubscriberServiceApiClientBuilder config */ }) + .SetStreamConfiguration(stream => { /* SubscriberClientBuilder config */ }) + .SetTopicManagerConfiguration(topic => { /* PublisherServiceApiClientBuilder config */ }) + .SetProjectsClientConfiguration(proj => { /* ProjectsClientBuilder config */ })) +``` + +### Connection Options + +| Method | Description | +|--------|-------------| +| `SetCredential(ICredential)` | GCP credentials | +| `SetProjectId(string)` | GCP project ID | +| `SetPublisherConfiguration(Action)` | Publisher client settings | +| `SetSubscriptionManagerConfiguration(Action)` | Subscription manager settings | +| `SetStreamConfiguration(Action)` | Stream subscriber settings | +| `SetTopicManagerConfiguration(Action)` | Topic manager settings | +| `SetProjectsClientConfiguration(Action)` | Projects client settings | + +## Publications + +```csharp +.UsePubSubPublication(pb => pb + .AddPublication(p => p + .SetTopic("user-registered-topic") + .SetSource("https://example.com/users") + .CreateQueueIfMissing()) + .AddPublication(p => p + .SetTopic("user-deleted-topic") + .SetSource("https://example.com/users"))) +``` + +### Publication Options + +| Method | Description | +|--------|-------------| +| `SetTopic(RoutingKey)` | Pub/Sub topic name | +| `SetTopicAttributes(TopicAttributes)` | Set topic attributes directly | +| `SetTopicAttributes(Action)` | Configure topic attributes via builder | +| `CreateQueueIfMissing()` | Auto-create topic if missing | +| `ValidIfQueueExists()` | Validate topic exists | +| `AssumeQueueExists()` | Assume topic exists | +| `SetSource(Uri)` / `SetSource(string)` | CloudEvents source URI | +| `SetDataSchema(Uri)` / `SetDataSchema(string)` | CloudEvents data schema | +| `SetSubject(string)` | CloudEvents subject | +| `SetType(CloudEventsType)` | CloudEvents type | +| `SetPublisherClientConfiguration(Action)` | Per-publication publisher config | +| `SetDefaultHeaders(IDictionary)` | Default message headers | +| `SetReplyTo(string)` | Reply-to address | +| `SetRequestType(Type)` | Request type | + +### Topic Attributes + +For advanced topic configuration: + +```csharp +.SetTopicAttributes(t => t + .SetName("my-topic") + .SetProjectId("my-project") + .SetLabels(new Dictionary { ["env"] = "prod" }) + .SetMessageRetentionDuration(TimeSpan.FromDays(7)) + .SetKmsKeyName("projects/my-project/locations/global/keyRings/my-ring/cryptoKeys/my-key")) +``` + +## Subscriptions + +```csharp +.UsePubSubSubscription(sb => sb + .AddSubscription(s => s + .SetSubscriptionName("user-registration-handler") + .SetSubscription("user-registered-queue") + .SetTopic("user-registered-topic") + .SetNoOfPerformers(5) + .EnableExactlyOnceDelivery() + .UseStreamMode())) +``` + +### Subscription Options + +| Method | Description | +|--------|-------------| +| `SetSubscriptionName(SubscriptionName)` | Logical subscription name | +| `SetSubscription(ChannelName)` | Pub/Sub subscription name (channel) | +| `SetTopic(RoutingKey)` | Topic to subscribe to | +| `SetNoOfPerformers(int)` | Number of concurrent handlers | +| `SetMessagePumpType(MessagePumpType)` | `Reactor` or `Proactor` | +| `UseReactor()` / `UseProactor()` | Convenience methods for pump type | +| `UseStreamMode()` / `UsePullMode()` | Subscription mode | +| `SetAckDeadlineSeconds(uint)` | Acknowledgment deadline | +| `EnableMessageOrdering()` / `DisableMessageOrdering()` | Message ordering | +| `EnableExactlyOnceDelivery()` / `DisableExactlyOnceDelivery()` | Exactly-once delivery | +| `SetTopicAttributes(TopicAttributes)` | Topic attributes | +| `SetTopicAttributes(Action)` | Topic attributes via builder | +| `SetDeadLetter(Action)` | Dead letter policy | +| `SetProjectId(string)` | Override project ID | +| `SetMakeChannels(OnMissingChannel)` | Channel creation mode | +| `SetEmptyChannelDelay(TimeSpan)` | Delay when channel is empty | +| `SetChannelFailureDelay(TimeSpan)` | Delay after channel failure | +| `SetBufferSize(int)` | Internal buffer size | +| `SetTimeOut(TimeSpan)` | Processing timeout | +| `SetRequeueCount(int)` | Max requeue attempts | +| `SetRequeueDelay(TimeSpan)` | Requeue delay | + +## Firestore Outbox & Inbox + +```csharp +.UsingGcp(gcp => gcp + // ... connection and pub/sub config ... + .SetFirestoreConfiguration("my-firestore-database") + .UseFirestoreOutbox("outbox") + .UseFirestoreInbox("inbox") + .UseFirestoreOutboxArchive()) +``` + +### Firestore Options + +| Method | Description | +|--------|-------------| +| `SetFirestoreConfiguration(string)` | Firestore database name | +| `UseFirestoreOutbox()` | Use Firestore outbox with defaults | +| `UseFirestoreOutbox(string)` | Use Firestore outbox with collection name | +| `UseFirestoreOutbox(Action)` | Advanced outbox config | +| `UseFirestoreInbox()` | Use Firestore inbox with defaults | +| `UseFirestoreInbox(string)` | Use Firestore inbox with collection name | +| `UseFirestoreInbox(Action)` | Advanced inbox config | +| `UseFirestoreOutboxArchive()` | Enable outbox archiving | +| `UseFirestoreOutboxArchive(Action)` | Configure archive options | +| `UseFirestoreLocking(Action)` | Distributed locking with Firestore | + +## Spanner Outbox & Inbox + +```csharp +.UsingGcp(gcp => gcp + // ... connection and pub/sub config ... + .UseSpannerOutbox(cfg => cfg + .SetConfiguration(spannerConfig)) + .UseSpannerInbox(cfg => cfg + .SetConfiguration(spannerConfig))) +``` + +## GCS Luggage Store + +Store large message payloads in Google Cloud Storage: + +```csharp +.UsingGcp(gcp => gcp + // ... other config ... + .UseGcsLuggageStore("my-luggage-bucket")) +``` + +## Full Example + +```csharp +services.AddFluentBrighter(brighter => brighter + .UseOutboxSweeper() + .UsingGcp(gcp => gcp + .SetConnection(c => c + .SetCredential(GoogleCredential.GetApplicationDefault()) + .SetProjectId(Environment.GetEnvironmentVariable("GOOGLE_CLOUD_PROJECT")!)) + .UsePubSubPublication(pb => pb + .AddPublication(p => p + .SetTopic("greeting-event-topic") + .SetSource("https://example.com/greeting")) + .AddPublication(p => p + .SetTopic("farewell-event-topic") + .SetSource("https://example.com/farewell"))) + .UsePubSubSubscription(sb => sb + .AddSubscription(s => s + .SetSubscriptionName("paramore.example.greeting") + .SetSubscription("greeting-event-queue") + .SetTopic("greeting-event-topic") + .SetNoOfPerformers(1)) + .AddSubscription(s => s + .SetSubscriptionName("paramore.example.farewell") + .SetSubscription("farewell-event-queue") + .SetTopic("farewell-event-topic") + .SetNoOfPerformers(1))) + .SetFirestoreConfiguration("brighter-firestore-database") + .UseFirestoreOutbox("outbox") + .UseFirestoreInbox("inbox") + .UseFirestoreOutboxArchive())); +``` diff --git a/docs/providers/kafka.md b/docs/providers/kafka.md new file mode 100644 index 0000000..c89a3bc --- /dev/null +++ b/docs/providers/kafka.md @@ -0,0 +1,180 @@ +# Apache Kafka + +Fluent Brighter provides support for Apache Kafka as a messaging transport. + +## Installation + +```bash +dotnet add package Fluent.Brighter.Kafka +``` + +## Basic Setup + +```csharp +using Fluent.Brighter; + +services.AddFluentBrighter(brighter => brighter + .UsingKafka(kafka => kafka + .SetConnection(c => c + .SetBootstrapServers("localhost:9092")) + .UsePublications(pb => pb + .AddPublication(p => p + .SetTopic("greeting"))) + .UseSubscriptions(sb => sb + .AddSubscription(s => s + .SetTopic("greeting") + .SetConsumerGroupId("greeting.consumer") + .SetMessagePumpType(MessagePumpType.Reactor))))); +``` + +## Connection Configuration + +```csharp +.SetConnection(c => c + .SetBootstrapServers("broker1:9092", "broker2:9092") + .SetName("my-kafka-connection") + .SetSecurityProtocol(SecurityProtocol.SaslSsl) + .SetSaslMechanisms(SaslMechanism.Plain) + .SetSaslUsername("admin") + .SetSaslPassword("admin-secret") + .SetSslCaLocation("/path/to/ca.pem") + .SetSslKeystoreLocation("/path/to/keystore.p12") + .SetSslKeystorePassword("keystore-password")) +``` + +### Connection Options + +| Method | Description | +|--------|-------------| +| `SetBootstrapServers(string)` | Comma-separated or single bootstrap server | +| `SetBootstrapServers(params string[])` | Multiple bootstrap servers | +| `SetName(string)` | Connection name | +| `SetSecurityProtocol(SecurityProtocol)` | Security protocol (`Plaintext`, `Ssl`, `SaslPlaintext`, `SaslSsl`) | +| `SetSaslMechanisms(SaslMechanism)` | SASL mechanism (`Plain`, `ScramSha256`, `ScramSha512`, etc.) | +| `SetSaslUsername(string)` | SASL username | +| `SetSaslPassword(string)` | SASL password | +| `SetSaslKerberosPrincipal(string)` | Kerberos principal | +| `SetSslCaLocation(string)` | CA certificate path | +| `SetSslKeystoreLocation(string)` | Keystore path | +| `SetSslKeystorePassword(string)` | Keystore password | +| `SetDebug(string)` | librdkafka debug flags | + +## Publications + +```csharp +.UsePublications(pb => pb + .AddPublication(p => p + .SetTopic("greeting") + .CreateTopicIfMissing() + .SetNumPartitions(3) + .SetReplicationFactor(2) + .SetReplication(Acks.All) + .EnableIdempotence())) +``` + +### Publication Options + +| Method | Description | +|--------|-------------| +| `SetTopic(RoutingKey)` | Kafka topic name | +| `CreateTopicIfMissing()` | Auto-create topic if missing | +| `ValidIfTopicExists()` | Validate topic exists | +| `AssumeTopicExists()` | Assume topic exists | +| `SetNumPartitions(int)` | Number of partitions (for auto-creation) | +| `SetReplicationFactor(short)` | Replication factor (for auto-creation) | +| `SetReplication(Acks)` | Acknowledgment level (`None`, `Leader`, `All`) | +| `EnableIdempotence()` / `DisableIdempotence()` | Toggle idempotent producer | +| `SetBatchNumberMessages(int)` | Batch size | +| `SetLingerMs(int)` | Linger time in ms | +| `SetMessageSendMaxRetries(int)` | Max send retries | +| `SetMessageTimeoutMs(int)` | Message timeout in ms | +| `SetMaxInFlightRequestsPerConnection(int)` | Max in-flight requests | +| `SetPartitioner(Partitioner)` | Partitioning strategy | +| `SetQueueBufferingMaxMessages(int)` | Max buffered messages | +| `SetQueueBufferingMaxKbytes(int)` | Max buffer size in KB | +| `SetRetryBackoff(int)` | Retry backoff in ms | +| `SetRequestTimeoutMs(int)` | Request timeout in ms | +| `SetTopicFindTimeoutMs(int)` | Topic discovery timeout in ms | +| `SetTransactionalId(string)` | Transactional ID | +| `SetMessageHeaderBuilder(IKafkaMessageHeaderBuilder)` | Custom header builder | +| `SetSource(Uri)` | CloudEvents source | +| `SetDataSchema(Uri)` | CloudEvents data schema | + +## Subscriptions + +```csharp +.UseSubscriptions(sb => sb + .AddSubscription(s => s + .SetTopic("greeting") + .SetConsumerGroupId("greeting.consumer") + .SetMessagePumpType(MessagePumpType.Reactor) + .SetNumberOfPerformers(5) + .SetOffsetDefault(AutoOffsetReset.Earliest) + .SetCommitBatchSize(10))) +``` + +### Subscription Options + +| Method | Description | +|--------|-------------| +| `SetSubscription(SubscriptionName)` | Unique subscription identifier | +| `SetTopic(RoutingKey)` | Kafka topic to consume | +| `SetConsumerGroupId(string)` | Consumer group ID | +| `SetMessagePumpType(MessagePumpType)` | `Reactor` or `Proactor` | +| `UseReactorMode()` / `UseProactorMode()` | Convenience methods for pump type | +| `SetNumberOfPerformers(int)` | Number of concurrent consumers | +| `SetOffsetDefault(AutoOffsetReset)` | Default offset (`Earliest`, `Latest`) | +| `SetCommitBatchSize(long)` | Commit after N messages | +| `SetSessionTimeout(TimeSpan)` | Consumer session timeout | +| `SetMaxPollInterval(TimeSpan)` | Max poll interval | +| `SetIsolationLevel(IsolationLevel)` | Read isolation level | +| `SetPartitionAssignmentStrategy(PartitionAssignmentStrategy)` | Partition assignment strategy | +| `SetNumPartitions(int)` | Number of partitions | +| `SetReplicationFactor(short)` | Replication factor | +| `SetConfigHook(Action)` | Direct access to consumer config | +| `SetTimeout(TimeSpan)` | Processing timeout | +| `SetBufferSize(int)` | Internal buffer size | +| `SetRequeueCount(int)` | Max requeue attempts | +| `SetRequeueDelay(TimeSpan)` | Requeue delay | +| `SetMakeChannels(OnMissingChannel)` | Channel creation mode | +| `CreateInfrastructureIfMissing()` | Auto-create infrastructure | +| `ValidIfInfrastructureExists()` | Validate infrastructure exists | +| `AssumeInfrastructureExists()` | Assume infrastructure exists | +| `SetSweepUncommittedOffsetsInterval(TimeSpan)` | Uncommitted offset sweep interval | +| `SetTopicFindTimeout(TimeSpan)` | Topic discovery timeout | +| `SetReadCommittedOffsetsTimeOut(TimeSpan)` | Committed offsets read timeout | + +## Full Example + +```csharp +services.AddFluentBrighter(builder => +{ + builder + .UseDbTransactionOutboxArchive() + .UseOutboxSweeper() + .UsingKafka(cfg => + { + cfg.SetConnection(c => c + .SetBootstrapServers("localhost:9092") + .SetName("sample") + .SetSaslUsername("admin") + .SetSaslPassword("admin-secret") + .SetSecurityProtocol(SecurityProtocol.Plaintext) + .SetSaslMechanisms(SaslMechanism.Plain)); + + cfg + .UsePublications(pp => pp + .AddPublication(p => p.SetTopic("greeting")) + .AddPublication(p => p.SetTopic("farewell"))) + .UseSubscriptions(sb => sb + .AddSubscription(s => s + .SetTopic("greeting") + .SetConsumerGroupId("greeting.consumer") + .SetMessagePumpType(MessagePumpType.Reactor)) + .AddSubscription(s => s + .SetTopic("farewell") + .SetConsumerGroupId("farewell.consumer") + .SetMessagePumpType(MessagePumpType.Reactor))); + }); +}); +``` diff --git a/docs/providers/mongodb.md b/docs/providers/mongodb.md new file mode 100644 index 0000000..049ae71 --- /dev/null +++ b/docs/providers/mongodb.md @@ -0,0 +1,143 @@ +# MongoDB + +Fluent Brighter supports MongoDB as an **outbox/inbox store**, **distributed lock provider**, and **luggage store** (via GridFS) for use with other messaging providers. + +## Installation + +```bash +dotnet add package Fluent.Brighter.MongoDb +``` + +## Setup + +Use MongoDB alongside a messaging provider like RabbitMQ: + +```csharp +services.AddFluentBrighter(brighter => brighter + .UseOutboxSweeper() + .UseOutboxArchiver(new NullOutboxArchiveProvider()) + .UsingRabbitMq(rabbitmq => rabbitmq + .SetConnection(conn => conn + .SetAmpq(amqp => amqp.SetUri("amqp://guest:guest@localhost:5672")) + .SetExchange(exchange => exchange.SetName("paramore.brighter.exchange"))) + .UsePublications(pb => pb + .AddPublication(p => p + .SetTopic("greeting.event.topic") + .CreateTopicIfMissing())) + .UseSubscriptions(sb => sb + .AddSubscription(s => s + .SetSubscription("greeting-handler") + .SetQueue("greeting.event.queue") + .SetTopic("greeting.event.topic")))) + .UsingMongoDb(mongodb => mongodb + .SetConnection(c => c + .SetConnectionString("mongodb://root:example@localhost:27017") + .SetDatabaseName("brighter")) + .UseOutbox("outbox") + .UseInbox("inbox") + .UseDistributedLock("locking") + .UseLuggageStore("bucket"))); +``` + +## Connection Configuration + +```csharp +.SetConnection(c => c + .SetConnectionString("mongodb://root:example@localhost:27017") + .SetDatabaseName("brighter")) +``` + +### Connection Options + +| Method | Description | +|--------|-------------| +| `SetConnectionString(string)` | MongoDB connection string | +| `SetDatabaseName(string)` | Database name | + +## Configurator Options + +| Method | Description | +|--------|-------------| +| `SetConnection(Action)` | Configure connection via builder | +| `SetConnection(IAmAMongoDbConfiguration)` | Set connection directly | +| `UseOutbox()` | Enable outbox with default collection | +| `UseOutbox(string collectionName)` | Enable outbox with specific collection | +| `UseOutbox(Action)` | Advanced outbox config | +| `UseInbox()` | Enable inbox with default collection | +| `UseInbox(string collectionName)` | Enable inbox with specific collection | +| `UseInbox(Action)` | Advanced inbox config | +| `UseDistributedLock()` | Enable distributed lock with default collection | +| `UseDistributedLock(string collectionName)` | Enable distributed lock with specific collection | +| `UseDistributedLock(Action)` | Advanced lock config | +| `UseLuggageStore(string bucketName)` | Enable GridFS luggage store | +| `UseLuggageStore(Action)` | Advanced luggage store config | + +## Extension Methods on ProducerBuilder + +| Method | Description | +|--------|-------------| +| `UseMongoDbOutbox(Action)` | Configure outbox | +| `UseMongoDbDistributedLock(Action)` | Configure distributed lock | + +## Extension Methods on ConsumerBuilder + +| Method | Description | +|--------|-------------| +| `UseMongoDbInbox(Action)` | Configure inbox | + +## Features + +### Outbox + +Stores outgoing messages in a MongoDB collection within the same transaction as your business data, ensuring reliable message publishing. + +### Inbox + +Tracks processed messages in a MongoDB collection to ensure idempotent message handling. + +### Distributed Lock + +Uses MongoDB for coordinating message processing across multiple service instances. + +### Luggage Store + +Uses MongoDB GridFS to store large message payloads that exceed the message broker's size limits. + +## Full Example + +```csharp +services.AddFluentBrighter(brighter => brighter + .UseOutboxSweeper() + .UseOutboxArchiver(new NullOutboxArchiveProvider()) + .UsingRabbitMq(rabbitmq => rabbitmq + .SetConnection(conn => conn + .SetAmpq(amqp => amqp.SetUri("amqp://guest:guest@localhost:5672")) + .SetExchange(exchange => exchange.SetName("paramore.brighter.exchange"))) + .UsePublications(pb => pb + .AddPublication(p => p + .SetTopic("greeting.event.topic") + .CreateTopicIfMissing()) + .AddPublication(p => p + .SetTopic("farewell.event.topic") + .CreateTopicIfMissing())) + .UseSubscriptions(sb => sb + .AddSubscription(s => s + .SetSubscription("paramore.example.greeting") + .SetQueue("greeting.event.queue") + .SetTopic("greeting.event.topic") + .SetTimeout(TimeSpan.FromSeconds(200)) + .EnableDurable() + .EnableHighAvailability()) + .AddSubscription(s => s + .SetSubscription("paramore.example.farewell") + .SetQueue("farewell.event.queue") + .SetTopic("farewell.event.topic")))) + .UsingMongoDb(mongodb => mongodb + .SetConnection(c => c + .SetConnectionString("mongodb://root:example@localhost:27017") + .SetDatabaseName("brighter")) + .UseInbox("inbox") + .UseOutbox("outbox") + .UseDistributedLock("locking") + .UseLuggageStore("bucket"))); +``` diff --git a/docs/providers/mysql.md b/docs/providers/mysql.md new file mode 100644 index 0000000..350c485 --- /dev/null +++ b/docs/providers/mysql.md @@ -0,0 +1,85 @@ +# MySQL + +Fluent Brighter supports MySQL as an **outbox/inbox store** for use with other messaging providers. + +## Installation + +```bash +dotnet add package Fluent.Brighter.MySql +``` + +## Setup + +Use MySQL for the outbox/inbox pattern alongside a messaging provider: + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingRabbitMq(rabbitmq => /* ... messaging config ... */) + .Producers(producer => producer + .UseMySqlOutbox(cfg => cfg + .SetConnectionString("Server=localhost;Database=brighter;User=root;Password=password;") + .SetOutboxTableName("outbox")) + .UseMySqlDistributedLock("Server=localhost;Database=brighter;User=root;Password=password;")) + .Subscriptions(sub => sub + .UseMySqlInbox(cfg => cfg + .SetConnectionString("Server=localhost;Database=brighter;User=root;Password=password;") + .SetInboxTableName("inbox"))) + .UseOutboxSweeper()); +``` + +## Integrated Configuration + +You can also use the integrated `UsingMySql` configurator: + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingRabbitMq(rabbitmq => /* ... messaging config ... */) + .UsingMySql(mysql => mysql + .SetConnection(cfg => cfg + .SetConnectionString("Server=localhost;Database=brighter;User=root;Password=password;")) + .UseOutbox() + .UseInbox() + .UseDistributedLock())); +``` + +### Configurator Options + +| Method | Description | +|--------|-------------| +| `SetConnection(Action)` | Configure connection via builder | +| `SetConnection(RelationalDatabaseConfiguration)` | Set connection directly | +| `UseOutbox()` | Enable MySQL outbox | +| `UseInbox()` | Enable MySQL inbox | +| `UseDistributedLock()` | Enable distributed locking | + +## Extension Methods on ProducerBuilder + +| Method | Description | +|--------|-------------| +| `UseMySqlOutbox(Action)` | Configure outbox via builder | +| `UseMySqlOutbox(RelationalDatabaseConfiguration)` | Set outbox config directly | +| `UseMySqlOutbox(Action)` | Advanced outbox config | +| `UseMySqlDistributedLock(IAmARelationalDatabaseConfiguration)` | Enable distributed lock | +| `UseMySqlDistributedLock(string connectionString)` | Enable distributed lock with connection string | + +## Extension Methods on ConsumerBuilder + +| Method | Description | +|--------|-------------| +| `UseMySqlInbox(Action)` | Configure inbox via builder | +| `UseMySqlInbox(IAmARelationalDatabaseConfiguration)` | Set inbox config directly | +| `UseMySqlInbox(Action)` | Advanced inbox config | + +## Database Configuration + +All relational providers share the `RelationalDatabaseConfigurationBuilder`: + +| Method | Description | +|--------|-------------| +| `SetConnectionString(string)` | Database connection string | +| `SetDatabaseName(string)` | Database name | +| `SetOutboxTableName(string)` | Outbox table name | +| `SetInboxTableName(string)` | Inbox table name | +| `SetQueueStoreTable(string)` | Queue store table name | +| `SetSchemaName(string)` | Database schema | +| `SetBinaryMessagePayload(bool)` | Enable binary message payload | diff --git a/docs/providers/postgres.md b/docs/providers/postgres.md new file mode 100644 index 0000000..a9527a6 --- /dev/null +++ b/docs/providers/postgres.md @@ -0,0 +1,146 @@ +# PostgreSQL + +Fluent Brighter supports PostgreSQL both as a **message transport** (queues backed by database tables) and as an **outbox/inbox store** for use with other messaging providers. + +## Installation + +```bash +dotnet add package Fluent.Brighter.Postgres +``` + +## As a Message Transport + +Use PostgreSQL tables as message queues: + +```csharp +services.AddFluentBrighter(builder => +{ + builder + .UseDbTransactionOutboxArchive() + .UseOutboxSweeper() + .UsingPostgres(cfg => + { + cfg.SetConnection(new RelationalDatabaseConfiguration( + "Host=localhost;Username=postgres;Password=password;Database=brighter;")); + + cfg + .UseOutbox() + .UseInbox() + .UseDistributedLock() + .UsePublications(pp => pp + .AddPublication(p => p + .SetQueue("greeting.queue") + .CreateQueueIfMissing()) + .AddPublication(p => p + .SetQueue("farewell.queue") + .CreateQueueIfMissing())) + .UseSubscriptions(sb => sb + .AddSubscription(s => s + .SetQueue("greeting.queue") + .SetMessagePumpType(MessagePumpType.Reactor)) + .AddSubscription(s => s + .SetQueue("farewell.queue") + .SetMessagePumpType(MessagePumpType.Reactor))); + }); +}); +``` + +### Configurator Options + +| Method | Description | +|--------|-------------| +| `SetConnection(Action)` | Configure connection via builder | +| `SetConnection(RelationalDatabaseConfiguration)` | Set connection directly | +| `UseOutbox()` | Enable PostgreSQL outbox | +| `UseInbox()` | Enable PostgreSQL inbox | +| `UseDistributedLock()` | Enable distributed locking | +| `UsePublications(...)` | Configure message publications | +| `UseSubscriptions(...)` | Configure message subscriptions | + +### Publication Options + +| Method | Description | +|--------|-------------| +| `SetQueue(RoutingKey)` | Queue name | +| `CreateQueueIfMissing()` | Auto-create queue | +| `ValidIfQueueExists()` | Validate queue exists | +| `AssumeQueueExists()` | Assume queue exists | +| `SetSchemaName(string)` | Database schema name | +| `SetQueueStoreTable(string)` | Queue store table name | +| `EnableBinaryMessagePayload()` / `DisableBinaryMessagePayload()` | Binary payload mode | +| `SetSource(Uri)` | CloudEvents source | +| `SetDataSchema(Uri)` | CloudEvents data schema | +| `SetRequestType(Type)` | Request type | + +### Subscription Options + +| Method | Description | +|--------|-------------| +| `SetSubscription(SubscriptionName)` | Subscription name | +| `SetQueue(ChannelName)` | Queue to consume from | +| `SetMessagePumpType(MessagePumpType)` | `Reactor` or `Proactor` | +| `UseReactorMode()` / `UseProactorMode()` | Convenience pump type methods | +| `SetNumberOfPerformers(int)` | Concurrent consumers | +| `SetSchemaName(string)` | Database schema name | +| `SetQueueStoreTable(string)` | Queue store table name | +| `SetVisibleTimeout(TimeSpan)` | Message visibility timeout | +| `SetTableWithLargeMessage(bool)` | Enable large message support | +| `EnableBinaryMessagePayload()` / `DisableBinaryMessagePayload()` | Binary payload mode | +| `SetTimeout(TimeSpan)` | Processing timeout | +| `SetBufferSize(int)` | Internal buffer size | +| `SetRequeueCount(int)` | Max requeue attempts | +| `SetRequeueDelay(TimeSpan)` | Requeue delay | +| `SetMakeChannels(OnMissingChannel)` | Channel creation mode | + +## As an Outbox/Inbox Store + +Use PostgreSQL for the outbox/inbox pattern alongside another messaging provider (e.g., RabbitMQ): + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingRabbitMq(rabbitmq => /* ... RabbitMQ config ... */) + .Producers(producer => producer + .UsePostgresOutbox(cfg => cfg + .SetConnectionString("Host=localhost;Database=brighter;") + .SetOutboxTableName("outbox")) + .UsePostgresDistributedLock("Host=localhost;Database=brighter;")) + .Subscriptions(sub => sub + .UsePostgresInbox(cfg => cfg + .SetConnectionString("Host=localhost;Database=brighter;") + .SetInboxTableName("inbox"))) + .UseOutboxSweeper(cfg => cfg + .SetTimerInterval(30) + .SetBatchSize(100))); +``` + +### Extension Methods on ProducerBuilder + +| Method | Description | +|--------|-------------| +| `UsePostgresOutbox(Action)` | Configure outbox via builder | +| `UsePostgresOutbox(RelationalDatabaseConfiguration)` | Set outbox config directly | +| `UsePostgresOutbox(Action)` | Advanced outbox config | +| `UsePostgresDistributedLock(IAmARelationalDatabaseConfiguration)` | Enable distributed lock | +| `UsePostgresDistributedLock(string connectionString)` | Enable distributed lock with connection string | + +### Extension Methods on ConsumerBuilder + +| Method | Description | +|--------|-------------| +| `UsePostgresInbox(Action)` | Configure inbox via builder | +| `UsePostgresInbox(IAmARelationalDatabaseConfiguration)` | Set inbox config directly | +| `UsePostgresInbox(Action)` | Advanced inbox config | + +### Database Configuration Builder + +The `RelationalDatabaseConfigurationBuilder` is used across all relational providers: + +| Method | Description | +|--------|-------------| +| `SetConnectionString(string)` | Database connection string | +| `SetDatabaseName(string)` | Database name | +| `SetOutboxTableName(string)` | Outbox table name | +| `SetInboxTableName(string)` | Inbox table name | +| `SetQueueStoreTable(string)` | Queue store table name | +| `SetSchemaName(string)` | Database schema | +| `SetBinaryMessagePayload(bool)` | Enable binary message payload | diff --git a/docs/providers/rabbitmq.md b/docs/providers/rabbitmq.md new file mode 100644 index 0000000..b54ee61 --- /dev/null +++ b/docs/providers/rabbitmq.md @@ -0,0 +1,205 @@ +# RabbitMQ + +Fluent Brighter provides full support for RabbitMQ via the `Fluent.Brighter.RMQ.Async` (recommended) and `Fluent.Brighter.RMQ.Sync` packages. + +## Installation + +```bash +# Async (recommended) +dotnet add package Fluent.Brighter.RMQ.Async + +# Sync (if needed) +dotnet add package Fluent.Brighter.RMQ.Sync +``` + +## Basic Setup + +```csharp +using Fluent.Brighter; + +services.AddFluentBrighter(brighter => brighter + .UsingRabbitMq(rabbitmq => rabbitmq + .SetConnection(conn => conn + .SetAmpq(amqp => amqp.SetUri("amqp://guest:guest@localhost:5672")) + .SetExchange(exchange => exchange.SetName("my.exchange"))) + .UsePublications(pb => pb + .AddPublication(p => p + .SetTopic("greeting.event.topic") + .CreateTopicIfMissing())) + .UseSubscriptions(sb => sb + .AddSubscription(s => s + .SetSubscription("greeting-handler") + .SetQueue("greeting.event.queue") + .SetTopic("greeting.event.topic"))))); +``` + +## Connection Configuration + +Configure the RabbitMQ connection using `SetConnection`: + +```csharp +.SetConnection(conn => conn + .SetAmpq(amqp => amqp + .SetUri("amqp://guest:guest@localhost:5672") + .SetConnectionRetryCount(3) + .SetRetryWaitInMilliseconds(500) + .SetCircuitBreakTimeInMilliseconds(60000)) + .SetExchange(exchange => exchange + .SetName("my.exchange") + .SetType("topic") + .SetDurable(true)) + .SetName("my-connection") + .SetHeartbeat(60) + .SetPersistMessages(true) + .SetContinuationTimeout(120)) +``` + +### Connection Options + +| Method | Description | +|--------|-------------| +| `SetAmpq(...)` | Configure the AMQP URI and retry settings | +| `SetExchange(...)` | Configure the exchange name, type, and durability | +| `SetDeadLetterExchange(...)` | Configure a dead letter exchange | +| `SetName(string)` | Set the connection name | +| `SetHeartbeat(ushort)` | Set heartbeat interval in seconds | +| `SetPersistMessages(bool)` | Enable/disable message persistence | +| `SetContinuationTimeout(ushort)` | Set the continuation timeout | + +### AMQP URI Options + +| Method | Description | +|--------|-------------| +| `SetUri(Uri)` | The AMQP connection URI | +| `SetConnectionRetryCount(int)` | Number of connection retry attempts | +| `SetRetryWaitInMilliseconds(int)` | Delay between retries | +| `SetCircuitBreakTimeInMilliseconds(int)` | Circuit breaker timeout | + +### Exchange Options + +| Method | Description | +|--------|-------------| +| `SetName(string)` | Exchange name | +| `SetType(string)` | Exchange type (e.g., `"topic"`, `"direct"`, `"fanout"`) | +| `SetDurable(bool)` | Survive broker restarts | +| `SetSupportDelay(bool)` | Enable delayed message support | + +## Publications + +Configure message publications with `UsePublications`: + +```csharp +.UsePublications(pb => pb + .AddPublication(p => p + .SetTopic("greeting.event.topic") + .CreateTopicIfMissing() + .SetWaitForConfirmsTimeOutInMilliseconds(5000)) + .AddPublication(p => p + .SetTopic("farewell.event.topic") + .CreateTopicIfMissing())) +``` + +### Publication Options + +| Method | Description | +|--------|-------------| +| `SetTopic(RoutingKey)` | The routing key / topic name | +| `CreateTopicIfMissing()` | Auto-create the topic if it doesn't exist | +| `SetMakeChannels(OnMissingChannel)` | Channel creation mode | +| `SetSource(Uri)` | CloudEvents source URI | +| `SetDataSchema(Uri)` | CloudEvents data schema | +| `SetSubject(string)` | CloudEvents subject | +| `SetType(CloudEventsType)` | CloudEvents type | +| `SetDefaultHeaders(IDictionary)` | Default message headers | +| `SetReplyTo(string)` | Reply-to address | +| `SetWaitForConfirmsTimeOutInMilliseconds(int)` | Publisher confirm timeout | +| `SetRequestType(Type)` | The request type for this publication | + +## Subscriptions + +Configure message subscriptions with `UseSubscriptions`: + +```csharp +.UseSubscriptions(sb => sb + .AddSubscription(s => s + .SetSubscription("paramore.example.greeting") + .SetQueue("greeting.event.queue") + .SetTopic("greeting.event.topic") + .SetTimeout(TimeSpan.FromSeconds(200)) + .EnableDurable() + .EnableHighAvailability()) + .AddSubscription(s => s + .SetSubscription("paramore.example.farewell") + .SetQueue("farewell.event.queue") + .SetTopic("farewell.event.topic"))) +``` + +### Subscription Options + +| Method | Description | +|--------|-------------| +| `SetSubscription(SubscriptionName)` | Unique subscription identifier | +| `SetQueue(ChannelName)` | RabbitMQ queue name | +| `SetTopic(RoutingKey)` | Routing key to bind to | +| `SetTimeout(TimeSpan)` | Message processing timeout | +| `SetNumberOfPerformers(int)` | Number of concurrent handlers | +| `SetMessagePumpType(MessagePumpType)` | `Reactor` or `Proactor` | +| `SetBufferSize(int)` | Internal buffer size | +| `SetRequeueCount(int)` | Max requeue attempts | +| `SetRequeueDelay(TimeSpan)` | Delay between requeues | +| `SetUnacceptableMessageLimit(int)` | Max unacceptable messages before stopping | +| `SetMakeChannels(OnMissingChannel)` | Channel creation mode | +| `SetEmptyChannelDelay(TimeSpan)` | Delay when channel is empty | +| `SetChannelFailureDelay(TimeSpan)` | Delay after channel failure | +| `SetIsDurable(bool)` | Queue survives broker restart | +| `SethHighAvailability(bool)` | Enable HA mirroring | +| `SetMaxQueueLenght(int)` | Maximum queue length | +| `SetQueueType(QueueType)` | Queue type (classic, quorum) | +| `SetDeadLetterChannel(ChannelName)` | Dead letter queue | +| `SetDeadLetterRoutingKey(RoutingKey)` | Dead letter routing key | +| `SetTTL(TimeSpan)` | Message time-to-live | + +### Convenience Extension Methods + +```csharp +// Enable durable queue +.EnableDurable() + +// Enable high availability +.EnableHighAvailability() +``` + +## Full Example + +```csharp +services + .AddHostedService() + .AddFluentBrighter(brighter => brighter + .UsingRabbitMq(rabbitmq => rabbitmq + .SetConnection(conn => conn + .SetAmpq(amqp => amqp.SetUri("amqp://guest:guest@localhost:5672")) + .SetExchange(exchange => exchange.SetName("paramore.brighter.exchange"))) + .UsePublications(pb => pb + .AddPublication(p => p + .SetTopic("greeting.event.topic") + .CreateTopicIfMissing()) + .AddPublication(p => p + .SetTopic("farewell.event.topic") + .CreateTopicIfMissing())) + .UseSubscriptions(sb => sb + .AddSubscription(s => s + .SetSubscription("paramore.example.greeting") + .SetQueue("greeting.event.queue") + .SetTopic("greeting.event.topic") + .SetTimeout(TimeSpan.FromSeconds(200)) + .EnableDurable() + .EnableHighAvailability()) + .AddSubscription(s => s + .SetSubscription("paramore.example.farewell") + .SetQueue("farewell.event.queue") + .SetTopic("farewell.event.topic"))))); +``` + +## Async vs Sync + +The `Fluent.Brighter.RMQ.Async` package uses the asynchronous RabbitMQ client and is recommended for most applications. The `Fluent.Brighter.RMQ.Sync` package provides the same API but uses the synchronous client. Both packages have identical builder APIs. diff --git a/docs/providers/redis.md b/docs/providers/redis.md new file mode 100644 index 0000000..5ef3b2b --- /dev/null +++ b/docs/providers/redis.md @@ -0,0 +1,155 @@ +# Redis + +Fluent Brighter provides support for Redis Streams as a messaging transport. + +## Installation + +```bash +dotnet add package Fluent.Brighter.Redis +``` + +## Basic Setup + +```csharp +using Fluent.Brighter; + +services.AddFluentBrighter(brighter => brighter + .UsingRedis(redis => redis + .SetConnection(c => c + .SetRedisConnectionString("redis://localhost:6379") + .SetMaxPoolSize(10) + .SetMessageTimeToLive(TimeSpan.FromMinutes(10))) + .UsePublications(pb => pb + .AddPublication(p => p + .SetTopic("greeting.topic"))) + .UseSubscriptions(sb => sb + .AddSubscription(s => s + .SetTopic("greeting.topic") + .SetQueue("greeting.queue") + .SetMessagePumpType(MessagePumpType.Reactor))))); +``` + +## Connection Configuration + +```csharp +.SetConnection(c => c + .SetRedisConnectionString("redis://localhost:6379?ConnectTimeout=1000&SendTimeout=1000") + .SetMaxPoolSize(10) + .SetMessageTimeToLive(TimeSpan.FromMinutes(10)) + .SetDefaultRetryTimeout(3_000) + .SetDefaultConnectTimeout(5_000) + .SetDefaultSendTimeout(5_000) + .SetDefaultReceiveTimeout(5_000)) +``` + +### Connection Options + +| Method | Description | +|--------|-------------| +| `SetRedisConnectionString(string)` | Redis connection string | +| `SetMaxPoolSize(int)` | Maximum connection pool size | +| `SetMessageTimeToLive(TimeSpan)` | Default message TTL | +| `SetDefaultConnectTimeout(int)` | Connect timeout in ms | +| `SetDefaultSendTimeout(int)` | Send timeout in ms | +| `SetDefaultReceiveTimeout(int)` | Receive timeout in ms | +| `SetDefaultIdleTimeOutSecs(int)` | Idle timeout in seconds | +| `SetDefaultRetryTimeout(int)` | Retry timeout in ms | +| `SetBufferPoolMaxSize(int)` | Buffer pool max size | +| `SetHostLookupTimeoutMs(int)` | Host lookup timeout in ms | +| `SetAssumeServerVersion(int)` | Assume Redis server version | +| `SetDeactivatedClientsExpiry(TimeSpan)` | Deactivated client expiry | +| `SetDisableVerboseLogging(bool)` | Disable verbose logging | +| `SetBackoffMultiplier(int)` | Backoff multiplier | +| `SetVerifyMasterConnections(bool)` | Verify master connections | + +## Publications + +```csharp +.UsePublications(pb => pb + .AddPublication(p => p + .SetTopic("greeting.topic") + .CreateTopicIfMissing()) + .AddPublication(p => p + .SetTopic("farewell.topic"))) +``` + +### Publication Options + +| Method | Description | +|--------|-------------| +| `SetTopic(RoutingKey)` | Redis stream/topic name | +| `CreateTopicIfMissing()` | Auto-create topic if missing | +| `ValidIfTopicExists()` | Validate topic exists | +| `AssumeTopicExists()` | Assume topic exists | +| `SetSource(Uri)` | CloudEvents source | +| `SetDataSchema(Uri)` | CloudEvents data schema | +| `SetSubject(string)` | CloudEvents subject | +| `SetType(CloudEventsType)` | CloudEvents type | +| `SetDefaultHeaders(IDictionary)` | Default message headers | +| `SetReplyTo(string)` | Reply-to address | +| `SetRequestType(Type)` | Request type | + +## Subscriptions + +```csharp +.UseSubscriptions(sb => sb + .AddSubscription(s => s + .SetTopic("greeting.topic") + .SetQueue("greeting.queue") + .SetMessagePumpType(MessagePumpType.Reactor) + .SetNumberOfPerformers(3) + .SetTimeout(TimeSpan.FromSeconds(30)))) +``` + +### Subscription Options + +| Method | Description | +|--------|-------------| +| `SetSubscription(SubscriptionName)` | Unique subscription identifier | +| `SetQueue(ChannelName)` | Consumer group / queue name | +| `SetTopic(RoutingKey)` | Redis stream to consume | +| `SetMessagePumpType(MessagePumpType)` | `Reactor` or `Proactor` | +| `UseReactorMode()` / `UseProactorMode()` | Convenience methods for pump type | +| `SetNumberOfPerformers(int)` | Number of concurrent consumers | +| `SetTimeout(TimeSpan)` | Processing timeout | +| `SetBufferSize(int)` | Internal buffer size | +| `SetRequeueCount(int)` | Max requeue attempts | +| `SetRequeueDelay(TimeSpan)` | Requeue delay | +| `SetMakeChannels(OnMissingChannel)` | Channel creation mode | +| `CreateInfrastructureIfMissing()` | Auto-create infrastructure | +| `ValidIfInfrastructureExists()` | Validate infrastructure exists | +| `AssumeInfrastructureExists()` | Assume infrastructure exists | +| `SetEmptyChannelDelay(TimeSpan)` | Delay when channel is empty | +| `SetChannelFailureDelay(TimeSpan)` | Delay after channel failure | +| `SetUnacceptableMessageLimit(int)` | Max unacceptable messages | + +## Full Example + +```csharp +services.AddFluentBrighter(builder => +{ + builder + .UsingRedis(cfg => + { + cfg.SetConnection(c => c + .SetRedisConnectionString("redis://localhost:6379?ConnectTimeout=1000&SendTimeout=1000") + .SetMaxPoolSize(10) + .SetMessageTimeToLive(TimeSpan.FromMinutes(10)) + .SetDefaultRetryTimeout(3_000)); + + cfg + .UsePublications(pp => pp + .AddPublication(p => p.SetTopic("greeting.topic")) + .AddPublication(p => p.SetTopic("farewell.topic"))) + .UseSubscriptions(sb => sb + .AddSubscription(s => s + .SetTopic("greeting.topic") + .SetQueue("greeting.queue") + .SetMessagePumpType(MessagePumpType.Reactor)) + .AddSubscription(s => s + .SetTopic("farewell.topic") + .SetQueue("farewell.queue") + .SetMessagePumpType(MessagePumpType.Reactor))); + }); +}); +``` diff --git a/docs/providers/rocketmq.md b/docs/providers/rocketmq.md new file mode 100644 index 0000000..fd62625 --- /dev/null +++ b/docs/providers/rocketmq.md @@ -0,0 +1,144 @@ +# Apache RocketMQ + +Fluent Brighter provides support for Apache RocketMQ as a messaging transport. + +## Installation + +```bash +dotnet add package Fluent.Brighter.RocketMQ +``` + +## Basic Setup + +```csharp +using Fluent.Brighter; + +services.AddFluentBrighter(brighter => brighter + .UsingRocketMq(rocket => rocket + .SetConnection(conn => conn + .SetClient(c => c + .SetEndpoints("localhost:8081") + .EnableSsl(false) + .SetRequestTimeout(TimeSpan.FromSeconds(10)) + .Build())) + .UsePublications(pub => pub + .AddPublication(p => p + .SetTopic("greeting"))) + .UseSubscriptions(sub => sub + .AddSubscription(s => s + .SetSubscriptionName("greeting-sub-name") + .SetTopic("greeting") + .SetConsumerGroup("greeting-consumer-group") + .UseReactorMode())))); +``` + +## Connection Configuration + +```csharp +.SetConnection(conn => conn + .SetClient(c => c + .SetEndpoints("localhost:8081") + .EnableSsl(false) + .SetRequestTimeout(TimeSpan.FromSeconds(10)) + .Build()) + .SetMaxAttempts(3) + .SetInstrumentation(InstrumentationOptions.All)) +``` + +### Connection Options + +| Method | Description | +|--------|-------------| +| `SetClient(ClientConfig)` | RocketMQ client configuration | +| `SetTimerProvider(TimeProvider)` | Custom time provider | +| `SetMaxAttempts(int)` | Maximum retry attempts | +| `SetTransactionChecker(ITransactionChecker)` | Transaction checker | +| `SetInstrumentation(InstrumentationOptions)` | Instrumentation level | + +## Publications + +```csharp +.UsePublications(pub => pub + .AddPublication(p => p + .SetTopic("greeting") + .SetTags("greeting-tag") + .AssumeTopicExists())) +``` + +### Publication Options + +| Method | Description | +|--------|-------------| +| `SetTopic(RoutingKey)` | RocketMQ topic name | +| `SetTags(string)` | Message tags for filtering | +| `ValidIfTopicExists()` | Validate topic exists | +| `AssumeTopicExists()` | Assume topic exists | +| `SetInstrumentation(InstrumentationOptions)` | Per-publication instrumentation | +| `SetSource(Uri)` | CloudEvents source | +| `SetDataSchema(Uri)` | CloudEvents data schema | +| `SetSubject(string)` | CloudEvents subject | +| `SetType(CloudEventsType)` | CloudEvents type | +| `SetDefaultHeaders(IDictionary)` | Default message headers | +| `SetReplyTo(string)` | Reply-to address | +| `SetRequestType(Type)` | Request type | +| `SetMakeChannels(OnMissingChannel)` | Channel creation mode | + +## Subscriptions + +```csharp +.UseSubscriptions(sub => sub + .AddSubscription(s => s + .SetSubscriptionName("greeting-sub-name") + .SetTopic("greeting") + .SetConsumerGroup("greeting-consumer-group") + .UseReactorMode() + .SetNoOfPerformers(5) + .SetFilter(new FilterExpression("greeting-tag")))) +``` + +### Subscription Options + +| Method | Description | +|--------|-------------| +| `SetSubscriptionName(SubscriptionName)` | Unique subscription name | +| `SetTopic(RoutingKey)` | RocketMQ topic to consume | +| `SetConsumerGroup(string)` | Consumer group name | +| `SetNoOfPerformers(int)` | Number of concurrent handlers | +| `SetMessagePumpType(MessagePumpType)` | `Reactor` or `Proactor` | +| `UseReactorMode()` / `UseProactorMode()` | Convenience methods for pump type | +| `SetFilter(FilterExpression)` | Message filter expression | +| `SetTimeout(TimeSpan)` | Processing timeout | +| `SetBufferSize(int)` | Internal buffer size | +| `SetRequeueCount(int)` | Max requeue attempts | +| `SetRequeueDelay(TimeSpan)` | Requeue delay | +| `SetUnacceptableMessageLimit(int)` | Max unacceptable messages | +| `SetMakeChannels(OnMissingChannel)` | Channel creation mode | +| `CreateInfrastructureIfMissing()` | Auto-create infrastructure | +| `ValidIfInfrastructureExists()` | Validate infrastructure exists | +| `AssumeInfrastructureExists()` | Assume infrastructure exists | +| `SetReceiveMessageTimeout(TimeSpan)` | Receive timeout | +| `SetInvisibilityTimeout(TimeSpan)` | Message invisibility timeout | +| `SetEmptyChannelDelay(TimeSpan)` | Delay when channel is empty | +| `SetChannelFailureDelay(TimeSpan)` | Delay after channel failure | + +## Full Example + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingRocketMq(rocket => rocket + .SetConnection(conn => conn + .SetClient(c => c + .SetEndpoints("localhost:8081") + .EnableSsl(false) + .SetRequestTimeout(TimeSpan.FromSeconds(10)) + .Build())) + .UsePublications(pub => pub + .AddPublication(p => p + .SetTopic("greeting"))) + .UseSubscriptions(sub => sub + .AddSubscription(s => s + .SetSubscriptionName("greeting-sub-name") + .SetTopic("greeting") + .SetConsumerGroup("greeting-consumer-group") + .UseReactorMode())))); +``` diff --git a/docs/providers/sqlite.md b/docs/providers/sqlite.md new file mode 100644 index 0000000..8f848a8 --- /dev/null +++ b/docs/providers/sqlite.md @@ -0,0 +1,82 @@ +# SQLite + +Fluent Brighter supports SQLite as an **outbox/inbox store** for use with other messaging providers. This is particularly useful for development and testing scenarios. + +## Installation + +```bash +dotnet add package Fluent.Brighter.Sqlite +``` + +## Setup + +Use SQLite for the outbox/inbox pattern alongside a messaging provider: + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingRabbitMq(rabbitmq => /* ... messaging config ... */) + .Producers(producer => producer + .UseSqliteOutbox(cfg => cfg + .SetConnectionString("Data Source=brighter.db") + .SetOutboxTableName("outbox"))) + .Subscriptions(sub => sub + .UseSqliteInbox(cfg => cfg + .SetConnectionString("Data Source=brighter.db") + .SetInboxTableName("inbox"))) + .UseOutboxSweeper()); +``` + +## Integrated Configuration + +You can also use the integrated `UsingSqlite` configurator: + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingRabbitMq(rabbitmq => /* ... messaging config ... */) + .UsingSqlite(sqlite => sqlite + .SetConnection(cfg => cfg + .SetConnectionString("Data Source=brighter.db")) + .UseOutbox() + .UseInbox())); +``` + +### Configurator Options + +| Method | Description | +|--------|-------------| +| `SetConnection(Action)` | Configure connection via builder | +| `SetConnection(RelationalDatabaseConfiguration)` | Set connection directly | +| `UseOutbox()` | Enable SQLite outbox | +| `UseInbox()` | Enable SQLite inbox | + +> **Note:** SQLite does not support distributed locking due to its single-writer nature. + +## Extension Methods on ProducerBuilder + +| Method | Description | +|--------|-------------| +| `UseSqliteOutbox(Action)` | Configure outbox via builder | +| `UseSqliteOutbox(RelationalDatabaseConfiguration)` | Set outbox config directly | +| `UseSqliteOutbox(Action)` | Advanced outbox config | + +## Extension Methods on ConsumerBuilder + +| Method | Description | +|--------|-------------| +| `UseSqliteInbox(Action)` | Configure inbox via builder | +| `UseSqliteInbox(IAmARelationalDatabaseConfiguration)` | Set inbox config directly | +| `UseSqliteInbox(Action)` | Advanced inbox config | + +## Database Configuration + +All relational providers share the `RelationalDatabaseConfigurationBuilder`: + +| Method | Description | +|--------|-------------| +| `SetConnectionString(string)` | Database connection string | +| `SetDatabaseName(string)` | Database name | +| `SetOutboxTableName(string)` | Outbox table name | +| `SetInboxTableName(string)` | Inbox table name | +| `SetQueueStoreTable(string)` | Queue store table name | +| `SetSchemaName(string)` | Database schema | +| `SetBinaryMessagePayload(bool)` | Enable binary message payload | diff --git a/docs/providers/sqlserver.md b/docs/providers/sqlserver.md new file mode 100644 index 0000000..f4366e5 --- /dev/null +++ b/docs/providers/sqlserver.md @@ -0,0 +1,122 @@ +# SQL Server + +Fluent Brighter supports SQL Server both as a **message transport** (queues backed by database tables) and as an **outbox/inbox store** for use with other messaging providers. + +## Installation + +```bash +dotnet add package Fluent.Brighter.SqlServer +``` + +## As a Message Transport + +Use SQL Server tables as message queues: + +```csharp +services.AddFluentBrighter(builder => +{ + builder + .UseDbTransactionOutboxArchive() + .UseOutboxSweeper() + .UsingMicrosoftSqlServer(cfg => + { + cfg.SetConnection(new RelationalDatabaseConfiguration( + "Server=localhost;Database=BrighterTests;", + databaseName: "BrighterTests", + queueStoreTable: "QueueData")); + + cfg + .UseOutbox() + .UseInbox() + .UseDistributedLock() + .UsePublications(pp => pp + .AddPublication(p => p + .SetQueue("greeting.queue")) + .AddPublication(p => p + .SetQueue("farewell.queue"))) + .UseSubscriptions(sb => sb + .AddSubscription(s => s + .SetQueue("greeting.queue") + .SetMessagePumpType(MessagePumpType.Reactor)) + .AddSubscription(s => s + .SetQueue("farewell.queue") + .SetMessagePumpType(MessagePumpType.Reactor))); + }); +}); +``` + +### Configurator Options + +| Method | Description | +|--------|-------------| +| `SetConnection(Action)` | Configure connection via builder | +| `SetConnection(RelationalDatabaseConfiguration)` | Set connection directly | +| `UseOutbox()` | Enable SQL Server outbox | +| `UseOutbox(Action)` | Outbox with custom config | +| `UseInbox()` | Enable SQL Server inbox | +| `UseInbox(Action)` | Inbox with custom config | +| `UseDistributedLock()` | Enable distributed locking | +| `UsePublications(...)` | Configure message publications | +| `UseSubscriptions(...)` | Configure message subscriptions | + +### Publication Options + +| Method | Description | +|--------|-------------| +| `SetQueue(RoutingKey)` | Queue name | +| `SetSource(Uri)` | CloudEvents source | +| `SetDataSchema(Uri)` | CloudEvents data schema | +| `SetRequestType(Type)` | Request type | +| `SetMakeChannels(OnMissingChannel)` | Channel creation mode | + +### Subscription Options + +| Method | Description | +|--------|-------------| +| `SetSubscription(SubscriptionName)` | Subscription name | +| `SetQueue(ChannelName)` | Queue to consume from | +| `SetMessagePumpType(MessagePumpType)` | `Reactor` or `Proactor` | +| `SetNumberOfPerformers(int)` | Concurrent consumers | +| `SetTimeout(TimeSpan)` | Processing timeout | +| `SetBufferSize(int)` | Internal buffer size | +| `SetRequeueCount(int)` | Max requeue attempts | +| `SetRequeueDelay(TimeSpan)` | Requeue delay | +| `SetMakeChannels(OnMissingChannel)` | Channel creation mode | +| `SetEmptyChannelDelay(TimeSpan)` | Delay when channel is empty | +| `SetChannelFailureDelay(TimeSpan)` | Delay after channel failure | + +## As an Outbox/Inbox Store + +Use SQL Server for the outbox/inbox pattern alongside another messaging provider: + +```csharp +services.AddFluentBrighter(brighter => brighter + .UsingRabbitMq(rabbitmq => /* ... RabbitMQ config ... */) + .Producers(producer => producer + .UseMicrosoftSqlServerOutbox(cfg => cfg + .SetConnectionString("Server=localhost;Database=Brighter;") + .SetOutboxTableName("Outbox")) + .UseMicrosoftSqlServerDistributedLock("Server=localhost;Database=Brighter;")) + .Subscriptions(sub => sub + .UseMicrosoftSqlServerInbox(cfg => cfg + .SetConnectionString("Server=localhost;Database=Brighter;") + .SetInboxTableName("Inbox")))); +``` + +### Extension Methods on ProducerBuilder + +| Method | Description | +|--------|-------------| +| `UseMicrosoftSqlServerOutbox(Action)` | Configure outbox via builder | +| `UseMicrosoftSqlServerOutbox(RelationalDatabaseConfiguration)` | Set outbox config directly | +| `UseMicrosoftSqlServerOutbox(Action)` | Advanced outbox config | +| `UseMicrosoftSqlServerDistributedLock(IAmARelationalDatabaseConfiguration)` | Enable distributed lock | +| `UseMicrosoftSqlServerDistributedLock(string connectionString)` | Enable distributed lock with connection string | + +### Extension Methods on ConsumerBuilder + +| Method | Description | +|--------|-------------| +| `UseMicrosoftSqlServerInbox(Action)` | Configure inbox via builder | +| `UseMicrosoftSqlServerInbox(IAmARelationalDatabaseConfiguration)` | Set inbox config directly | +| `UseMicrosoftSqlServerInbox(Action)` | Advanced inbox config |