Merge pull request #138 from TraGicCode/chore/add-support-for-postgres-sql-transport

TraGicCode · web-flow · commit fe015df31e2a · 2025-12-06T15:47:00.000-06:00
Implement Postgre Sql transport
diff --git a/src/BuslyCLI.Console/BuslyCLI.Console.csproj b/src/BuslyCLI.Console/BuslyCLI.Console.csproj
@@ -34,6 +34,7 @@
     <PackageReference Include="NServiceBus.RabbitMQ" Version="10.1.6" />
     <PackageReference Include="NServiceBus.Transport.AzureServiceBus" Version="5.1.2" />
     <PackageReference Include="NServiceBus.Transport.AzureStorageQueues" Version="13.0.3" />
+    <PackageReference Include="NServiceBus.Transport.PostgreSql" Version="8.1.10" />
     <PackageReference Include="NServiceBus.Transport.SqlServer" Version="8.1.10" />
     <PackageReference Include="Spectre.Console.Cli" Version="0.53.1" />
     <PackageReference Include="Spectre.Console.Cli.Extensions.DependencyInjection" Version="0.20.0" />
diff --git a/src/BuslyCLI.Console/Commands/Transport/ListTransportsCommand.cs b/src/BuslyCLI.Console/Commands/Transport/ListTransportsCommand.cs
@@ -57,10 +57,14 @@ public string TransportConfigTypeToString(ITransportConfig transportConfig)
                 return "learning";
             case AzureServiceBusTransportConfig azureServiceBusConfig:
                 return "azure-service-bus";
+            case AzureStorageQueuesTransportConfig azureStorageQueuesTransportConfig:
+                return "azure-storage-queues";
             case AmazonsqsTransportConfig amazonsqsTransportConfig:
                 return "amazon-sqs";
             case SqlServerTransportConfig sqlServerTransportConfig:
                 return "sql-server";
+            case PostgreSqlTransportConfig postgreSqlTransportConfig:
+                return "postgre-sql";
             default:
                 throw new ApplicationException("Unknown transport type");
         }
diff --git a/src/BuslyCLI.Console/Config/PostgreSqlTransportConfig.cs b/src/BuslyCLI.Console/Config/PostgreSqlTransportConfig.cs
@@ -0,0 +1,6 @@
+﻿namespace BuslyCLI.Config;
+
+public class PostgreSqlTransportConfig : ITransportConfig
+{
+    public string ConnectionString { get; set; }
+}
diff --git a/src/BuslyCLI.Console/Config/TransportConfig.cs b/src/BuslyCLI.Console/Config/TransportConfig.cs
@@ -11,6 +11,7 @@ public class TransportConfig
     public AzureServiceBusTransportConfig AzureServiceBusTransportConfig { get; set; }
     public AzureStorageQueuesTransportConfig AzureStorageQueuesTransportConfig { get; set; }
     public SqlServerTransportConfig SqlServerTransportConfig { get; set; }
+    public PostgreSqlTransportConfig PostgreSqlTransportConfig { get; set; }
 
     // Helper property to unify config access:
     [YamlIgnore]
@@ -19,5 +20,6 @@ public class TransportConfig
                                       ?? (ITransportConfig)AmazonsqsTransportConfig
                                       ?? (ITransportConfig)AzureServiceBusTransportConfig
                                       ?? (ITransportConfig)AzureStorageQueuesTransportConfig
-                                      ?? SqlServerTransportConfig;
+                                      ?? (ITransportConfig)SqlServerTransportConfig
+                                      ?? PostgreSqlTransportConfig;
 }
diff --git a/src/BuslyCLI.Console/Config/Validators/PostgreSqlTransportConfigValidator.cs b/src/BuslyCLI.Console/Config/Validators/PostgreSqlTransportConfigValidator.cs
@@ -0,0 +1,12 @@
+﻿using FluentValidation;
+
+namespace BuslyCLI.Config.Validators;
+
+public class PostgreSqlTransportConfigValidator : AbstractValidator<PostgreSqlTransportConfig>
+{
+    public PostgreSqlTransportConfigValidator()
+    {
+        RuleFor(x => x.ConnectionString)
+            .NotEmpty();
+    }
+}
diff --git a/src/BuslyCLI.Console/Config/Validators/TransportConfigValidator.cs b/src/BuslyCLI.Console/Config/Validators/TransportConfigValidator.cs
@@ -16,7 +16,10 @@ public TransportConfigValidator()
                 v.Add(new LearningTransportConfigValidator());
                 v.Add(new RabbitMQTransportConfigValidator());
                 v.Add(new AzureServiceBusTransportConfigValidator());
+                v.Add(new AzureStorageQueuesTransportConfigValidator());
                 v.Add(new AmazonsqsTransportConfigValidator());
+                v.Add(new SqlServerTransportConfigValidator());
+                v.Add(new PostgreSqlTransportConfigValidator());
             });
 
         // RuleFor(x => x.LearningTransportConfig)
diff --git a/src/BuslyCLI.Console/DependencyInjection/ServiceCollectionExtensions.cs b/src/BuslyCLI.Console/DependencyInjection/ServiceCollectionExtensions.cs
@@ -36,7 +36,8 @@ private static IServiceCollection AddYamlDeserializer(this IServiceCollection se
                         { "amazonsqs-transport-config", typeof(AmazonsqsTransportConfig) },
                         { "azure-service-bus-transport-config", typeof(AzureServiceBusTransportConfig) },
                         { "azure-storage-queues-transport-config", typeof(AzureStorageQueuesTransportConfig) },
-                        { "sql-server-transport-config", typeof(SqlServerTransportConfig) }
+                        { "sql-server-transport-config", typeof(SqlServerTransportConfig) },
+                        { "postgre-sql-transport-config", typeof(PostgreSqlTransportConfig) }
                     };
 
                     o.AddUniqueKeyTypeDiscriminator<ITransportConfig>(keyMappings);
diff --git a/src/BuslyCLI.Console/Factories/RawEndpointFactory.cs b/src/BuslyCLI.Console/Factories/RawEndpointFactory.cs
@@ -36,6 +36,8 @@ private TransportDefinition CreateTransport(TransportConfig transportConfig)
                 return CreateAmazonSQSTransport(amazonSqsTransportConfig);
             case SqlServerTransportConfig sqlServerTransportConfig:
                 return CreateSqlServerTransport(sqlServerTransportConfig);
+            case PostgreSqlTransportConfig postgreSqlTransportConfig:
+                return CreatePostgreSqlTransport(postgreSqlTransportConfig);
             case LearningTransportConfig learningTransportConfig:
                 return new LearningTransport
                 {
@@ -47,6 +49,12 @@ private TransportDefinition CreateTransport(TransportConfig transportConfig)
         }
     }
 
+    private TransportDefinition CreatePostgreSqlTransport(PostgreSqlTransportConfig postgreSqlTransportConfig)
+    {
+        var transport = new PostgreSqlTransport(postgreSqlTransportConfig.ConnectionString);
+        return transport;
+    }
+
     private TransportDefinition CreateAzureStorageQueuesTransport(string connectionString)
     {
         var transport = new AzureStorageQueueTransport(connectionString);
diff --git a/tests/BuslyCLI.Console.Tests/BuslyCLI.Console.Tests.csproj b/tests/BuslyCLI.Console.Tests/BuslyCLI.Console.Tests.csproj
@@ -27,6 +27,7 @@
     <PackageReference Include="Testcontainers.Azurite" Version="4.9.0" />
     <PackageReference Include="Testcontainers.LocalStack" Version="4.9.0" />
     <PackageReference Include="Testcontainers.MsSql" Version="4.9.0" />
+    <PackageReference Include="Testcontainers.PostgreSql" Version="4.9.0" />
     <PackageReference Include="Testcontainers.RabbitMq" Version="4.9.0" />
     <PackageReference Include="Testcontainers.ServiceBus" Version="4.9.0" />
     <PackageReference Include="YamlDotNet" Version="16.3.0" />
diff --git a/tests/BuslyCLI.Console.Tests/Config/Validators/PostgreSqlTransportConfigValidatorTests.cs b/tests/BuslyCLI.Console.Tests/Config/Validators/PostgreSqlTransportConfigValidatorTests.cs
@@ -0,0 +1,47 @@
+﻿using BuslyCLI.Config;
+using BuslyCLI.Config.Validators;
+using FluentValidation.TestHelper;
+
+namespace BuslyCLI.Console.Tests.Config.Validators;
+
+[TestFixture]
+public class PostgreSqlTransportConfigValidatorTests
+{
+    private readonly PostgreSqlTransportConfigValidator _validator;
+
+    public PostgreSqlTransportConfigValidatorTests()
+    {
+        _validator = new PostgreSqlTransportConfigValidator();
+    }
+
+    [Test]
+    public async Task ShouldErrorWhenConnectionStringIsNotPassed()
+    {
+        // Arrange
+        var postgreSqlTransportConfig = new PostgreSqlTransportConfig
+        {
+            ConnectionString = null
+        };
+        // Act
+        var result = await _validator.TestValidateAsync(postgreSqlTransportConfig);
+
+        // Assert
+        result.ShouldHaveValidationErrorFor(c => c.ConnectionString)
+            .WithErrorMessage("'Connection String' must not be empty.");
+    }
+
+    [Test]
+    public async Task ShouldNotErrorConnectionStringIsPassed()
+    {
+        // Arrange
+        var postgreSqlTransportConfig = new PostgreSqlTransportConfig
+        {
+            ConnectionString = "Data Source=(local);Initial Catalog=Ordering;Integrated Security=SSPI;Application Name=Busly-CLI;TrustServerCertificate=true"
+        };
+        // Act
+        var result = await _validator.TestValidateAsync(postgreSqlTransportConfig);
+
+        // Assert
+        result.ShouldNotHaveValidationErrorFor(c => c.ConnectionString);
+    }
+}
diff --git a/tests/BuslyCLI.Console.Tests/EndToEnd/Infrastructure/ITestEndpointFactory.cs b/tests/BuslyCLI.Console.Tests/EndToEnd/Infrastructure/ITestEndpointFactory.cs
@@ -1,7 +1,6 @@
 ﻿using Amazon.Runtime;
 using Amazon.SimpleNotificationService;
 using Amazon.SQS;
-using NServiceBus.Settings;
 using NServiceBus.Transport;
 
 namespace BuslyCLI.Console.Tests.EndToEnd.Infrastructure;
@@ -101,6 +100,13 @@ public async Task<TestEndpoint> CreateSqlServerTestEndpoint(string sqlConnection
         return await InternalCreateTestEndpoint(name, transport);
     }
 
+    public async Task<TestEndpoint> CreatePostgreSqlTransport(string connectionString)
+    {
+        var name = GenerateUniqueEndpointName();
+        var transport = new PostgreSqlTransport(connectionString);
+        return await InternalCreateTestEndpoint(name, transport);
+    }
+
     public async Task<TestEndpoint> CreateAzureStorageQueuesTestEndpoint(string connectionString)
     {
         var name = GenerateUniqueEndpointName();
diff --git a/tests/BuslyCLI.Console.Tests/EndToEnd/PostgreSql/PostgreSqlEndToEndTestBase.cs b/tests/BuslyCLI.Console.Tests/EndToEnd/PostgreSql/PostgreSqlEndToEndTestBase.cs
@@ -0,0 +1,20 @@
+﻿using Testcontainers.PostgreSql;
+
+namespace BuslyCLI.Console.Tests.EndToEnd.PostgreSql;
+
+[TestFixture]
+public abstract class PostgreSqlEndToEndTestBase : SingletonTestFixtureBase<PostgreSqlContainer>
+{
+    protected PostgreSqlContainer PostgreSqlContainer => Container;
+
+    protected override PostgreSqlContainer CreateContainer()
+    {
+        return new PostgreSqlBuilder()
+            .Build();
+    }
+
+    protected override async Task StartContainerAsync(PostgreSqlContainer container)
+    {
+        await container.StartAsync();
+    }
+}
diff --git a/tests/BuslyCLI.Console.Tests/EndToEnd/PostgreSql/SendCommandPostgreSqlEndToEndTests.cs b/tests/BuslyCLI.Console.Tests/EndToEnd/PostgreSql/SendCommandPostgreSqlEndToEndTests.cs
@@ -0,0 +1,120 @@
+﻿using System.Text;
+using System.Text.Json;
+using BuslyCLI.Console.Tests.EndToEnd.Infrastructure;
+using BuslyCLI.Console.Tests.TestHelpers;
+using BuslyCLI.DependencyInjection;
+using BuslyCLI.Spectre;
+using Microsoft.Extensions.DependencyInjection;
+using Spectre.Console.Cli.Extensions.DependencyInjection;
+using Spectre.Console.Cli.Testing;
+
+namespace BuslyCLI.Console.Tests.EndToEnd.PostgreSql;
+
+
+[TestFixture]
+public class SendCommandPostgreSqlEndToEndTests : PostgreSqlEndToEndTestBase
+{
+    [SetUp]
+    public void Setup()
+    {
+        var registrations = new ServiceCollection();
+        registrations.AddBuslyCLIServices();
+        using var registrar = new DependencyInjectionRegistrar(registrations);
+        _sut = new CommandAppTester(registrar);
+        _sut.Configure(AppConfiguration.GetSpectreCommandConfiguration());
+    }
+
+    private CommandAppTester _sut;
+
+    private readonly JsonSerializerOptions _jsonObjectOptions =
+        new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase, WriteIndented = true };
+
+    [Test]
+    public async Task ShouldSendCommand()
+    {
+        await RunWithTestEndpoint(async testEndpoint =>
+        {
+            // Arrange
+            await testEndpoint.StartEndpoint();
+            var messageBody = new { OrderNumber = Guid.NewGuid() };
+            var json = JsonSerializer.Serialize(messageBody, _jsonObjectOptions);
+            var yamlFile = $"""
+                            ---
+                            current-transport: local-postgre-sql
+                            transports:
+                              - name: local-postgre-sql
+                                postgre-sql-transport-config:
+                                  connection-string: {Container.GetConnectionString()}
+                            """;
+            using var configFile = new TestableNServiceBusConfigurationFile(yamlFile);
+
+            // Act
+            var result = _sut.Run(
+                "command",
+                "send",
+                "--content-type", "application/json",
+                "--enclosed-message-type", "MessageContracts.Commands.CreateOrder",
+                "--destination-endpoint", testEndpoint.EndpointName,
+                "--message-body", json,
+                "--config", configFile.FilePath);
+
+            // Assert
+            Assert.That(result.ExitCode, Is.EqualTo(0));
+            var message = testEndpoint.TryReceiveMessage();
+            Assert.That(message.Headers["NServiceBus.EnclosedMessageTypes"],
+                Is.EqualTo("MessageContracts.Commands.CreateOrder"));
+            Assert.That(message.Headers["NServiceBus.ContentType"], Is.EqualTo("application/json"));
+            Assert.That(Encoding.UTF8.GetString(message.Body.Span), Is.EqualTo(json));
+        });
+    }
+
+    [Test]
+    public async Task ShouldPublishEvent()
+    {
+        await RunWithTestEndpoint(async testEndpoint =>
+        {
+            // Arrange
+            await testEndpoint.StartEndpoint();
+            await testEndpoint.Subscribe("MessageContracts.Events.OrderCreated");
+            var messageBody = new { OrderNumber = Guid.NewGuid() };
+            var json = JsonSerializer.Serialize(messageBody, _jsonObjectOptions);
+            var yamlFile = $"""
+                            ---
+                            current-transport: local-postgre-sql
+                            transports:
+                              - name: local-postgre-sql
+                                postgre-sql-transport-config:
+                                  connection-string: {Container.GetConnectionString()}
+                            """;
+            using var configFile = new TestableNServiceBusConfigurationFile(yamlFile);
+
+            // Act
+            var result = _sut.Run(
+                "event",
+                "publish",
+                "--content-type", "application/json",
+                "--enclosed-message-type", "MessageContracts.Events.OrderCreated",
+                "--message-body", json,
+                "--config", configFile.FilePath);
+
+            // Assert
+            Assert.That(result.ExitCode, Is.EqualTo(0));
+            var message = testEndpoint.TryReceiveMessage();
+            Assert.That(message.Headers["NServiceBus.EnclosedMessageTypes"],
+                Is.EqualTo("MessageContracts.Events.OrderCreated"));
+            Assert.That(message.Headers["NServiceBus.ContentType"], Is.EqualTo("application/json"));
+            Assert.That(Encoding.UTF8.GetString(message.Body.Span), Is.EqualTo(json));
+        });
+    }
+
+    // Test Endpoint
+    // Example of how to wait for and get messages
+    // https://github.com/Particular/NServiceBus.RabbitMQ/blob/dba627a5a2c50519d7a2466efe3f76c8d5c8828d/src/NServiceBus.Transport.RabbitMQ.Tests/RabbitMqContext.cs#L41
+    private async Task RunWithTestEndpoint(Func<TestEndpoint, Task> testAction)
+    {
+        var testEndpoint = await new TestEndpointFactory().CreatePostgreSqlTransport(Container.GetConnectionString());
+
+        await testAction(testEndpoint);
+        await testEndpoint.ShutDownAndCleanUp();
+    }
+}
diff --git a/website/docs/transports/postgre-sql.md b/website/docs/transports/postgre-sql.md
@@ -0,0 +1,40 @@
+# Postgre Sql
+
+The **Postgre Sql Transport** is used to communicate to Postgre Sql. It is suitable for development, testing, and production environments.
+
+## Configuration
+
+To use the Postgre Sql Transport, define it under `transports` and reference it as `current-transport`.
+
+### Example
+
+```yaml
+current-transport: local-postgre-sql
+
+transports:
+  - name: local-postgre-sql
+    postgre-sql-transport-config:
+      connection-string: Data Source=(local);Initial Catalog=Ordering;Integrated Security=SSPI;Application Name=Busly-CLI;TrustServerCertificate=true
+```
+
+---
+
+## `postgre-sql-transport-config` Fields
+
+| Field               | Required | Type   | Default | Description                         |
+| ------------------- | -------- | ------ | ------- | ----------------------------------- |
+| `connection-string` | **Yes**  | string | —       | Full Postgre Sql Connection string. |
+
+---
+
+## Field Details
+
+### `connection-string` (required)
+
+Postgre Sql connection string used to connect to the database.
+
+Examples:
+
+```yaml
+connection-string: Data Source=(local);Initial Catalog=Ordering;Integrated Security=SSPI;Application Name=Busly-CLI;TrustServerCertificate=true
+```

Original file line number	Diff line number	Diff line change
`@@ -36,6 +36,8 @@ private TransportDefinition CreateTransport(TransportConfig transportConfig)`
`36`	`36`	`return CreateAmazonSQSTransport(amazonSqsTransportConfig);`
`37`	`37`	`case SqlServerTransportConfig sqlServerTransportConfig:`
`38`	`38`	`return CreateSqlServerTransport(sqlServerTransportConfig);`
	`39`	`+ case PostgreSqlTransportConfig postgreSqlTransportConfig:`
	`40`	`+ return CreatePostgreSqlTransport(postgreSqlTransportConfig);`
`39`	`41`	`case LearningTransportConfig learningTransportConfig:`
`40`	`42`	`return new LearningTransport`
`41`	`43`	`{`
`@@ -47,6 +49,12 @@ private TransportDefinition CreateTransport(TransportConfig transportConfig)`
`47`	`49`	`}`
`48`	`50`	`}`
`49`	`51`
	`52`	`+ private TransportDefinition CreatePostgreSqlTransport(PostgreSqlTransportConfig postgreSqlTransportConfig)`
	`53`	`+ {`
	`54`	`+ var transport = new PostgreSqlTransport(postgreSqlTransportConfig.ConnectionString);`
	`55`	`+ return transport;`
	`56`	`+ }`
	`57`	`+`
`50`	`58`	`private TransportDefinition CreateAzureStorageQueuesTransport(string connectionString)`
`51`	`59`	`{`
`52`	`60`	`var transport = new AzureStorageQueueTransport(connectionString);`