V4.1.0/support for rabbitmq #35
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,121 @@ | ||
| using Cuemon; | ||
| using Cuemon.Extensions; | ||
| using Cuemon.Extensions.IO; | ||
| using Cuemon.Extensions.Reflection; | ||
| using Cuemon.Threading; | ||
| using RabbitMQ.Client; | ||
| using RabbitMQ.Client.Events; | ||
| using Savvyio.Commands; | ||
| using Savvyio.Messaging; | ||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Threading; | ||
| using System.Threading.Channels; | ||
| using System.Threading.Tasks; | ||
|
|
||
| namespace Savvyio.Extensions.RabbitMQ.Commands | ||
| { | ||
| /// <summary> | ||
| /// Represents a RabbitMQ-based implementation of a point-to-point command queue. | ||
| /// </summary> | ||
| public class RabbitMqCommandQueue : RabbitMqMessage, IPointToPointChannel<ICommand> | ||
| { | ||
| private readonly RabbitMqCommandQueueOptions _options; | ||
|
|
||
| /// <summary> | ||
| /// Initializes a new instance of the <see cref="RabbitMqCommandQueue"/> class. | ||
| /// </summary> | ||
| /// <param name="marshaller">The marshaller used for serializing and deserializing messages.</param> | ||
| /// <param name="options">The options used to configure the RabbitMQ command queue.</param> | ||
| /// <exception cref="ArgumentNullException"> | ||
| /// <paramref name="marshaller"/> cannot be null -or- | ||
| /// <paramref name="options"/> cannot be null. | ||
| /// </exception> | ||
| /// <exception cref="ArgumentException"> | ||
| /// <paramref name="options"/> are not in a valid state. | ||
| /// </exception> | ||
| public RabbitMqCommandQueue(IMarshaller marshaller, RabbitMqCommandQueueOptions options) : base(marshaller, options) | ||
| { | ||
| Validator.ThrowIfInvalidOptions(options); | ||
| _options = options; | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Sends the specified command messages asynchronously to the configured RabbitMQ queue. | ||
| /// </summary> | ||
| /// <param name="messages">The messages to send.</param> | ||
| /// <param name="setup">The <see cref="AsyncOptions"/> which may be configured.</param> | ||
| /// <returns>A <see cref="Task"/> that represents the asynchronous operation.</returns> | ||
| public async Task SendAsync(IEnumerable<IMessage<ICommand>> messages, Action<AsyncOptions> setup = null) | ||
| { | ||
| Validator.ThrowIfInvalidConfigurator(setup, out var options); | ||
|
|
||
| await EnsureConnectivityAsync(options.CancellationToken).ConfigureAwait(false); | ||
|
|
||
| await RabbitMqChannel.QueueDeclareAsync(_options.QueueName, false, false, false, cancellationToken: options.CancellationToken).ConfigureAwait(false); | ||
|
|
||
| foreach (var message in messages) | ||
| { | ||
| await RabbitMqChannel.BasicPublishAsync("", _options.QueueName, true, basicProperties: new BasicProperties() | ||
| { | ||
| //Persistent = true, | ||
| Headers = new Dictionary<string, object>() | ||
| { | ||
| { MessageType, message.GetType().ToFullNameIncludingAssemblyName() } | ||
| } | ||
| }, body: await Marshaller.Serialize(message).ToByteArrayAsync(o => o.CancellationToken = options.CancellationToken).ConfigureAwait(false), options.CancellationToken).ConfigureAwait(false); | ||
| } | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Receives command messages asynchronously from the configured RabbitMQ queue. | ||
| /// </summary> | ||
| /// <param name="setup">The <see cref="AsyncOptions"/> which may be configured.</param> | ||
| /// <returns> | ||
| /// An <see cref="IAsyncEnumerable{T}"/> that yields <see cref="IMessage{ICommand}"/> instances as they are received. | ||
| /// </returns> | ||
| public async IAsyncEnumerable<IMessage<ICommand>> ReceiveAsync(Action<AsyncOptions> setup = null) | ||
| { | ||
| Validator.ThrowIfInvalidConfigurator(setup, out var options); | ||
|
|
||
| await EnsureConnectivityAsync(options.CancellationToken).ConfigureAwait(false); | ||
|
|
||
| await RabbitMqChannel.QueueDeclareAsync(_options.QueueName, false, false, false, cancellationToken: options.CancellationToken).ConfigureAwait(false); | ||
|
|
||
| var buffer = Channel.CreateUnbounded<IMessage<ICommand>>(); | ||
| var consumer = new AsyncEventingBasicConsumer(RabbitMqChannel); | ||
| consumer.ReceivedAsync += async (_, e) => | ||
| { | ||
| var messageType = Type.GetType((e.BasicProperties.Headers[MessageType] as byte[]).ToEncodedString()); | ||
| var deserialized = Marshaller.Deserialize(e.Body.ToArray().ToStream(), messageType) as IMessage<ICommand>; | ||
| deserialized!.Properties.Add(nameof(BasicDeliverEventArgs.DeliveryTag), e.DeliveryTag); | ||
| deserialized.Properties.Add(nameof(CancellationToken), options.CancellationToken); | ||
| deserialized.Properties.Add(nameof(QueueDeclareOk.QueueName), _options.QueueName); | ||
| await buffer.Writer.WriteAsync(deserialized, options.CancellationToken).ConfigureAwait(false); | ||
| }; | ||
|
|
||
| await RabbitMqChannel.BasicConsumeAsync(_options.QueueName, autoAck: false, consumer: consumer, cancellationToken: options.CancellationToken).ConfigureAwait(false); | ||
|
|
||
| while (await buffer.Reader.WaitToReadAsync(options.CancellationToken).ConfigureAwait(false)) | ||
| { | ||
| while (buffer.Reader.TryRead(out var message)) | ||
| { | ||
| message.Acknowledged += OnMessageAcknowledgedAsync; | ||
| if (_options.AutoAcknowledge) | ||
| { | ||
| await message.AcknowledgeAsync().ConfigureAwait(false); | ||
| } | ||
| yield return message; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private async Task OnMessageAcknowledgedAsync(object sender, AcknowledgedEventArgs e) | ||
| { | ||
| var ct = (CancellationToken)e.Properties[nameof(CancellationToken)]; | ||
| var queueName = e.Properties[nameof(QueueDeclareOk.QueueName)] as string; | ||
| await RabbitMqChannel.QueueDeclareAsync(queueName, false, false, false, cancellationToken: ct).ConfigureAwait(false); | ||
| await RabbitMqChannel.BasicAckAsync((ulong)e.Properties[nameof(BasicDeliverEventArgs.DeliveryTag)], false, ct).ConfigureAwait(false); | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| using System; | ||
| using Cuemon; | ||
|
|
||
| namespace Savvyio.Extensions.RabbitMQ.Commands | ||
| { | ||
| /// <summary> | ||
| /// Configuration options for <see cref="RabbitMqCommandQueue"/>. | ||
| /// </summary> | ||
| /// <seealso cref="RabbitMqMessageOptions"/> | ||
| public class RabbitMqCommandQueueOptions : RabbitMqMessageOptions | ||
| { | ||
| /// <summary> | ||
| /// Initializes a new instance of the <see cref="RabbitMqCommandQueueOptions"/> class with default values. | ||
| /// </summary> | ||
| /// <remarks> | ||
| /// The following table shows the initial property values for an instance of <see cref="RabbitMqCommandQueueOptions"/>. | ||
| /// <list type="table"> | ||
| /// <listheader> | ||
| /// <term>Property</term> | ||
| /// <description>Initial Value</description> | ||
| /// </listheader> | ||
| /// <item> | ||
| /// <term><see cref="QueueName"/></term> | ||
| /// <description><c>null</c></description> | ||
| /// </item> | ||
| /// <item> | ||
| /// <term><see cref="AutoAcknowledge"/></term> | ||
| /// <description><c>false</c></description> | ||
| /// </item> | ||
| /// </list> | ||
| /// </remarks> | ||
| public RabbitMqCommandQueueOptions() | ||
| { | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Gets or sets the name of the queue. | ||
| /// </summary> | ||
| /// <value> | ||
| /// The name of the RabbitMQ queue to be used for command messages. | ||
| /// </value> | ||
| public string QueueName { get; set; } | ||
|
|
||
| /// <summary> | ||
| /// Gets or sets a value indicating whether messages should be automatically acknowledged. | ||
| /// </summary> | ||
| /// <value> | ||
| /// <c>true</c> if messages are automatically acknowledged; otherwise, <c>false</c>. | ||
| /// </value> | ||
| public bool AutoAcknowledge { get; set; } | ||
|
|
||
| /// <summary> | ||
| /// Determines whether the public read-write properties of this instance are in a valid state. | ||
| /// </summary> | ||
| /// <remarks>This method is expected to throw exceptions when one or more conditions fails to be in a valid state.</remarks> | ||
| /// <exception cref="InvalidOperationException"> | ||
| /// <see cref="QueueName"/> cannot be null or empty. | ||
| /// </exception> | ||
| public override void ValidateOptions() | ||
| { | ||
| Validator.ThrowIfInvalidState(string.IsNullOrEmpty(QueueName)); | ||
| base.ValidateOptions(); | ||
| } | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,108 @@ | ||
| using Cuemon; | ||
| using Cuemon.Extensions; | ||
| using Cuemon.Extensions.IO; | ||
| using Cuemon.Extensions.Reflection; | ||
| using Cuemon.Threading; | ||
| using RabbitMQ.Client; | ||
| using RabbitMQ.Client.Events; | ||
| using Savvyio.EventDriven; | ||
| using Savvyio.Messaging; | ||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Threading; | ||
| using System.Threading.Channels; | ||
| using System.Threading.Tasks; | ||
|
|
||
| namespace Savvyio.Extensions.RabbitMQ.EventDriven | ||
| { | ||
| /// <summary> | ||
| /// Represents a RabbitMQ-based implementation of a publish-subscribe event bus for integration events. | ||
| /// </summary> | ||
| public class RabbitMqEventBus : RabbitMqMessage, IPublishSubscribeChannel<IIntegrationEvent> | ||
| { | ||
| private readonly ConnectionFactory _factory; | ||
|
Check warning on line 23 in src/Savvyio.Extensions.RabbitMQ/EventDriven/RabbitMqEventBus.cs
|
||
| private readonly RabbitMqEventBusOptions _options; | ||
coderabbitai[bot] marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
|
|
||
| /// <summary> | ||
| /// Initializes a new instance of the <see cref="RabbitMqEventBus"/> class. | ||
| /// </summary> | ||
| /// <param name="marshaller">The marshaller used for serializing and deserializing messages.</param> | ||
| /// <param name="options">The options used to configure the RabbitMQ event bus.</param> | ||
| /// <exception cref="ArgumentNullException"> | ||
| /// <paramref name="marshaller"/> cannot be null -or- | ||
| /// <paramref name="options"/> cannot be null. | ||
| /// </exception> | ||
| /// <exception cref="ArgumentException"> | ||
| /// <paramref name="options"/> are not in a valid state. | ||
| /// </exception> | ||
| public RabbitMqEventBus(IMarshaller marshaller, RabbitMqEventBusOptions options) : base(marshaller, options) | ||
| { | ||
| Validator.ThrowIfInvalidOptions(options); | ||
| _options = options; | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Publishes the specified integration event message asynchronously to the configured RabbitMQ exchange. | ||
| /// </summary> | ||
| /// <param name="message">The message to publish.</param> | ||
| /// <param name="setup">The <see cref="AsyncOptions"/> which may be configured.</param> | ||
| /// <returns>A <see cref="Task"/> that represents the asynchronous operation.</returns> | ||
| public async Task PublishAsync(IMessage<IIntegrationEvent> message, Action<AsyncOptions> setup = null) | ||
| { | ||
| Validator.ThrowIfInvalidConfigurator(setup, out var options); | ||
|
|
||
| await EnsureConnectivityAsync(options.CancellationToken).ConfigureAwait(false); | ||
|
|
||
| await RabbitMqChannel.ExchangeDeclareAsync(_options.ExchangeName, ExchangeType.Fanout, cancellationToken: options.CancellationToken).ConfigureAwait(false); | ||
|
|
||
| await RabbitMqChannel.BasicPublishAsync(_options.ExchangeName, "", false, basicProperties: new BasicProperties() | ||
| { | ||
| Headers = new Dictionary<string, object>() | ||
| { | ||
| { MessageType, message.GetType().ToFullNameIncludingAssemblyName() } | ||
| } | ||
| }, body: await Marshaller.Serialize(message).ToByteArrayAsync(o => o.CancellationToken = options.CancellationToken).ConfigureAwait(false), options.CancellationToken).ConfigureAwait(false); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Subscribes to integration event messages from the configured RabbitMQ exchange and invokes the specified asynchronous handler for each received message. | ||
| /// </summary> | ||
| /// <param name="asyncHandler">The function delegate that will handle the message.</param> | ||
| /// <param name="setup">The <see cref="SubscribeAsyncOptions"/> which may be configured.</param> | ||
| /// <returns>A <see cref="Task"/> that represents the asynchronous operation.</returns> | ||
| public async Task SubscribeAsync(Func<IMessage<IIntegrationEvent>, CancellationToken, Task> asyncHandler, Action<SubscribeAsyncOptions> setup = null) | ||
| { | ||
| Validator.ThrowIfInvalidConfigurator(setup, out var options); | ||
|
|
||
| await EnsureConnectivityAsync(options.CancellationToken).ConfigureAwait(false); | ||
|
|
||
| await RabbitMqChannel.ExchangeDeclareAsync(_options.ExchangeName, ExchangeType.Fanout, cancellationToken: options.CancellationToken).ConfigureAwait(false); | ||
|
|
||
| var queue = await RabbitMqChannel.QueueDeclareAsync(cancellationToken: options.CancellationToken).ConfigureAwait(false); | ||
|
|
||
| await RabbitMqChannel.QueueBindAsync(queue.QueueName, _options.ExchangeName, "", cancellationToken: options.CancellationToken).ConfigureAwait(false); | ||
|
|
||
| var buffer = Channel.CreateUnbounded<IMessage<IIntegrationEvent>>(); | ||
| var consumer = new AsyncEventingBasicConsumer(RabbitMqChannel); | ||
| consumer.ReceivedAsync += async (_, e) => | ||
| { | ||
| var messageType = Type.GetType((e.BasicProperties.Headers[MessageType] as byte[]).ToEncodedString()); | ||
| var deserialized = Marshaller.Deserialize(e.Body.ToArray().ToStream(), messageType) as IMessage<IIntegrationEvent>; | ||
| deserialized!.Properties.Add(nameof(BasicDeliverEventArgs.DeliveryTag), e.DeliveryTag); | ||
| deserialized.Properties.Add(nameof(CancellationToken), options.CancellationToken); | ||
| deserialized.Properties.Add(nameof(QueueDeclareOk.QueueName), queue.QueueName); | ||
| await buffer.Writer.WriteAsync(deserialized, options.CancellationToken).ConfigureAwait(false); | ||
| }; | ||
|
|
||
| await RabbitMqChannel.BasicConsumeAsync(queue.QueueName, autoAck: true, consumer: consumer, options.CancellationToken).ConfigureAwait(false); | ||
|
|
||
| while (await buffer.Reader.WaitToReadAsync(options.CancellationToken).ConfigureAwait(false)) | ||
| { | ||
| while (buffer.Reader.TryRead(out var message)) | ||
| { | ||
| await asyncHandler(message, options.CancellationToken).ConfigureAwait(false); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.