update: migrate RabbitMQ operations to async methods +semver: minor#953
Conversation
- Upgrade `RabbitMQ.Client` package from version 6.8.1 to 7.0.0. - Modify `MessageReceiver`, `RabbitMQLogProvider`, and `RabbitMQWrapper` classes to use asynchronous methods for RabbitMQ operations. - Change method names in `MessageReceiver` to `DoWorkAsync` and adapt logic to use `AsyncEventingBasicConsumer`. - Use `async` versions of `CreateConnection`, `CreateChannel`, `QueueDeclare`, and `BasicPublish` methods across the affected classes. - Improve graceful shutdown handling in `MessageReceiver` using `OperationCanceledException`. These changes enhance the scalability and responsiveness of the RabbitMQ operations by leveraging asynchronous programming.
Reviewer's GuideMigrates RabbitMQ usage from synchronous to asynchronous APIs across message receiving, publishing, and logging, aligned with the RabbitMQ.Client 7.x async model and improved cancellation-aware shutdown. Sequence diagram for asynchronous message receiving with cancellation-aware shutdownsequenceDiagram
participant Application
participant MessageReceiver
participant ConnectionFactory
participant Connection
participant Channel
participant AsyncConsumer
participant Subscriber
Application->>MessageReceiver: ReceiveFromQueue<T>(autoAck, cancellationToken)
MessageReceiver->>MessageReceiver: DoWorkAsync("", queueName, autoAck, cancellationToken)
MessageReceiver->>ConnectionFactory: CreateConnectionAsync(cancellationToken)
ConnectionFactory-->>MessageReceiver: connection
MessageReceiver->>Connection: CreateChannelAsync(cancellationToken)
Connection-->>MessageReceiver: channel
MessageReceiver->>Channel: QueueDeclareAsync(cancellationToken)
Channel-->>MessageReceiver: queueName
MessageReceiver->>Channel: QueueBindAsync(queueName, exchange, "", cancellationToken)
MessageReceiver->>AsyncConsumer: new AsyncEventingBasicConsumer(channel)
MessageReceiver->>Channel: BasicConsumeAsync(queueName, autoAck, AsyncConsumer, cancellationToken)
AsyncConsumer-->>MessageReceiver: ReceivedAsync(sender, args)
MessageReceiver->>Subscriber: MessageReceived(this, MessageReceivedArgs)
MessageReceiver->>MessageReceiver: Task.Delay(Timeout.Infinite, cancellationToken)
Note over MessageReceiver: Throws OperationCanceledException on cancellation
MessageReceiver-->>Application: graceful shutdown
Sequence diagram for asynchronous message publishing via RabbitMQWrappersequenceDiagram
participant Application
participant RabbitMQWrapper
participant ConnectionFactory
participant Connection
participant Channel
Application->>RabbitMQWrapper: SendToExchangeAsync<T>(item, exchangeDeclareType, cancellationToken)
RabbitMQWrapper->>ConnectionFactory: CreateConnectionAsync(cancellationToken)
ConnectionFactory-->>RabbitMQWrapper: connection
RabbitMQWrapper->>Connection: CreateChannelAsync(cancellationToken)
Connection-->>RabbitMQWrapper: channel
RabbitMQWrapper->>Channel: ExchangeDeclareAsync(exchangeName, exchangeDeclareType, true, false, cancellationToken)
RabbitMQWrapper->>RabbitMQWrapper: item.GetSerializer()
RabbitMQWrapper->>Channel: BasicPublishAsync(exchangeName, "", false, BasicProperties{Persistent=true}, body, cancellationToken)
Application->>RabbitMQWrapper: SendToQueueAsync<T>(item, queueDeclare, cancellationToken)
RabbitMQWrapper->>ConnectionFactory: CreateConnectionAsync(cancellationToken)
ConnectionFactory-->>RabbitMQWrapper: connection
RabbitMQWrapper->>Connection: CreateChannelAsync(cancellationToken)
Connection-->>RabbitMQWrapper: channel
RabbitMQWrapper->>Channel: QueueDeclareAsync(queueName, true, false, false, cancellationToken)
RabbitMQWrapper->>RabbitMQWrapper: item.GetSerializer()
RabbitMQWrapper->>Channel: BasicPublishAsync("", queueName, false, BasicProperties{Persistent=true}, body, cancellationToken)
File-Level Changes
Assessment against linked issues
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
github-actions
Bot
added
the
size/XL
WalkthroughUpgrades RabbitMQ.Client from 6.8.1 to 7.0.0 and refactors RabbitMQ consumption (MessageReceiver), log publishing (RabbitMQLogProvider), and message publishing (RabbitMQWrapper) from synchronous to async/await APIs with cancellation token support. Also updates the CSharpier linter GitHub Action reference to ChangesRabbitMQ.Client 7.0.0 async migration
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Caller
participant MessageReceiver
participant Channel as IChannel
participant Consumer as AsyncEventingBasicConsumer
Caller->>MessageReceiver: ReceiveFromQueue/ReceiveFromExchange
MessageReceiver->>MessageReceiver: DoWorkAsync(queue/exchange, autoAck, cancellationToken)
MessageReceiver->>Channel: CreateConnectionAsync/CreateChannelAsync
MessageReceiver->>Channel: QueueDeclareAsync/QueueBindAsync
MessageReceiver->>Consumer: register ReceivedAsync callback
Consumer-->>MessageReceiver: message body received
MessageReceiver->>MessageReceiver: raise MessageReceived event
MessageReceiver->>MessageReceiver: await Task.Delay until cancelled
sequenceDiagram
participant Caller
participant RabbitMQWrapper
participant Channel as IChannel
participant Broker as RabbitMQ Broker
Caller->>RabbitMQWrapper: SendToExchangeAsync/SendToQueueAsync(item, cancellationToken)
RabbitMQWrapper->>Channel: CreateConnectionAsync/CreateChannelAsync
RabbitMQWrapper->>Channel: ExchangeDeclareAsync/QueueDeclareAsync (if configured)
RabbitMQWrapper->>Channel: BasicPublishAsync(serialized item, BasicProperties)
Channel->>Broker: publish message
Channel-->>RabbitMQWrapper: dispose channel/connection (await using)
Possibly related issues
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
🔥 Formatting issues detected
⚡ Please run |
|
Review the following changes in direct dependencies. Learn more about Socket for GitHub.
|
|
Caution Review the following alerts detected in dependencies. According to your organization's Security Policy, you must resolve all "Block" alerts before proceeding. Learn more about Socket for GitHub.
|
|
@gstraccini csharpier |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 1 issue, and left some high level feedback:
- The
await using var connectionDisposable = connection.ConfigureAwait(false);/channelDisposablepattern is incorrect forIAsyncDisposable; you should useawait using var connection = connection;(and likewise forchannel) withoutConfigureAwaiton the disposable itself. - Both
ReceiveFromQueueandReceiveFromExchangeare still fire-and-forget (Task.Runwithout awaiting the returned task), which makes exception handling and lifecycle control of the consumer harder; consider making these methods async and returning the task so callers can observe failures and manage cancellation more explicitly.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The `await using var connectionDisposable = connection.ConfigureAwait(false);` / `channelDisposable` pattern is incorrect for `IAsyncDisposable`; you should use `await using var connection = connection;` (and likewise for `channel`) without `ConfigureAwait` on the disposable itself.
- Both `ReceiveFromQueue` and `ReceiveFromExchange` are still fire-and-forget (`Task.Run` without awaiting the returned task), which makes exception handling and lifecycle control of the consumer harder; consider making these methods async and returning the task so callers can observe failures and manage cancellation more explicitly.
## Individual Comments
### Comment 1
<location path="Src/CrispyWaffle.RabbitMQ/Helpers/MessageReceiver.cs" line_range="106-114" />
<code_context>
+ 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);
</code_context>
<issue_to_address>
**issue (bug_risk):** Incorrect use of ConfigureAwait with IAsyncDisposable will not compile and prevents proper disposal.
`ConfigureAwait` can only be called on `Task`/`ValueTask`, so `await using var connectionDisposable = connection.ConfigureAwait(false);` (and the equivalent for `channel`) will not compile and does not dispose the resources correctly. Instead, await the async factory methods and use `await using` directly on the resulting objects:
```csharp
await using var connection = await _connector
.ConnectionFactory.CreateConnectionAsync(cancellationToken)
.ConfigureAwait(false);
await using var channel = await connection
.CreateChannelAsync(cancellationToken: cancellationToken)
.ConfigureAwait(false);
```
This ensures the connection and channel are properly asynchronously disposed without extra `*Disposable` variables.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
|
Running CSharpier on this branch! 🔧 |
|
✅ CSharpier result: |
|
ℹ️ Workflow Template updated! A new workflow template has been added: |
gstraccini
Bot
added
☑️ auto-merge
gstraccini
Bot
added
🕔 high effort
![gstraccini[bot]](https://avatars.githubusercontent.com/in/480132?s=60&v=4){kind=small}
|
Infisical secrets check: ✅ No secrets leaked! 💻 Scan logs2026-07-04T19:42:24Z INF scanning for exposed secrets...
7:42PM INF 811 commits scanned.
2026-07-04T19:42:25Z INF scan completed in 1.26s
2026-07-04T19:42:25Z INF no leaks found
|
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
.github/workflows/linter.yml (1)
15-15: 🔒 Security & Privacy | 🔵 Trivial | ⚡ Quick winUnpinned action reference (
@latest) is a supply-chain and reproducibility risk.Switching from a pinned version/commit to the mutable
@latesttag means the workflow's behavior can silently change whenever the upstream action publishes a new release (or a compromised release), with no corresponding change in this repository. Prefer pinning to a specific version tag or, ideally, a commit SHA for tamper-resistance and reproducible CI runs.🔒️ Suggested fix
- uses: guibranco/github-csharpier-linter-action@latest + uses: guibranco/github-csharpier-linter-action@v1.0.25🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In @.github/workflows/linter.yml at line 15, The workflow step using guibranco/github-csharpier-linter-action is currently pinned to a mutable `@latest` reference, which should be replaced with a fixed version tag or commit SHA. Update the action reference in the linter workflow to a specific immutable version so CI stays reproducible and resistant to upstream changes. Keep the same action and step structure, only change the version selector to a pinned identifier.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@Src/CrispyWaffle.RabbitMQ/Helpers/MessageReceiver.cs`:
- Around line 55-65: The ReceiveFromQueue<T> and ReceiveFromExchange<T> startup
paths discard the Task.Run result, so failures in DoWorkAsync can be lost and
the caller never knows the consumer did not start. Update the MessageReceiver
methods to either return/await the Task from Task.Run or attach explicit
exception handling and fault logging around DoWorkAsync, and remove the
ineffective ConfigureAwait(false) on the discarded task. Use the
ReceiveFromQueue<T> and ReceiveFromExchange<T> methods to locate the change.
In `@Src/CrispyWaffle.RabbitMQ/Log/RabbitMQLogProvider.cs`:
- Around line 59-75: The RabbitMQ connection created in
RabbitMQLogProvider.Initialize is never retained or disposed, causing a resource
leak. Store the result of ConnectionFactory.CreateConnectionAsync in a dedicated
field on RabbitMQLogProvider, then use that field when creating the channel and
ensure Dispose(bool) closes and disposes both the channel and the connection.
Keep the cleanup logic aligned with the existing _channel disposal path so the
provider owns the full connection lifecycle.
---
Nitpick comments:
In @.github/workflows/linter.yml:
- Line 15: The workflow step using guibranco/github-csharpier-linter-action is
currently pinned to a mutable `@latest` reference, which should be replaced with a
fixed version tag or commit SHA. Update the action reference in the linter
workflow to a specific immutable version so CI stays reproducible and resistant
to upstream changes. Keep the same action and step structure, only change the
version selector to a pinned identifier.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 71e52877-b2e8-410e-ad7a-f1e8068b3d47
📒 Files selected for processing (5)
.github/workflows/linter.ymlDirectory.Packages.propsSrc/CrispyWaffle.RabbitMQ/Helpers/MessageReceiver.csSrc/CrispyWaffle.RabbitMQ/Log/RabbitMQLogProvider.csSrc/CrispyWaffle.RabbitMQ/Utils/Communications/RabbitMQWrapper.cs
2 participants
📑 Description
RabbitMQ.Clientpackage from version 6.8.1 to 7.0.0.MessageReceiver,RabbitMQLogProvider, andRabbitMQWrapperclasses to use asynchronous methods for RabbitMQ operations.MessageReceivertoDoWorkAsyncand adapt logic to useAsyncEventingBasicConsumer.asyncversions ofCreateConnection,CreateChannel,QueueDeclare, andBasicPublishmethods across the affected classes.MessageReceiverusingOperationCanceledException.These changes enhance the scalability and responsiveness of the RabbitMQ operations by leveraging asynchronous programming.
✅ Checks
☢️ Does this introduce a breaking change?
Summary by Sourcery
Migrate RabbitMQ integration to the client library’s asynchronous APIs to improve scalability and cancellation-aware processing.
New Features:
Enhancements:
Build:
Summary by CodeRabbit
New Features
Bug Fixes
Chores