Skip to content

update: migrate RabbitMQ operations to async methods +semver: minor #953

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
guibranco merged 3 commits into from
Jul 4, 2026
Merged

update: migrate RabbitMQ operations to async methods +semver: minor #953

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f71fa3f
update: migrate RabbitMQ operations to async methods +semver: minor
guibranco Jul 4, 2026
05c1079
CSharpier format + Add workflow template
gstraccini[bot] Jul 4, 2026
b954ce9
Update linter.yml
guibranco Jul 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Directory.Packages.props
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
<PackageVersion Include="NEST" Version="7.17.5" />
<PackageVersion Include="Newtonsoft.Json" Version="13.0.4" />
<PackageVersion Include="NSubstitute" Version="5.3.0" />
<PackageVersion Include="RabbitMQ.Client" Version="6.8.1" />
<PackageVersion Include="RabbitMQ.Client" Version="7.0.0" />
<PackageVersion Include="SonarAnalyzer.CSharp" Version="10.22.0.136894" />
<PackageVersion Include="StackExchange.Redis" Version="2.12.14" />
<PackageVersion Include="StackExchange.Redis.Extensions.Core" Version="9.1.0" />
Expand Down
313 changes: 170 additions & 143 deletions Src/CrispyWaffle.RabbitMQ/Helpers/MessageReceiver.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,143 +1,170 @@
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;

/// <summary>
/// A class that facilitates receiving messages from RabbitMQ queues or exchanges.
/// </summary>
/// <remarks>
/// The <see cref="MessageReceiver"/> class provides methods to receive messages from RabbitMQ queues or exchanges,
/// handling the message reception process and invoking events when messages are received.
/// </remarks>
public class MessageReceiver
{
/// <summary>
/// The RabbitMQ connection connector.
/// </summary>
private readonly RabbitMQConnector _connector;

/// <summary>
/// Initializes a new instance of the <see cref="MessageReceiver"/> class.
/// </summary>
/// <param name="connector">The <see cref="RabbitMQConnector"/> to be used for connecting to RabbitMQ.</param>
/// <exception cref="ArgumentNullException">Thrown if the <paramref name="connector"/> is null.</exception>
public MessageReceiver(RabbitMQConnector connector) =>
_connector = connector ?? throw new ArgumentNullException(nameof(connector));

/// <summary>
/// Delegate for handling received messages.
/// </summary>
/// <param name="sender">The sender of the message.</param>
/// <param name="e">The <see cref="MessageReceivedArgs"/> containing details of the received message.</param>
public delegate void MessageReceivedHandler(object sender, MessageReceivedArgs e);

/// <summary>
/// Event triggered when a message is received from the queue or exchange.
/// </summary>
public event MessageReceivedHandler MessageReceived;

/// <summary>
/// Starts receiving messages from a RabbitMQ queue.
/// </summary>
/// <typeparam name="T">The type of message to receive, which should implement <see cref="IQueuing"/>.</typeparam>
/// <param name="autoAck">If set to <c>true</c>, messages will be acknowledged automatically.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> to allow cancellation of the operation.</param>
/// <remarks>
/// This method will begin a background task to receive messages from the specified queue.
/// </remarks>
public void ReceiveFromQueue<T>(bool autoAck, CancellationToken cancellationToken)
where T : class, IQueuing, new()
{
var queueName = Extensions.GetQueueName<T>();

Task.Run(
() => DoWork(string.Empty, queueName, autoAck, cancellationToken),
cancellationToken
)
.ConfigureAwait(false);
}

/// <summary>
/// Starts receiving messages from a RabbitMQ exchange.
/// </summary>
/// <typeparam name="T">The type of message to receive, which should implement <see cref="IQueuing"/>.</typeparam>
/// <param name="autoAck">If set to <c>true</c>, messages will be acknowledged automatically.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> to allow cancellation of the operation.</param>
/// <remarks>
/// This method will begin a background task to receive messages from the specified exchange.
/// </remarks>
public void ReceiveFromExchange<T>(bool autoAck, CancellationToken cancellationToken)
where T : class, IQueuing, new()
{
var exchangeName = Extensions.GetExchangeName<T>();

Task.Run(
() => DoWork(exchangeName, string.Empty, autoAck, cancellationToken),
cancellationToken
)
.ConfigureAwait(false);
}

/// <summary>
/// Handles the work of receiving messages from a specified exchange or queue.
/// </summary>
/// <param name="exchange">The name of the exchange (optional).</param>
/// <param name="queue">The name of the queue (optional).</param>
/// <param name="autoAck">If set to <c>true</c>, messages will be acknowledged automatically.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> to allow cancellation of the operation.</param>
/// <remarks>
/// This method connects to RabbitMQ, binds the queue to the exchange (if specified),
/// and starts listening for messages. It invokes the <see cref="MessageReceived"/> event when a message is received.
/// </remarks>
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;

/// <summary>
/// A class that facilitates receiving messages from RabbitMQ queues or exchanges.
/// </summary>
/// <remarks>
/// The <see cref="MessageReceiver"/> class provides methods to receive messages from RabbitMQ queues or exchanges,
/// handling the message reception process and invoking events when messages are received.
/// </remarks>
public class MessageReceiver
{
/// <summary>
/// The RabbitMQ connection connector.
/// </summary>
private readonly RabbitMQConnector _connector;

/// <summary>
/// Initializes a new instance of the <see cref="MessageReceiver"/> class.
/// </summary>
/// <param name="connector">The <see cref="RabbitMQConnector"/> to be used for connecting to RabbitMQ.</param>
/// <exception cref="ArgumentNullException">Thrown if the <paramref name="connector"/> is null.</exception>
public MessageReceiver(RabbitMQConnector connector) =>
_connector = connector ?? throw new ArgumentNullException(nameof(connector));

/// <summary>
/// Delegate for handling received messages.
/// </summary>
/// <param name="sender">The sender of the message.</param>
/// <param name="e">The <see cref="MessageReceivedArgs"/> containing details of the received message.</param>
public delegate void MessageReceivedHandler(object sender, MessageReceivedArgs e);

/// <summary>
/// Event triggered when a message is received from the queue or exchange.
/// </summary>
public event MessageReceivedHandler MessageReceived;

/// <summary>
/// Starts receiving messages from a RabbitMQ queue.
/// </summary>
/// <typeparam name="T">The type of message to receive, which should implement <see cref="IQueuing"/>.</typeparam>
/// <param name="autoAck">If set to <c>true</c>, messages will be acknowledged automatically.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> to allow cancellation of the operation.</param>
/// <remarks>
/// This method will begin a background task to receive messages from the specified queue.
/// </remarks>
public void ReceiveFromQueue<T>(bool autoAck, CancellationToken cancellationToken)
where T : class, IQueuing, new()
{
var queueName = Extensions.GetQueueName<T>();

Task.Run(
() => DoWorkAsync(string.Empty, queueName, autoAck, cancellationToken),
cancellationToken
)
.ConfigureAwait(false);
}
Comment thread
guibranco marked this conversation as resolved.

/// <summary>
/// Starts receiving messages from a RabbitMQ exchange.
/// </summary>
/// <typeparam name="T">The type of message to receive, which should implement <see cref="IQueuing"/>.</typeparam>
/// <param name="autoAck">If set to <c>true</c>, messages will be acknowledged automatically.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> to allow cancellation of the operation.</param>
/// <remarks>
/// This method will begin a background task to receive messages from the specified exchange.
/// </remarks>
public void ReceiveFromExchange<T>(bool autoAck, CancellationToken cancellationToken)
where T : class, IQueuing, new()
{
var exchangeName = Extensions.GetExchangeName<T>();

Task.Run(
() => DoWorkAsync(exchangeName, string.Empty, autoAck, cancellationToken),
cancellationToken
)
.ConfigureAwait(false);
}

/// <summary>
/// Handles the work of receiving messages from a specified exchange or queue.
/// </summary>
/// <param name="exchange">The name of the exchange (optional).</param>
/// <param name="queue">The name of the queue (optional).</param>
/// <param name="autoAck">If set to <c>true</c>, messages will be acknowledged automatically.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> to allow cancellation of the operation.</param>
/// <remarks>
/// This method connects to RabbitMQ, binds the queue to the exchange (if specified),
/// and starts listening for messages. It invokes the <see cref="MessageReceived"/> event when a message is received.
/// </remarks>
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);
Comment thread
guibranco marked this conversation as resolved.

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.
}
}
}
35 changes: 30 additions & 5 deletions Src/CrispyWaffle.RabbitMQ/Log/RabbitMQLogProvider.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ public class RabbitMQLogProvider : ILogProvider, IDisposable
/// <summary>
/// The channel.
/// </summary>
private readonly IModel _channel;
private readonly IChannel _channel;

/// <summary>
/// The cancellation token.
Expand All @@ -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();
Comment thread
guibranco marked this conversation as resolved.
_cancellationToken = cancellationToken;
_queue = new ConcurrentQueue<string>();
var thread = new Thread(Worker);
Expand Down Expand Up @@ -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<byte>)body,
_cancellationToken
)
.GetAwaiter()
.GetResult();
}
catch (Exception e)
{
Expand All @@ -126,7 +150,8 @@ private void Dispose(bool disposing)
return;
}

_channel.Close();
_channel.CloseAsync().GetAwaiter().GetResult();
_channel.Dispose();
}

/// <summary>
Expand Down
Loading
Loading