Add basic functions from Rabbit.Client: ExchangeDeclare, QueueBind.

ifgris · Sep 9, 2024 · d745d64 · d745d64
1 parent 5ba118b
commit d745d64
Show file tree

Hide file tree

Showing 2 changed files with 48 additions and 6 deletions.
diff --git a/Src/NanoRabbit/IRabbitHelper.cs b/Src/NanoRabbit/IRabbitHelper.cs
@@ -1,4 +1,5 @@
 using RabbitMQ.Client;
+using System.Threading.Channels;
 
 namespace NanoRabbit;
 
@@ -44,6 +45,15 @@ public interface IRabbitHelper
 
     #region utils
 
+    /// <summary>
+    /// Declare an exchange.
+    /// </summary>
+    /// <param name="exchangeName"></param>
+    /// <param name="exchangeType"></param>
+    /// <param name="durable"></param>
+    /// <param name="autoDelete"></param>
+    /// <param name="arguments"></param>
+    public void ExchangeDeclare(string exchangeName, string exchangeType, bool durable = false, bool autoDelete = false, IDictionary<string, object>? arguments = null);
     /// <summary>
     /// Declare a queue.
     /// </summary>
@@ -54,6 +64,14 @@ public interface IRabbitHelper
     /// <param name="arguments"></param>
     public void DeclareQueue(string queueName, bool durable = true, bool exclusive = false, bool autoDelete = false, IDictionary<string, object>? arguments = null);
     /// <summary>
+    /// Bind a queue to an exchange.
+    /// </summary>
+    /// <param name="queueName"></param>
+    /// <param name="exchangeName"></param>
+    /// <param name="routingKey"></param>
+    /// <param name="arguments"></param>
+    public void QueueBind(string queueName, string exchangeName, string routingKey, IDictionary<string, object> arguments);
+    /// <summary>
     /// Create a custom BasicProperties.
     /// </summary>
     /// <returns></returns>

diff --git a/Src/NanoRabbit/RabbitHelper.cs b/Src/NanoRabbit/RabbitHelper.cs
@@ -48,7 +48,7 @@ public RabbitHelper(RabbitConfiguration rabbitConfig, ILogger logger)
                 };
 
                 // TODO needs testing.
-                if (_rabbitConfig.TLSConfig != null )
+                if (_rabbitConfig.TLSConfig != null)
                 {
                     factory.Ssl.Enabled = _rabbitConfig.TLSConfig.Enabled;
                     factory.Ssl.ServerName = _rabbitConfig.TLSConfig.ServerName;
@@ -124,7 +124,7 @@ public void Publish<T>(string producerName, T message, IBasicProperties? propert
             var option = GetProducerOption(producerName);
 
             var messageStr = typeof(T) == typeof(string) ? message.ToString() : JsonConvert.SerializeObject(message);
-            
+
             var body = Encoding.UTF8.GetBytes(messageStr);
 
             _pipeline.Execute(token =>
@@ -134,9 +134,9 @@ public void Publish<T>(string producerName, T message, IBasicProperties? propert
                     properties ??= _channel.CreateBasicProperties();
                     properties.Persistent = true;
                     _channel.BasicPublish(
-                        exchange: option.ExchangeName, 
-                        routingKey: option.RoutingKey, 
-                        basicProperties: properties, 
+                        exchange: option.ExchangeName,
+                        routingKey: option.RoutingKey,
+                        basicProperties: properties,
                         body: body);
 
                     _logger?.LogInformation($"{producerName}|Published|{messageStr}");
@@ -264,6 +264,19 @@ public void AddAsyncConsumer(string consumerName, Func<string, Task> onMessageRe
 
         #region utils
 
+        /// <summary>
+        /// Declare an exchange.
+        /// </summary>
+        /// <param name="exchangeName"></param>
+        /// <param name="exchangeType"></param>
+        /// <param name="durable"></param>
+        /// <param name="autoDelete"></param>
+        /// <param name="arguments"></param>
+        public void ExchangeDeclare(string exchangeName, string exchangeType, bool durable = false, bool autoDelete = false, IDictionary<string, object>? arguments = null)
+        {
+            _channel.ExchangeDeclare(exchangeName, exchangeType, durable, autoDelete, arguments);
+        }
+
         /// <summary>
         /// Declare a queue based on RabbitMQ.Client.
         /// </summary>
@@ -275,7 +288,18 @@ public void AddAsyncConsumer(string consumerName, Func<string, Task> onMessageRe
         public void DeclareQueue(string queueName, bool durable = true, bool exclusive = false, bool autoDelete = false, IDictionary<string, object>? arguments = null)
         {
             _channel.QueueDeclare(queue: queueName, durable, exclusive, autoDelete, arguments);
-            _channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
+        }
+
+        /// <summary>
+        /// Bind a queue to an exchange.
+        /// </summary>
+        /// <param name="queueName"></param>
+        /// <param name="exchangeName"></param>
+        /// <param name="routingKey"></param>
+        /// <param name="arguments"></param>
+        public void QueueBind(string queueName, string exchangeName, string routingKey, IDictionary<string, object> arguments)
+        {
+            _channel.QueueBind(queueName, exchangeName, routingKey, arguments);
         }
 
         /// <summary>