forked from LykkeCity/Lykke.RabbitMqDotNetBroker
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
test(LT-5705): covered message read strategy configuration with tests
- Loading branch information
atarutin
committed
Sep 24, 2024
1 parent
c477a04
commit bae0d2c
Showing
11 changed files
with
1,541 additions
and
41 deletions.
There are no files selected for viewing
35 changes: 35 additions & 0 deletions
35
...itMqBroker/Subscriber/MessageReadStrategies/MessageReadStrategyConfigurationExtensions.cs
This file contains 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,35 @@ | ||
| // Copyright (c) 2024 Lykke Corp. | ||
| // See the LICENSE file in the project root for more information. | ||
|
|
||
| using System; | ||
|
|
||
| using RabbitMQ.Client; | ||
|
|
||
| namespace Lykke.RabbitMqBroker.Subscriber.MessageReadStrategies; | ||
|
|
||
| internal static class MessageReadStrategyConfigurationExtensions | ||
| { | ||
| public static IConfigurationResult<QueueName> StrategyTryConfigure(this Func<IModel> channelFactory, QueueConfigurationOptions options) | ||
| { | ||
| var queueConfigurationResult = QueueConfigurator.Configure(channelFactory, options); | ||
|
|
||
| if (queueConfigurationResult.IsSuccess && options.DeadLetterExchangeName is not null) | ||
| { | ||
| var dlxConfigurationResult = ExchangeConfigurator.ConfigureDlx(channelFactory, options).Match( | ||
| onSuccess: () => QueueConfigurator.ConfigurePoison(channelFactory, options)); | ||
|
|
||
| return dlxConfigurationResult.IsSuccess | ||
| ? queueConfigurationResult | ||
| : ConfigurationResult<QueueName>.Failure(dlxConfigurationResult.Error); | ||
| } | ||
|
|
||
| return queueConfigurationResult; | ||
| } | ||
|
|
||
| public static IConfigurationResult<QueueName> StrategyRetryWithQueueRecreation(this Func<IModel> channelFactory, QueueConfigurationOptions options) | ||
| { | ||
| return channelFactory.SafeDeleteQueue(options.QueueName.ToString()).Match( | ||
| onSuccess: _ => StrategyTryConfigure(channelFactory, options), | ||
| onFailure: _ => throw new InvalidOperationException($"Failed to delete queue [{options.QueueName}].")); | ||
| } | ||
| } |
This file contains 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 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
264 changes: 264 additions & 0 deletions
264
tests/Lykke.RabbitMqBroker.Tests/Fakes/ExchangeDeclarationExpectedFailureFakeChannel.cs
This file contains 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,264 @@ | ||
| using System; | ||
| using System.Collections.Generic; | ||
|
|
||
| using RabbitMQ.Client; | ||
| using RabbitMQ.Client.Events; | ||
| using RabbitMQ.Client.Exceptions; | ||
|
|
||
| namespace Lykke.RabbitMqBroker.Tests.Fakes; | ||
|
|
||
| public class ExchangeDeclarationExpectedFailureFakeChannel : IModel | ||
| { | ||
| public int ChannelNumber => throw new NotImplementedException(); | ||
|
|
||
| public ShutdownEventArgs CloseReason => throw new NotImplementedException(); | ||
|
|
||
| public IBasicConsumer DefaultConsumer { get => throw new NotImplementedException(); set => throw new NotImplementedException(); } | ||
|
|
||
| public bool IsClosed => throw new NotImplementedException(); | ||
|
|
||
| public bool IsOpen => throw new NotImplementedException(); | ||
|
|
||
| public ulong NextPublishSeqNo => throw new NotImplementedException(); | ||
|
|
||
| public TimeSpan ContinuationTimeout { get => throw new NotImplementedException(); set => throw new NotImplementedException(); } | ||
|
|
||
| public event EventHandler<BasicAckEventArgs> BasicAcks; | ||
| public event EventHandler<BasicNackEventArgs> BasicNacks; | ||
| public event EventHandler<EventArgs> BasicRecoverOk; | ||
| public event EventHandler<BasicReturnEventArgs> BasicReturn; | ||
| public event EventHandler<CallbackExceptionEventArgs> CallbackException; | ||
| public event EventHandler<FlowControlEventArgs> FlowControl; | ||
| public event EventHandler<ShutdownEventArgs> ModelShutdown; | ||
|
|
||
| public void Abort() | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void Abort(ushort replyCode, string replyText) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void BasicAck(ulong deliveryTag, bool multiple) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void BasicCancel(string consumerTag) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void BasicCancelNoWait(string consumerTag) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public string BasicConsume(string queue, bool autoAck, string consumerTag, bool noLocal, bool exclusive, IDictionary<string, object> arguments, IBasicConsumer consumer) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public BasicGetResult BasicGet(string queue, bool autoAck) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void BasicNack(ulong deliveryTag, bool multiple, bool requeue) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void BasicPublish(string exchange, string routingKey, bool mandatory, IBasicProperties basicProperties, ReadOnlyMemory<byte> body) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void BasicQos(uint prefetchSize, ushort prefetchCount, bool global) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void BasicRecover(bool requeue) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void BasicRecoverAsync(bool requeue) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void BasicReject(ulong deliveryTag, bool requeue) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void Close() | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void Close(ushort replyCode, string replyText) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void ConfirmSelect() | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public uint ConsumerCount(string queue) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public IBasicProperties CreateBasicProperties() | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public IBasicPublishBatch CreateBasicPublishBatch() | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void Dispose() | ||
| { | ||
| } | ||
|
|
||
| public void ExchangeBind(string destination, string source, string routingKey, IDictionary<string, object> arguments) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void ExchangeBindNoWait(string destination, string source, string routingKey, IDictionary<string, object> arguments) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete, IDictionary<string, object> arguments) | ||
| { | ||
| throw new OperationInterruptedException(new ShutdownEventArgs(ShutdownInitiator.Application, Constants.PreconditionFailed, "Precondition failed", nameof(ExchangeDeclare))); | ||
| } | ||
|
|
||
| public void ExchangeDeclareNoWait(string exchange, string type, bool durable, bool autoDelete, IDictionary<string, object> arguments) | ||
| { | ||
| throw new OperationInterruptedException(new ShutdownEventArgs(ShutdownInitiator.Application, Constants.PreconditionFailed, "Precondition failed", nameof(ExchangeDeclareNoWait))); | ||
| } | ||
|
|
||
| public void ExchangeDeclarePassive(string exchange) | ||
| { | ||
| throw new OperationInterruptedException(new ShutdownEventArgs(ShutdownInitiator.Application, Constants.PreconditionFailed, "Precondition failed", nameof(ExchangeDeclarePassive))); | ||
| } | ||
|
|
||
| public void ExchangeDelete(string exchange, bool ifUnused) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void ExchangeDeleteNoWait(string exchange, bool ifUnused) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void ExchangeUnbind(string destination, string source, string routingKey, IDictionary<string, object> arguments) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void ExchangeUnbindNoWait(string destination, string source, string routingKey, IDictionary<string, object> arguments) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public uint MessageCount(string queue) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void QueueBind(string queue, string exchange, string routingKey, IDictionary<string, object> arguments) | ||
| { | ||
| } | ||
|
|
||
| public void QueueBindNoWait(string queue, string exchange, string routingKey, IDictionary<string, object> arguments) | ||
| { | ||
| } | ||
|
|
||
| public QueueDeclareOk QueueDeclare(string queue, bool durable, bool exclusive, bool autoDelete, IDictionary<string, object> arguments) | ||
| { | ||
| return new QueueDeclareOk(queue, 0, 0); | ||
| } | ||
|
|
||
| public void QueueDeclareNoWait(string queue, bool durable, bool exclusive, bool autoDelete, IDictionary<string, object> arguments) | ||
| { | ||
| } | ||
|
|
||
| public QueueDeclareOk QueueDeclarePassive(string queue) | ||
| { | ||
| return new QueueDeclareOk(queue, 0, 0); | ||
| } | ||
|
|
||
| public uint QueueDelete(string queue, bool ifUnused, bool ifEmpty) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void QueueDeleteNoWait(string queue, bool ifUnused, bool ifEmpty) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public uint QueuePurge(string queue) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void QueueUnbind(string queue, string exchange, string routingKey, IDictionary<string, object> arguments) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void TxCommit() | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void TxRollback() | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void TxSelect() | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public bool WaitForConfirms() | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public bool WaitForConfirms(TimeSpan timeout) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public bool WaitForConfirms(TimeSpan timeout, out bool timedOut) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void WaitForConfirmsOrDie() | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
|
|
||
| public void WaitForConfirmsOrDie(TimeSpan timeout) | ||
| { | ||
| throw new NotImplementedException(); | ||
| } | ||
| } |
Oops, something went wrong.