Merge pull request #140 from TraGicCode/chore/add-routing-topology-and-queue-type-for-rabbit-transport

Add Routing Topology and QueueType to RabbitMQ Transport

Author: TraGicCode

src/BuslyCLI.Console/Config/RabbitmqTransportConfig.cs

@@ -2,15 +2,29 @@
 
 public class RabbitmqTransportConfig : ITransportConfig
 {
+    // TODO: Add Support for TLS Client Certificate Authentication
+    // https://github.com/Particular/NServiceBus.RabbitMQ/blob/master/src/NServiceBus.Transport.RabbitMQ/Connection/ConnectionFactory.cs#L69
     // TODO: Test TLS Connections to broker using "amqps://your-username:[email protected]:5671/vhost"
     public string AmqpConnectionString { get; set; }
 
-    // TODO: Add Support for TLS Client Certificate Authentication
-    // https://github.com/Particular/NServiceBus.RabbitMQ/blob/master/src/NServiceBus.Transport.RabbitMQ/Connection/ConnectionFactory.cs#L69
+    public RabbitmqRoutingTopology RoutingTopology { get; set; } = RabbitmqRoutingTopology.Conventional;
+    public RabbitmqQueueType QueueType { get; set; } = RabbitmqQueueType.Quorum;
 
     public ManagementApi ManagementApi { get; set; }
 }
 
+public enum RabbitmqRoutingTopology
+{
+    Conventional,
+    Direct,
+}
+
+public enum RabbitmqQueueType
+{
+    Quorum,
+    Classic,
+}
+
 public class ManagementApi
 {
     public string Url { get; set; }

src/BuslyCLI.Console/Config/Validators/RabbitMQTransportConfigValidator.cs

@@ -9,6 +9,9 @@ public RabbitMQTransportConfigValidator()
         RuleFor(x => x.AmqpConnectionString)
             .NotEmpty();
 
+        RuleFor(x => x.RoutingTopology)
+            .IsInEnum();
+
         RuleFor(x => x.ManagementApi)
             .SetValidator(new ManagementApiConfigValidator());
     }

src/BuslyCLI.Console/Factories/RawEndpointFactory.cs

@@ -69,12 +69,17 @@ private TransportDefinition CreateSqlServerTransport(SqlServerTransportConfig sq
 
     private RabbitMQTransport CreateRabbitMQTransport(RabbitmqTransportConfig rabbitmqTransportConfig)
     {
-        var t = new RabbitMQTransport(RoutingTopology.Conventional(QueueType.Quorum), rabbitmqTransportConfig.AmqpConnectionString);
+        var routingTopology = rabbitmqTransportConfig.RoutingTopology switch
+        {
+            RabbitmqRoutingTopology.Conventional => RoutingTopology.Conventional(rabbitmqTransportConfig.QueueType == RabbitmqQueueType.Quorum ? QueueType.Quorum : QueueType.Classic),
+            RabbitmqRoutingTopology.Direct => RoutingTopology.Direct(rabbitmqTransportConfig.QueueType == RabbitmqQueueType.Quorum ? QueueType.Quorum : QueueType.Classic),
+            _ => throw new Exception("Unknown RabbitMQ routing topology: " + rabbitmqTransportConfig.RoutingTopology)
+        };
+        var t = new RabbitMQTransport(routingTopology, rabbitmqTransportConfig.AmqpConnectionString);
 
         if (rabbitmqTransportConfig.ManagementApi != null)
         {
-            t.ManagementApiConfiguration =
-                CreateManagementApiConfig(rabbitmqTransportConfig.ManagementApi);
+            t.ManagementApiConfiguration = CreateManagementApiConfig(rabbitmqTransportConfig.ManagementApi);
         }
         return t;
     }

src/BuslyCLI.Console/Spectre/AppConfiguration.cs

@@ -3,7 +3,9 @@
 using BuslyCLI.Commands.Demo;
 using BuslyCLI.Commands.Event;
 using BuslyCLI.Commands.Transport;
+using Spectre.Console;
 using Spectre.Console.Cli;
+using YamlDotNet.Core;
 
 namespace BuslyCLI.Spectre;
 
@@ -53,9 +55,24 @@ public static Action<IConfigurator> GetSpectreCommandConfiguration()
                 demo.AddCommand<StartDemoCommand>("start")
                     .WithDescription("Start a demo endpoint that can receive any command and a single 'Messages.Events.OrderPlaced' event.");
             });
-#if DEBUG
-            config.PropagateExceptions();
-#endif
+
+            config.SetExceptionHandler((ex, _) =>
+            {
+                // if (ex.InnerException is OptionsValidationException)
+                // {
+                //     AnsiConsole.Write(new Markup($"{ConsoleExtensions.ErrorMarkup}{ex.InnerException.Message}"));
+                //     return;
+                // }
+                // if (ex is CommandAppException)
+                // {
+                //     AnsiConsole.Write(new Markup($"{ConsoleExtensions.ErrorMarkup}{ex.Message}"));
+                //     return;
+                // }
+                // AnsiConsole.WriteException(ex, ExceptionFormats.ShortenPaths);
+                AnsiConsole.WriteException(ex, ExceptionFormats.ShortenEverything);
+            });
+
+
         };
     }
 }

website/docs/transports/rabbitmq.md

@@ -22,10 +22,12 @@ transports:
 
 ## `rabbitmq-transport-config` Fields
 
-| Field                    | Required | Type   | Default | Description                                                       |
-| ------------------------ | -------- | ------ | ------- | ----------------------------------------------------------------- |
-| `amqp-connection-string` | **Yes**  | string | —       | Full AMQP connection string used to connect to RabbitMQ.          |
-| `management-api`         | No       | object | —       | Optional configuration to connect to the RabbitMQ Management API. |
+| Field                    | Required | Type   | Default      | Description                                                       |
+| ------------------------ | -------- | ------ | ------------ | ----------------------------------------------------------------- |
+| `amqp-connection-string` | **Yes**  | string | —            | Full AMQP connection string used to connect to RabbitMQ.          |
+| `routing-toplogy`        | No       | string | conventional | Routing toplogy to be used (Conventional or Direct).              |
+| `queue-type`             | No       | string | quorum       | Type of queues RabbitMQ is using (Quorum or Classic).             |
+| `management-api`         | No       | object | —            | Optional configuration to connect to the RabbitMQ Management API. |
 
 ---
 
@@ -57,6 +59,38 @@ amqp-connection-string: amqps://user:[email protected]:5671/my-vhost
 
 ---
 
+### `routing-topology` (optional)
+
+The Routing topology to be used for the transport
+
+Examples:
+
+```yaml
+routing-toplogy: conventional
+```
+
+```yaml
+routing-toplogy: direct
+```
+
+---
+
+### `queue-type` (optional)
+
+The type of queues being used.
+
+Examples:
+
+```yaml
+queue-type: quorum
+```
+
+```yaml
+queue-type: classic
+```
+
+---
+
 ### `management-api` (optional)
 
 Allows Busly to interact with the RabbitMQ Management API for monitoring or queue management.