#392 Host.Outbox.PostgreSql

Signed-off-by: Richard Pringle <[email protected]>

Author: EtherZa

README.md

@@ -167,6 +167,8 @@ The configuration can be [modularized](docs/intro.md#modularization-of-configura
 | `.Host.AspNetCore`                   | Integration for ASP.NET Core                                                                                        | [![NuGet](https://img.shields.io/nuget/v/SlimMessageBus.Host.AspNetCore.svg)](https://www.nuget.org/packages/SlimMessageBus.Host.AspNetCore)                                     |
 | `.Host.Interceptor`                  | Core interface for interceptors                                                                                     | [![NuGet](https://img.shields.io/nuget/v/SlimMessageBus.Host.Interceptor.svg)](https://www.nuget.org/packages/SlimMessageBus.Host.Interceptor)                                   |
 | `.Host.FluentValidation`             | Validation for messages based on [FluentValidation](https://www.nuget.org/packages/FluentValidation)                | [![NuGet](https://img.shields.io/nuget/v/SlimMessageBus.Host.FluentValidation.svg)](https://www.nuget.org/packages/SlimMessageBus.Host.FluentValidation)                         |
+| `.Host.Outbox.PostgreSql`            | Transactional Outbox using PostgreSQL                                                                               | [![NuGet](https://img.shields.io/nuget/v/SlimMessageBus.Host.Outbox.PostgreSql.svg)](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.PostgreSql)                       |
+| `.Host.Outbox.PostgreSql.DbContext`  | Transactional Outbox using PostgreSQL with EF DataContext integration                                               | [![NuGet](https://img.shields.io/nuget/v/SlimMessageBus.Host.Outbox.PostgreSql.DbContext.svg)](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.PostgreSql.DbContext)   |
 | `.Host.Outbox.Sql`                   | Transactional Outbox using MSSQL                                                                                    | [![NuGet](https://img.shields.io/nuget/v/SlimMessageBus.Host.Outbox.Sql.svg)](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.Sql)                                     |
 | `.Host.Outbox.Sql.DbContext`         | Transactional Outbox using MSSQL with EF DataContext integration                                                    | [![NuGet](https://img.shields.io/nuget/v/SlimMessageBus.Host.Outbox.Sql.DbContext.svg)](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.Sql.DbContext)                 |
 | `.Host.AsyncApi`                     | [AsyncAPI](https://www.asyncapi.com/) specification generation via [Saunter](https://github.com/tehmantra/saunter)  | [![NuGet](https://img.shields.io/nuget/v/SlimMessageBus.Host.AsyncApi.svg)](https://www.nuget.org/packages/SlimMessageBus.Host.AsyncApi)                                         |

docs/README.md

@@ -13,7 +13,12 @@
   - [RabbitMQ](provider_rabbitmq.md)
   - [Redis](provider_redis.md)
   - [SQL](provider_sql.md)
-- [Serialization Plugins](serialization.md)
+- Plugins
+  - [Serialization](serialization.md)
+  - [Transactional Outbox](plugin_outbox.md)
+  - [Validation using FluentValidation](plugin_fluent_validation.md)
+  - [AsyncAPI specification generation](plugin_asyncapi.md)
+  - [Consumer Circuit Breaker](intro.md#health-check-circuit-breaker)
 - [Use Cases](UseCases/)
 - For Maintainers:
   - [Build & Test](Maintainers/build.md)

docs/plugin_outbox.md

@@ -5,11 +5,16 @@ Please read the [Introduction](intro.md) before reading this provider documentat
 - [Introduction](#introduction)
 - [Configuration](#configuration)
   - [Entity Framework](#entity-framework)
-  - [SQL Connection](#sql-connection)
+    - [PostgreSQL](#postgresql)
+    - [SQL Server](#sql-server)
+  - [Direct Connection](#direct-connection)
+    - [PostgreSQL](#postgresql-1)
+    - [SQL Server](#sql-server-1)
 - [Options](#options)
   - [UseOutbox for Producers](#useoutbox-for-producers)
   - [Transactions for Consumers](#transactions-for-consumers)
     - [UseTransactionScope](#usetransactionscope)
+    - [UsePostgreSqlTransaction](#usepostgresqltransaction)
     - [UseSqlTransaction](#usesqltransaction)
 - [How it works](#how-it-works)
 - [Clean up](#clean-up)
@@ -18,16 +23,29 @@ Please read the [Introduction](intro.md) before reading this provider documentat
 ## Introduction
 
 The [`Host.Outbox`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox) introduces [Transactional Outbox](https://microservices.io/patterns/data/transactional-outbox.html) pattern to the SlimMessageBus.
-It comes in two flavors:
 
+PostgreSQL
+- [`Host.Outbox.PostgreSql`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.PostgreSql) as integration with the [Npgsql](https://www.npgsql.org/) client
+- [`Host.Outbox.PostgreSql.DbContext`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.PostgreSql.DbContext) as integration with Entity Framework Core using Npgsql
+
+SQL server
 - [`Host.Outbox.Sql`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.Sql) as integration with the System.Data.Sql client (MSSQL)
-- [`Host.Outbox.Sql.DbContext`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.Sql.DbContext) as integration with Entity Framework Core
+- [`Host.Outbox.Sql.DbContext`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.Sql.DbContext) as integration with Entity Framework Core using System.Data.Sql
 
 Outbox plugin can work in combination with any transport provider.
 
 ## Configuration
 
 ### Entity Framework
+#### PostgreSQL
+
+> Required: [`SlimMessageBus.Host.Outbox.PostgreSql.DbContext`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.PostgreSql.DbContext)
+
+```cs
+using SlimMessageBus.Host.Outbox.PostgreSql.DbContext;
+```
+
+#### SQL Server
 
 > Required: [`SlimMessageBus.Host.Outbox.Sql.DbContext`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.Sql.DbContext)
 
@@ -40,7 +58,7 @@ Consider the following example (from [Samples](../src/Samples/Sample.OutboxWebAp
 - `services.AddOutboxUsingDbContext<CustomerContext>(...)` is used to add the [Outbox.DbContext](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.Sql.DbContext) plugin to the container.
 - `CustomerContext` is the application specific Entity Framework `DbContext`.
 - `CustomerCreatedEvent` is produced on the `AzureSB` child bus, the bus will deliver these events via outbox - see `.UseOutbox()`
-- `CreateCustomerCommand` is consumed on the `Memory` child bus, each command is wrapped in an SQL transaction - see `UseSqlTransaction()`
+- `CreateCustomerCommand` is consumed on the `Memory` child bus, each command is wrapped in an SQL transaction - see `UsePostgreSqlTransaction()` / `UseSqlTransaction()`
 
 Startup setup:
 
@@ -52,9 +70,20 @@ builder.Services.AddSlimMessageBus(mbb =>
         .AddChildBus("Memory", mbb =>
         {
             mbb.WithProviderMemory()
-                .AutoDeclareFrom(Assembly.GetExecutingAssembly(), consumerTypeFilter: t => t.Name.EndsWith("CommandHandler"))
-                //.UseTransactionScope(messageTypeFilter: t => t.Name.EndsWith("Command")) // Consumers/Handlers will be wrapped in a TransactionScope
-                .UseSqlTransaction(messageTypeFilter: t => t.Name.EndsWith("Command")); // Consumers/Handlers will be wrapped in a SqlTransaction ending with Command
+                .AutoDeclareFrom(Assembly.GetExecutingAssembly(), consumerTypeFilter: t => t.Name.EndsWith("CommandHandler"));
+            //.UseTransactionScope(messageTypeFilter: t => t.Name.EndsWith("Command")) // Consumers/Handlers will be wrapped in a TransactionScope
+            //.UseSqlTransaction(messageTypeFilter: t => t.Name.EndsWith("Command")); // Consumers/Handlers will be wrapped in a SqlTransaction ending with Command
+
+            switch (dbProvider)
+            {
+                case DbProvider.SqlServer:
+                    mbb.UseSqlTransaction(messageTypeFilter: t => t.Name.EndsWith("Command")); // Consumers/Handlers will be wrapped in a SqlTransaction ending with Command
+                    break;
+
+                case DbProvider.PostgreSql:
+                    mbb.UsePostgreSqlTransaction(messageTypeFilter: t => t.Name.EndsWith("Command")); // Consumers/Handlers will be wrapped in a SqlTransaction ending with Command
+                    break;
+            }
         })
         .AddChildBus("AzureSB", mbb =>
         {
@@ -81,20 +110,40 @@ builder.Services.AddSlimMessageBus(mbb =>
                     // x.UseOutbox();
                 })
                 // All outgoing messages from this bus will go out via an outbox
-                .UseOutbox(/* messageTypeFilter: t => t.Name.EndsWith("Command") */); // Additionaly, can apply filter do determine messages that should go out via outbox                
+                .UseOutbox(/* messageTypeFilter: t => t.Name.EndsWith("Command") */); // Additionally, can apply filter do determine messages that should go out via outbox                
         })
         .AddServicesFromAssembly(Assembly.GetExecutingAssembly())
         .AddJsonSerializer()
-        .AddAspNet()
-        .AddOutboxUsingDbContext<CustomerContext>(opts =>
-        {
-            opts.PollBatchSize = 100;
-            opts.PollIdleSleep = TimeSpan.FromSeconds(10);
-            opts.MessageCleanup.Interval = TimeSpan.FromSeconds(10);
-            opts.MessageCleanup.Age = TimeSpan.FromMinutes(1);
-            //opts.SqlSettings.TransactionIsolationLevel = System.Data.IsolationLevel.RepeatableRead;
-            //opts.SqlSettings.Dialect = SqlDialect.SqlServer;
-        });
+        .AddAspNet();
+
+    switch (dbProvider)
+    {
+        case DbProvider.SqlServer:
+            SlimMessageBus.Host.Outbox.Sql.DbContext.MessageBusBuilderExtensions.AddOutboxUsingDbContext<CustomerContext>(mbb, opts =>
+            {
+                opts.PollBatchSize = 500;
+                opts.PollIdleSleep = TimeSpan.FromSeconds(10);
+                opts.MessageCleanup.Interval = TimeSpan.FromSeconds(10);
+                opts.MessageCleanup.Age = TimeSpan.FromMinutes(1);
+                //opts.SqlSettings.TransactionIsolationLevel = System.Data.IsolationLevel.RepeatableRead;
+                //opts.SqlSettings.Dialect = SqlDialect.SqlServer;
+            });
+
+            break;
+
+        case DbProvider.PostgreSql:
+            SlimMessageBus.Host.Outbox.PostgreSql.DbContext.MessageBusBuilderExtensions.AddOutboxUsingDbContext<CustomerContext>(mbb, opts =>
+            {
+                opts.PollBatchSize = 500;
+                opts.PollIdleSleep = TimeSpan.FromSeconds(10);
+                opts.MessageCleanup.Interval = TimeSpan.FromSeconds(10);
+                opts.MessageCleanup.Age = TimeSpan.FromMinutes(1);
+                //opts.SqlSettings.TransactionIsolationLevel = System.Data.IsolationLevel.RepeatableRead;
+                //opts.SqlSettings.Dialect = SqlDialect.SqlServer;
+            });
+
+            break;
+    }
 });
 ```
 
@@ -119,7 +168,16 @@ public record CreateCustomerCommandHandler(IMessageBus Bus, CustomerContext Cust
 }
 ```
 
-### SQL Connection
+### Direct Connection
+#### PostgreSQL
+
+> Required: [`SlimMessageBus.Host.Outbox.PostgreSql`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.PostgreSql)
+
+```cs
+using SlimMessageBus.Host.Outbox.PostgreSql;
+```
+
+#### SQL Server
 
 > Required: [`SlimMessageBus.Host.Outbox.Sql`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.Sql)
 
@@ -129,7 +187,7 @@ using SlimMessageBus.Host.Outbox.Sql;
 
 Consider the following example:
 
-- `services.AddMessageBusOutboxUsingSql(...)` is used to add the [Outbox.Sql](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.Sql) plugin to the container.
+- `services.AddOutboxUsingSql(...)` is used to add the [Outbox.Sql](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.Sql) plugin to the container.
 - `SqlConnection` is registered in the container
 
 ```cs
@@ -181,6 +239,20 @@ using SlimMessageBus.Host.Outbox;
 
 When applied on the (child) bus level then all consumers (or handlers) will inherit that option.
 
+#### UsePostgreSqlTransaction
+
+> Required: [`SlimMessageBus.Host.Outbox.PostgreSql`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.PostgreSql) or [`SlimMessageBus.Host.Outbox.PostgreSql.DbContext`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.PostgreSql.DbContext)
+
+```cs
+using SlimMessageBus.Host.Outbox.PostgreSql;
+```
+
+`.UsePostgreSqlTransaction()` can be used on consumers (or handlers) declaration to force the consumer to start a `PostgreSqlTransaction` prior the message `OnHandle` and to complete that transaction after it. Any exception raised by the consumer would cause the transaction to be rolled back.
+
+When applied on the (child) bus level then all consumers (or handlers) will inherit that option.
+
+`PostgreSqlTransaction`-s are created off the associated `NpgsqlConnection`.
+
 #### UseSqlTransaction
 
 > Required: [`SlimMessageBus.Host.Outbox.Sql`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.Sql) or [`SlimMessageBus.Host.Outbox.Sql.DbContext`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.Sql.DbContext)
@@ -204,7 +276,7 @@ When applied on the (child) bus level then all consumers (or handlers) will inhe
 - When a message is sent via a bus or producer marked with `.UseOutbox()` then such message will be inserted into the `Outbox` table.
   It is important that message publish happens in the context of an transaction to ensure consistency.
 
-- When the message publication happens in the context of a consumer (or handler) of another message, the `.UseTransactionScope()`, `.UseSqlTransaction()` can be used to start a transaction.
+- When the message publication happens in the context of a consumer (or handler) of another message, the `.UseTransactionScope()`, `.UseSqlTransaction()` or `.UseSqlTransaction()` can be used to start a transaction.
 
 - The transaction can be managed by the application, starting it either explicitly using `DbContext.Database.BeginTransactionAsync()` or creating a `TransactionScope()`.