diff --git a/Directory.Packages.props b/Directory.Packages.props index ab956547..890910fc 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -20,7 +20,7 @@ - + diff --git a/Src/CrispyWaffle.RabbitMQ/Helpers/MessageReceiver.cs b/Src/CrispyWaffle.RabbitMQ/Helpers/MessageReceiver.cs index 955c7a6a..7edfddcb 100644 --- a/Src/CrispyWaffle.RabbitMQ/Helpers/MessageReceiver.cs +++ b/Src/CrispyWaffle.RabbitMQ/Helpers/MessageReceiver.cs @@ -1,143 +1,171 @@ -using System; -using System.Text; -using System.Threading; -using System.Threading.Tasks; -using CrispyWaffle.Log; -using CrispyWaffle.RabbitMQ.Utils.Communications; -using RabbitMQ.Client; -using RabbitMQ.Client.Events; - -namespace CrispyWaffle.RabbitMQ.Helpers; - -/// -/// A class that facilitates receiving messages from RabbitMQ queues or exchanges. -/// -/// -/// The class provides methods to receive messages from RabbitMQ queues or exchanges, -/// handling the message reception process and invoking events when messages are received. -/// -public class MessageReceiver -{ - /// - /// The RabbitMQ connection connector. - /// - private readonly RabbitMQConnector _connector; - - /// - /// Initializes a new instance of the class. - /// - /// The to be used for connecting to RabbitMQ. - /// Thrown if the is null. - public MessageReceiver(RabbitMQConnector connector) => - _connector = connector ?? throw new ArgumentNullException(nameof(connector)); - - /// - /// Delegate for handling received messages. - /// - /// The sender of the message. - /// The containing details of the received message. - public delegate void MessageReceivedHandler(object sender, MessageReceivedArgs e); - - /// - /// Event triggered when a message is received from the queue or exchange. - /// - public event MessageReceivedHandler MessageReceived; - - /// - /// Starts receiving messages from a RabbitMQ queue. - /// - /// The type of message to receive, which should implement . - /// If set to true, messages will be acknowledged automatically. - /// A to allow cancellation of the operation. - /// - /// This method will begin a background task to receive messages from the specified queue. - /// - public void ReceiveFromQueue(bool autoAck, CancellationToken cancellationToken) - where T : class, IQueuing, new() - { - var queueName = Extensions.GetQueueName(); - - Task.Run( - () => DoWork(string.Empty, queueName, autoAck, cancellationToken), - cancellationToken - ) - .ConfigureAwait(false); - } - - /// - /// Starts receiving messages from a RabbitMQ exchange. - /// - /// The type of message to receive, which should implement . - /// If set to true, messages will be acknowledged automatically. - /// A to allow cancellation of the operation. - /// - /// This method will begin a background task to receive messages from the specified exchange. - /// - public void ReceiveFromExchange(bool autoAck, CancellationToken cancellationToken) - where T : class, IQueuing, new() - { - var exchangeName = Extensions.GetExchangeName(); - - Task.Run( - () => DoWork(exchangeName, string.Empty, autoAck, cancellationToken), - cancellationToken - ) - .ConfigureAwait(false); - } - - /// - /// Handles the work of receiving messages from a specified exchange or queue. - /// - /// The name of the exchange (optional). - /// The name of the queue (optional). - /// If set to true, messages will be acknowledged automatically. - /// A to allow cancellation of the operation. - /// - /// This method connects to RabbitMQ, binds the queue to the exchange (if specified), - /// and starts listening for messages. It invokes the event when a message is received. - /// - private void DoWork( - string exchange, - string queue, - bool autoAck, - CancellationToken cancellationToken - ) - { - using (var connection = _connector.ConnectionFactory.CreateConnection()) - using (var channel = connection.CreateModel()) - { - var consumer = new EventingBasicConsumer(channel); - - var queueName = string.IsNullOrWhiteSpace(queue) - ? channel.QueueDeclare().QueueName - : queue; - - if (!string.IsNullOrWhiteSpace(exchange)) - { - channel.QueueBind(queueName, exchange, string.Empty); - } - - consumer.Received += (_, args) => - { - if (MessageReceived == null) - { - return; - } - - LogConsumer.Trace( - $"Message received from exchange: {exchange} | queue: {queue} | queue name: {queueName}" - ); - - var body = Encoding.UTF8.GetString(args.Body.ToArray()); - - var eventArgs = new MessageReceivedArgs { QueueName = queueName, Body = body }; - - MessageReceived.Invoke(this, eventArgs); - }; - - channel.BasicConsume(queue: queueName, autoAck: autoAck, consumer: consumer); - - cancellationToken.WaitHandle.WaitOne(); - } - } -} +using System; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using CrispyWaffle.Log; +using CrispyWaffle.RabbitMQ.Utils.Communications; +using RabbitMQ.Client; +using RabbitMQ.Client.Events; + +namespace CrispyWaffle.RabbitMQ.Helpers; + +/// +/// A class that facilitates receiving messages from RabbitMQ queues or exchanges. +/// +/// +/// The class provides methods to receive messages from RabbitMQ queues or exchanges, +/// handling the message reception process and invoking events when messages are received. +/// +public class MessageReceiver +{ + /// + /// The RabbitMQ connection connector. + /// + private readonly RabbitMQConnector _connector; + + /// + /// Initializes a new instance of the class. + /// + /// The to be used for connecting to RabbitMQ. + /// Thrown if the is null. + public MessageReceiver(RabbitMQConnector connector) => + _connector = connector ?? throw new ArgumentNullException(nameof(connector)); + + /// + /// Delegate for handling received messages. + /// + /// The sender of the message. + /// The containing details of the received message. + public delegate void MessageReceivedHandler(object sender, MessageReceivedArgs e); + + /// + /// Event triggered when a message is received from the queue or exchange. + /// + public event MessageReceivedHandler MessageReceived; + + /// + /// Starts receiving messages from a RabbitMQ queue. + /// + /// The type of message to receive, which should implement . + /// If set to true, messages will be acknowledged automatically. + /// A to allow cancellation of the operation. + /// + /// This method will begin a background task to receive messages from the specified queue. + /// + public void ReceiveFromQueue(bool autoAck, CancellationToken cancellationToken) + where T : class, IQueuing, new() + { + var queueName = Extensions.GetQueueName(); + + Task.Run( + () => DoWorkAsync(string.Empty, queueName, autoAck, cancellationToken), + cancellationToken + ) + .ConfigureAwait(false); + } + + /// + /// Starts receiving messages from a RabbitMQ exchange. + /// + /// The type of message to receive, which should implement . + /// If set to true, messages will be acknowledged automatically. + /// A to allow cancellation of the operation. + /// + /// This method will begin a background task to receive messages from the specified exchange. + /// + public void ReceiveFromExchange(bool autoAck, CancellationToken cancellationToken) + where T : class, IQueuing, new() + { + var exchangeName = Extensions.GetExchangeName(); + + Task.Run( + () => DoWorkAsync(exchangeName, string.Empty, autoAck, cancellationToken), + cancellationToken + ) + .ConfigureAwait(false); + } + + /// + /// Handles the work of receiving messages from a specified exchange or queue. + /// + /// The name of the exchange (optional). + /// The name of the queue (optional). + /// If set to true, messages will be acknowledged automatically. + /// A to allow cancellation of the operation. + /// + /// This method connects to RabbitMQ, binds the queue to the exchange (if specified), + /// and starts listening for messages. It invokes the event when a message is received. + /// + private async Task DoWorkAsync( + string exchange, + string queue, + bool autoAck, + CancellationToken cancellationToken + ) + { + var connection = await _connector + .ConnectionFactory.CreateConnectionAsync(cancellationToken) + .ConfigureAwait(false); + await using var connectionDisposable = connection.ConfigureAwait(false); + + var channel = await connection + .CreateChannelAsync(cancellationToken: cancellationToken) + .ConfigureAwait(false); + await using var channelDisposable = channel.ConfigureAwait(false); + + var consumer = new AsyncEventingBasicConsumer(channel); + + var queueName = string.IsNullOrWhiteSpace(queue) + ? ( + await channel + .QueueDeclareAsync(cancellationToken: cancellationToken) + .ConfigureAwait(false) + ).QueueName + : queue; + + if (!string.IsNullOrWhiteSpace(exchange)) + { + await channel + .QueueBindAsync( + queueName, + exchange, + string.Empty, + cancellationToken: cancellationToken + ) + .ConfigureAwait(false); + } + + consumer.ReceivedAsync += (_, args) => + { + if (MessageReceived == null) + { + return Task.CompletedTask; + } + + LogConsumer.Trace( + $"Message received from exchange: {exchange} | queue: {queue} | queue name: {queueName}" + ); + + var body = Encoding.UTF8.GetString(args.Body.ToArray()); + + var eventArgs = new MessageReceivedArgs { QueueName = queueName, Body = body }; + + MessageReceived.Invoke(this, eventArgs); + + return Task.CompletedTask; + }; + + await channel + .BasicConsumeAsync(queueName, autoAck, consumer, cancellationToken) + .ConfigureAwait(false); + + try + { + await Task.Delay(Timeout.Infinite, cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // Expected when cancellation is requested; allows graceful shutdown. + } + } +} diff --git a/Src/CrispyWaffle.RabbitMQ/Log/RabbitMQLogProvider.cs b/Src/CrispyWaffle.RabbitMQ/Log/RabbitMQLogProvider.cs index 35d1016d..e9e375c1 100644 --- a/Src/CrispyWaffle.RabbitMQ/Log/RabbitMQLogProvider.cs +++ b/Src/CrispyWaffle.RabbitMQ/Log/RabbitMQLogProvider.cs @@ -34,7 +34,7 @@ public class RabbitMQLogProvider : ILogProvider, IDisposable /// /// The channel. /// - private readonly IModel _channel; + private readonly IChannel _channel; /// /// The cancellation token. @@ -55,8 +55,24 @@ public class RabbitMQLogProvider : ILogProvider, IDisposable public RabbitMQLogProvider(RabbitMQConnector connector, CancellationToken cancellationToken) { _connector = connector ?? throw new ArgumentNullException(nameof(connector)); - _channel = _connector.ConnectionFactory.CreateConnection().CreateModel(); - _channel.ExchangeDeclare(_connector.DefaultExchangeName, ExchangeType.Fanout, true); + + var connection = _connector + .ConnectionFactory.CreateConnectionAsync(cancellationToken) + .GetAwaiter() + .GetResult(); + _channel = connection + .CreateChannelAsync(cancellationToken: cancellationToken) + .GetAwaiter() + .GetResult(); + _channel + .ExchangeDeclareAsync( + _connector.DefaultExchangeName, + ExchangeType.Fanout, + true, + cancellationToken: cancellationToken + ) + .GetAwaiter() + .GetResult(); _cancellationToken = cancellationToken; _queue = new ConcurrentQueue(); var thread = new Thread(Worker); @@ -107,7 +123,15 @@ private void PropagateMessageInternal(string message) try { var body = Encoding.UTF8.GetBytes(message); - _channel.BasicPublish(_connector.DefaultExchangeName, string.Empty, null, body); + _channel + .BasicPublishAsync( + _connector.DefaultExchangeName, + string.Empty, + (ReadOnlyMemory)body, + _cancellationToken + ) + .GetAwaiter() + .GetResult(); } catch (Exception e) { @@ -126,7 +150,8 @@ private void Dispose(bool disposing) return; } - _channel.Close(); + _channel.CloseAsync().GetAwaiter().GetResult(); + _channel.Dispose(); } /// diff --git a/Src/CrispyWaffle.RabbitMQ/Utils/Communications/RabbitMQWrapper.cs b/Src/CrispyWaffle.RabbitMQ/Utils/Communications/RabbitMQWrapper.cs index 2d6e0bf9..2e02d425 100644 --- a/Src/CrispyWaffle.RabbitMQ/Utils/Communications/RabbitMQWrapper.cs +++ b/Src/CrispyWaffle.RabbitMQ/Utils/Communications/RabbitMQWrapper.cs @@ -1,104 +1,154 @@ -using System; -using System.Text; -using CrispyWaffle.Log; -using CrispyWaffle.RabbitMQ.Helpers; -using CrispyWaffle.Serialization; - -namespace CrispyWaffle.RabbitMQ.Utils.Communications; - -/// -/// Provides methods for sending messages to RabbitMQ exchanges and queues. -/// The class uses a to establish -/// a connection with RabbitMQ and send serialized messages to a specified exchange or queue. -/// -public class RabbitMQWrapper -{ - /// - /// The RabbitMQ connector used to establish connections. - /// - private readonly RabbitMQConnector _connector; - - /// - /// Initializes a new instance of the class. - /// - /// The instance used to establish RabbitMQ connections. - /// Thrown when the is null. - public RabbitMQWrapper(RabbitMQConnector connector) => - _connector = connector ?? throw new ArgumentNullException(nameof(connector)); - - /// - /// Sends a serialized message to a RabbitMQ exchange. - /// - /// The type of the item being sent. It must implement and provide a serializer. - /// The item to send. It will be serialized to JSON before being sent. - /// The type of exchange to declare. If null, the exchange will not be declared. - /// - /// This method establishes a connection to RabbitMQ, declares an exchange if necessary, - /// serializes the provided item to JSON, and sends the message to the exchange. - /// - public void SendToExchange(T item, string exchangeDeclareType = null) - where T : class, IQueuing, new() - { - var exchangeName = Helpers.Extensions.GetExchangeName(); - - using (var connection = _connector.ConnectionFactory.CreateConnection()) - using (var channel = connection.CreateModel()) - { - // Declare the exchange if the type is provided - if (!string.IsNullOrWhiteSpace(exchangeDeclareType)) - { - channel.ExchangeDeclare(exchangeName, exchangeDeclareType, true, false, null); - } - - // Serialize the item and prepare the message - var json = (string)item.GetSerializer(); - var body = Encoding.UTF8.GetBytes(json); - - var properties = channel.CreateBasicProperties(); - properties.Persistent = true; - - // Publish the message to the exchange - channel.BasicPublish(exchangeName, string.Empty, false, properties, body); - - LogConsumer.Trace("Sent to exchange {0} the item {1}", exchangeName, json); - } - } - - /// - /// Sends a serialized message to a RabbitMQ queue. - /// - /// The type of the item being sent. It must implement and provide a serializer. - /// The item to send. It will be serialized to JSON before being sent. - /// If true, the queue will be declared before sending the message. Default is true. - /// - /// This method establishes a connection to RabbitMQ, declares a queue if necessary, - /// serializes the provided item to JSON, and sends the message to the queue. - /// - public void SendToQueue(T item, bool queueDeclare = true) - where T : class, IQueuing, new() - { - var queueName = Helpers.Extensions.GetQueueName(); - - using (var connection = _connector.ConnectionFactory.CreateConnection()) - using (var channel = connection.CreateModel()) - { - // Declare the queue if specified - if (queueDeclare) - { - channel.QueueDeclare(queueName, true, false, false, null); - } - - // Serialize the item and prepare the message - var json = (string)item.GetSerializer(); - var body = Encoding.UTF8.GetBytes(json); - - var properties = channel.CreateBasicProperties(); - properties.Persistent = true; - - // Publish the message to the queue - channel.BasicPublish(string.Empty, queueName, false, properties, body); - - LogConsumer.Trace("Sent to queue {0} the item: {1}", queueName, json); - } - } -} +using System; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using CrispyWaffle.Log; +using CrispyWaffle.RabbitMQ.Helpers; +using CrispyWaffle.Serialization; +using RabbitMQ.Client; + +namespace CrispyWaffle.RabbitMQ.Utils.Communications; + +/// +/// Provides methods for sending messages to RabbitMQ exchanges and queues. +/// The class uses a to establish +/// a connection with RabbitMQ and send serialized messages to a specified exchange or queue. +/// +public class RabbitMQWrapper +{ + /// + /// The RabbitMQ connector used to establish connections. + /// + private readonly RabbitMQConnector _connector; + + /// + /// Initializes a new instance of the class. + /// + /// The instance used to establish RabbitMQ connections. + /// Thrown when the is null. + public RabbitMQWrapper(RabbitMQConnector connector) => + _connector = connector ?? throw new ArgumentNullException(nameof(connector)); + + /// + /// Sends a serialized message to a RabbitMQ exchange. + /// + /// The type of the item being sent. It must implement and provide a serializer. + /// The item to send. It will be serialized to JSON before being sent. + /// The type of exchange to declare. If null, the exchange will not be declared. + /// A to allow cancellation of the operation. + /// + /// This method establishes a connection to RabbitMQ, declares an exchange if necessary, + /// serializes the provided item to JSON, and sends the message to the exchange. + /// + public async Task SendToExchangeAsync( + T item, + string exchangeDeclareType = null, + CancellationToken cancellationToken = default + ) + where T : class, IQueuing, new() + { + var exchangeName = Helpers.Extensions.GetExchangeName(); + + var connection = await _connector + .ConnectionFactory.CreateConnectionAsync(cancellationToken) + .ConfigureAwait(false); + await using var connectionDisposable = connection.ConfigureAwait(false); + + var channel = await connection + .CreateChannelAsync(cancellationToken: cancellationToken) + .ConfigureAwait(false); + await using var channelDisposable = channel.ConfigureAwait(false); + + // Declare the exchange if the type is provided + if (!string.IsNullOrWhiteSpace(exchangeDeclareType)) + { + await channel + .ExchangeDeclareAsync( + exchangeName, + exchangeDeclareType, + true, + false, + cancellationToken: cancellationToken + ) + .ConfigureAwait(false); + } + + // Serialize the item and prepare the message + var json = (string)item.GetSerializer(); + var body = Encoding.UTF8.GetBytes(json); + + var properties = new BasicProperties { Persistent = true }; + + // Publish the message to the exchange + await channel + .BasicPublishAsync( + exchangeName, + string.Empty, + false, + properties, + body, + cancellationToken + ) + .ConfigureAwait(false); + + LogConsumer.Trace("Sent to exchange {0} the item {1}", exchangeName, json); + } + + /// + /// Sends a serialized message to a RabbitMQ queue. + /// + /// The type of the item being sent. It must implement and provide a serializer. + /// The item to send. It will be serialized to JSON before being sent. + /// If true, the queue will be declared before sending the message. Default is true. + /// A to allow cancellation of the operation. + /// + /// This method establishes a connection to RabbitMQ, declares a queue if necessary, + /// serializes the provided item to JSON, and sends the message to the queue. + /// + public async Task SendToQueueAsync( + T item, + bool queueDeclare = true, + CancellationToken cancellationToken = default + ) + where T : class, IQueuing, new() + { + var queueName = Helpers.Extensions.GetQueueName(); + + var connection = await _connector + .ConnectionFactory.CreateConnectionAsync(cancellationToken) + .ConfigureAwait(false); + await using var connectionDisposable = connection.ConfigureAwait(false); + + var channel = await connection + .CreateChannelAsync(cancellationToken: cancellationToken) + .ConfigureAwait(false); + await using var channelDisposable = channel.ConfigureAwait(false); + + // Declare the queue if specified + if (queueDeclare) + { + await channel + .QueueDeclareAsync( + queueName, + true, + false, + false, + cancellationToken: cancellationToken + ) + .ConfigureAwait(false); + } + + // Serialize the item and prepare the message + var json = (string)item.GetSerializer(); + var body = Encoding.UTF8.GetBytes(json); + + var properties = new BasicProperties { Persistent = true }; + + // Publish the message to the queue + await channel + .BasicPublishAsync(string.Empty, queueName, false, properties, body, cancellationToken) + .ConfigureAwait(false); + + LogConsumer.Trace("Sent to queue {0} the item: {1}", queueName, json); + } +}