diff --git a/README.md b/README.md index a52bb766..0d27efda 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,8 @@ -# RabbitMQ Amqp1.0 DotNet Client +# RabbitMQ AMQP 1.0 DotNet Client -See the [internal documentation](https://docs.google.com/document/d/1afO2ugGpTIZYUeXH_0GtMxedV51ZzmsbC3-mRdoSI_o/edit#heading=h.kqd38uu4iku) + +This library is in early stages of development. +It is meant to be used with RabbitMQ 4.0. diff --git a/RabbitMQ.AMQP.Client/IClosable.cs b/RabbitMQ.AMQP.Client/IClosable.cs index 0a67165b..84c1b5b2 100644 --- a/RabbitMQ.AMQP.Client/IClosable.cs +++ b/RabbitMQ.AMQP.Client/IClosable.cs @@ -1,10 +1,12 @@ namespace RabbitMQ.AMQP.Client; -public enum Status +public enum State { - Closed, - Reconneting, + // Opening, Open, + Reconnecting, + Closing, + Closed, } public class Error @@ -15,13 +17,11 @@ public class Error public interface IClosable { - public Status Status { get; } + public State State { get; } Task CloseAsync(); - public delegate void ChangeStatusCallBack(object sender, Status from, Status to, Error? error); + public delegate void LifeCycleCallBack(object sender, State previousState, State currentState, Error? failureCause); - event ChangeStatusCallBack ChangeStatus; - - + event LifeCycleCallBack ChangeState; } \ No newline at end of file diff --git a/RabbitMQ.AMQP.Client/IEntities.cs b/RabbitMQ.AMQP.Client/IEntities.cs index e029bec0..1a48f0e7 100644 --- a/RabbitMQ.AMQP.Client/IEntities.cs +++ b/RabbitMQ.AMQP.Client/IEntities.cs @@ -4,6 +4,11 @@ public interface IEntityInfo { } + +/// +/// Generic interface for declaring entities +/// +/// public interface IEntityDeclaration where T : IEntityInfo { Task Declare(); diff --git a/RabbitMQ.AMQP.Client/IManagement.cs b/RabbitMQ.AMQP.Client/IManagement.cs index 3b2fa4b5..ad5a8f7f 100644 --- a/RabbitMQ.AMQP.Client/IManagement.cs +++ b/RabbitMQ.AMQP.Client/IManagement.cs @@ -2,7 +2,7 @@ namespace RabbitMQ.AMQP.Client; public class ModelException(string message) : Exception(message); -public class PreconditionFailException(string message) : Exception(message); +public class PreconditionFailedException(string message) : Exception(message); public interface IManagement : IClosable { diff --git a/RabbitMQ.AMQP.Client/IRecoveryConfiguration.cs b/RabbitMQ.AMQP.Client/IRecoveryConfiguration.cs index 0a88a8b0..7afeaaac 100644 --- a/RabbitMQ.AMQP.Client/IRecoveryConfiguration.cs +++ b/RabbitMQ.AMQP.Client/IRecoveryConfiguration.cs @@ -1,15 +1,63 @@ namespace RabbitMQ.AMQP.Client; + +/// +/// Interface for the recovery configuration. +/// public interface IRecoveryConfiguration { + /// + /// Define if the recovery is activated. + /// If is not activated the connection will not try to reconnect. + /// + /// + /// IRecoveryConfiguration Activated(bool activated); bool IsActivate(); - // IRecoveryConfiguration BackOffDelayPolicy(BackOffDelayPolicy backOffDelayPolicy); + /// + /// Define the backoff delay policy. + /// It is used when the connection is trying to reconnect. + /// + /// + /// + IRecoveryConfiguration BackOffDelayPolicy(IBackOffDelayPolicy backOffDelayPolicy); + /// + /// Define if the recovery of the topology is activated. + /// When Activated the connection will try to recover the topology after a reconnection. + /// It is valid only if the recovery is activated. + /// + /// + /// IRecoveryConfiguration Topology(bool activated); bool IsTopologyActive(); +} + +/// +/// Interface for the backoff delay policy. +/// Used during the recovery of the connection. +/// +public interface IBackOffDelayPolicy +{ + /// + /// Get the next delay in milliseconds. + /// + /// + int Delay(); + + /// + /// Reset the backoff delay policy. + /// + void Reset(); + + /// + /// Define if the backoff delay policy is active. + /// Can be used to disable the backoff delay policy after a certain number of retries. + /// or when the user wants to disable the backoff delay policy. + /// + bool IsActive { get; } } \ No newline at end of file diff --git a/RabbitMQ.AMQP.Client/ITopologyListener.cs b/RabbitMQ.AMQP.Client/ITopologyListener.cs index 82586414..0cf94f0e 100644 --- a/RabbitMQ.AMQP.Client/ITopologyListener.cs +++ b/RabbitMQ.AMQP.Client/ITopologyListener.cs @@ -5,4 +5,8 @@ public interface ITopologyListener void QueueDeclared(IQueueSpecification specification); void QueueDeleted(string name); + + void Clear(); + + int QueueCount(); } \ No newline at end of file diff --git a/RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs b/RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs index 9071a7ee..66662c91 100644 --- a/RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs +++ b/RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs @@ -8,32 +8,39 @@ internal class Visitor(AmqpManagement management) : IVisitor { private AmqpManagement Management { get; } = management; - - public void VisitQueues(List queueSpec) + public async Task VisitQueues(List queueSpec) { foreach (var spec in queueSpec) { Trace.WriteLine(TraceLevel.Information, $"Recovering queue {spec.Name}"); - Management.Queue(spec).Declare(); + await Management.Queue(spec).Declare(); } } } - /// /// AmqpConnection is the concrete implementation of /// It is a wrapper around the AMQP.Net Lite class /// public class AmqpConnection : IConnection { + private const string ConnectionNotRecoveredCode = "CONNECTION_NOT_RECOVERED"; + private const string ConnectionNotRecoveredMessage = "Connection not recovered"; + // The native AMQP.Net Lite connection private Connection? _nativeConnection; private readonly AmqpManagement _management = new(); + + private readonly RecordingTopologyListener _recordingTopologyListener = new(); + private readonly ConnectionSettings _connectionSettings; /// /// Creates a new instance of + /// Through the Connection is possible to create: + /// - Management. See + /// - Publishers and Consumers: TODO: Implement /// /// /// @@ -41,6 +48,7 @@ public static async Task CreateAsync(ConnectionSettings connecti { var connection = new AmqpConnection(connectionSettings); await connection.EnsureConnectionAsync(); + return connection; } @@ -86,75 +94,116 @@ [new Symbol("connection_name")] = _connectionSettings.ConnectionName(), _nativeConnection.AddClosedCallback(MaybeRecoverConnection()); } - OnNewStatus(Status.Open, null); + OnNewStatus(State.Open, null); } catch (AmqpException e) { - throw new ConnectionException("AmqpException: Connection failed", e); + throw new ConnectionException($"AmqpException: Connection failed. Info: {ToString()} ", e); } catch (OperationCanceledException e) { // wrong virtual host - throw new ConnectionException("OperationCanceledException: Connection failed", e); + throw new ConnectionException($"OperationCanceledException: Connection failed. Info: {ToString()}", e); } catch (NotSupportedException e) { // wrong schema - throw new ConnectionException("NotSupportedException: Connection failed", e); + throw new ConnectionException($"NotSupportedException: Connection failed. Info: {ToString()}", e); } } - - private void OnNewStatus(Status newStatus, Error? error) + + private void OnNewStatus(State newState, Error? error) { - if (Status == newStatus) return; - var oldStatus = Status; - Status = newStatus; - ChangeStatus?.Invoke(this, oldStatus, newStatus, error); + if (State == newState) return; + var oldStatus = State; + State = newState; + ChangeState?.Invoke(this, oldStatus, newState, error); } private ClosedCallback MaybeRecoverConnection() { - return (sender, error) => + return async (sender, error) => { if (error != null) { - // TODO: Implement Dump Interface - Trace.WriteLine(TraceLevel.Warning, $"connection is closed unexpected" + - $"{sender} {error} {Status} " + - $"{_nativeConnection!.IsClosed}"); + Trace.WriteLine(TraceLevel.Warning, $"connection is closed unexpectedly. " + + $"Info: {ToString()}"); if (!_connectionSettings.RecoveryConfiguration.IsActivate()) { - OnNewStatus(Status.Closed, Utils.ConvertError(error)); + OnNewStatus(State.Closed, Utils.ConvertError(error)); return; } + // TODO: Block the publishers and consumers + OnNewStatus(State.Reconnecting, Utils.ConvertError(error)); - OnNewStatus(Status.Reconneting, Utils.ConvertError(error)); - - Thread.Sleep(1000); - // TODO: Replace with Backoff pattern - var t = Task.Run(async () => + await Task.Run(async () => { - Trace.WriteLine(TraceLevel.Information, "Recovering connection"); - await EnsureConnectionAsync(); - Trace.WriteLine(TraceLevel.Information, "Recovering topology"); + var connected = false; + // as first step we try to recover the connection + // so the connected flag is false + while (!connected && + // we have to check if the recovery is active. + // The user may want to disable the recovery mechanism + // the user can use the lifecycle callback to handle the error + _connectionSettings.RecoveryConfiguration.IsActivate() && + // we have to check if the backoff policy is active + // the user may want to disable the backoff policy or + // the backoff policy is not active due of some condition + // for example: Reaching the maximum number of retries and avoid the forever loop + _connectionSettings.RecoveryConfiguration.GetBackOffDelayPolicy().IsActive) + { + try + { + var next = _connectionSettings.RecoveryConfiguration.GetBackOffDelayPolicy().Delay(); + Trace.WriteLine(TraceLevel.Information, + $"Trying Recovering connection in {next} milliseconds. Info: {ToString()})"); + await Task.Delay( + TimeSpan.FromMilliseconds(next)); + + await EnsureConnectionAsync(); + connected = true; + } + catch (Exception e) + { + Trace.WriteLine(TraceLevel.Warning, + $"Error trying to recover connection {e}. Info: {this}"); + } + } + + _connectionSettings.RecoveryConfiguration.GetBackOffDelayPolicy().Reset(); + var connectionDescription = connected ? "recovered" : "not recovered"; + Trace.WriteLine(TraceLevel.Information, + $"Connection {connectionDescription}. Info: {ToString()}"); + + if (!connected) + { + Trace.WriteLine(TraceLevel.Verbose, $"connection is closed. Info: {ToString()}"); + OnNewStatus(State.Closed, new Error() + { + Description = + $"{ConnectionNotRecoveredMessage}, recover status: {_connectionSettings.RecoveryConfiguration}", + ErrorCode = ConnectionNotRecoveredCode + }); + return; + } + + if (_connectionSettings.RecoveryConfiguration.IsTopologyActive()) { - _recordingTopologyListener.Accept(new Visitor(_management)); + Trace.WriteLine(TraceLevel.Information, $"Recovering topology. Info: {ToString()}"); + await _recordingTopologyListener.Accept(new Visitor(_management)); } }); - t.WaitAsync(TimeSpan.FromSeconds(10)); return; } - Trace.WriteLine(TraceLevel.Verbose, $"connection is closed" + - $"{sender} {error} {Status} " + - $"{_nativeConnection!.IsClosed}"); - OnNewStatus(Status.Closed, Utils.ConvertError(error)); + Trace.WriteLine(TraceLevel.Verbose, $"connection is closed. Info: {ToString()}"); + OnNewStatus(State.Closed, Utils.ConvertError(error)); }; } @@ -166,12 +215,20 @@ private ClosedCallback MaybeRecoverConnection() public async Task CloseAsync() { - OnNewStatus(Status.Closed, null); + _recordingTopologyListener.Clear(); + if (State == State.Closed) return; + OnNewStatus(State.Closing, null); if (_nativeConnection is { IsClosed: false }) await _nativeConnection.CloseAsync(); await _management.CloseAsync(); } - public event IClosable.ChangeStatusCallBack? ChangeStatus; + public event IClosable.LifeCycleCallBack? ChangeState; + + public State State { get; private set; } = State.Closed; - public Status Status { get; private set; } = Status.Closed; + public override string ToString() + { + var info = $"AmqpConnection{{ConnectionSettings='{_connectionSettings}', Status='{State.ToString()}'}}"; + return info; + } } \ No newline at end of file diff --git a/RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs b/RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs index b4b08054..a9cd43be 100644 --- a/RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs +++ b/RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs @@ -14,6 +14,8 @@ namespace RabbitMQ.AMQP.Client.Impl; /// public class AmqpManagement : IManagement { + // The requests are stored in a dictionary with the correlationId as the key + // The correlationId is used to match the request with the response private readonly ConcurrentDictionary> _requests = new(); // private static readonly long IdSequence = 0; @@ -32,7 +34,7 @@ public class AmqpManagement : IManagement private const string ReplyTo = "$me"; - public virtual Status Status { get; protected set; } = Status.Closed; + public virtual State State { get; protected set; } = State.Closed; public IQueueSpecification Queue() @@ -71,7 +73,7 @@ public ITopologyListener TopologyListener() internal void Init(AmqpManagementParameters parameters) { - if (Status == Status.Open) + if (State == State.Open) return; _amqpConnection = parameters.Connection(); @@ -85,45 +87,47 @@ internal void Init(AmqpManagementParameters parameters) // TODO: find a better way to ensure that the sender link is open before the receiver link Thread.Sleep(500); EnsureReceiverLink(); - _ = Task.Run(async () => - { - try - { - while (_managementSession.IsClosed == false && - _amqpConnection.NativeConnection()!.IsClosed == false) - { - if (_receiverLink == null) continue; - var msg = await _receiverLink.ReceiveAsync(); - if (msg == null) - { - Trace.WriteLine(TraceLevel.Warning, "Received null message"); - continue; - } - - _receiverLink.Accept(msg); - HandleResponseMessage(msg); - msg.Dispose(); - } - } - catch (Exception e) - { - Trace.WriteLine(TraceLevel.Error, - $"Receiver link error in management session {e}. Receiver link closed: {_receiverLink?.IsClosed}"); - } - - Trace.WriteLine(TraceLevel.Information, "AMQP Management session closed"); - }); + _ = Task.Run(async () => { await ProcessResponses(); }); _managementSession.Closed += (sender, error) => { Trace.WriteLine(TraceLevel.Warning, $"Management session closed " + $"sender: {sender} error: {error} " + - $"Amqp Status:{Status} senderLink closed: {_senderLink?.IsClosed}" + + $"Amqp Status:{State} senderLink closed: {_senderLink?.IsClosed}" + $"_receiverLink closed: {_receiverLink?.IsClosed} " + $"_managementSession is closed: {_managementSession.IsClosed}" + $"native connection is closed: {_amqpConnection.NativeConnection()!.IsClosed}"); - OnNewStatus(Status.Closed, Utils.ConvertError(error)); + OnNewStatus(State.Closed, Utils.ConvertError(error)); }; - OnNewStatus(Status.Open, null); + OnNewStatus(State.Open, null); + } + + private async Task ProcessResponses() + { + try + { + while (_managementSession?.IsClosed == false && + _amqpConnection?.NativeConnection()!.IsClosed == false) + { + if (_receiverLink == null) continue; + var msg = await _receiverLink.ReceiveAsync(); + if (msg == null) + { + Trace.WriteLine(TraceLevel.Warning, "Received null message"); + continue; + } + + _receiverLink.Accept(msg); + HandleResponseMessage(msg); + msg.Dispose(); + } + } + catch (Exception e) + { + Trace.WriteLine(TraceLevel.Error, + $"Receiver link error in management session {e}. Receiver link closed: {_receiverLink?.IsClosed}"); + } + + Trace.WriteLine(TraceLevel.Information, "AMQP Management session closed"); } @@ -157,11 +161,11 @@ private void EnsureReceiverLink() } } - private void OnNewStatus(Status newStatus, Error? error) + private void OnNewStatus(State newState, Error? error) { - var oldStatus = Status; - Status = newStatus; - ChangeStatus?.Invoke(this, oldStatus, Status, error); + var oldStatus = State; + State = newState; + ChangeState?.Invoke(this, oldStatus, State, error); } private void EnsureSenderLink() @@ -207,12 +211,12 @@ protected void HandleResponseMessage(Message msg) { if (mre.TrySetResult(msg)) { - Trace.WriteLine(TraceLevel.Information, $"Set result for {msg.Properties.CorrelationId}"); + Trace.WriteLine(TraceLevel.Verbose, $"Set result for: {msg.Properties.CorrelationId}"); } } else { - Trace.WriteLine(TraceLevel.Error, $"No request found for message {msg.Properties.CorrelationId}"); + Trace.WriteLine(TraceLevel.Error, $"No request found for message: {msg.Properties.CorrelationId}"); } } @@ -238,16 +242,32 @@ internal async ValueTask Request(string id, object? body, string path, return await Request(message, expectedResponseCodes, timeout); } + /// + /// Core function to send a request and wait for the response + /// The request is an AMQP message with the following properties: + /// - Properties.MessageId: Mandatory to identify the request + /// - Properties.To: The path of the request, for example "/queues/my-queue" + /// - Properties.Subject: The method of the request, for example "PUT" + /// - Properties.ReplyTo: The address where the response will be sent. Default is: "$me" + /// - Body: The body of the request. For example the QueueSpec to create a queue + /// + /// Request Message. Contains all the info to create/delete a resource + /// The response codes expected for a specific call. See Code* Constants + /// Default timeout for a request + /// A message with the Info response. For example in case of Queue creation is DefaultQueueInfo + /// Application errors, see internal async ValueTask Request(Message message, int[] expectedResponseCodes, TimeSpan? timeout = null) { - if (Status != Status.Open) + if (State != State.Open) { throw new ModelException("Management is not open"); } TaskCompletionSource mre = new(TaskCreationOptions.RunContinuationsAsynchronously); + // Add TaskCompletionSource to the dictionary _requests.TryAdd(message.Properties.MessageId, mre); - using var cts = new CancellationTokenSource(timeout ?? TimeSpan.FromSeconds(5)); + using var cts = + new CancellationTokenSource(timeout ?? TimeSpan.FromSeconds(5)); // TODO: make the timeout configurable await using (cts.Token.Register( () => { @@ -256,23 +276,39 @@ internal async ValueTask Request(Message message, int[] expectedRespons })) { await InternalSendAsync(message); + + // The response is handled in a separate thread, see ProcessResponses method in the Init method var result = await mre.Task.WaitAsync(cts.Token); + // Check the responses and throw exceptions if needed. + CheckResponse(message, expectedResponseCodes, result); return result; } } + /// + /// Check the response of a request and throw exceptions if needed + /// + /// The message sent + /// The expected response codes + /// The message received from the server + /// + /// + /// internal void CheckResponse(Message sentMessage, int[] expectedResponseCodes, Message receivedMessage) { + // Check if the response code is a number + // by protocol the response code is in the Subject property if (!int.TryParse(receivedMessage.Properties.Subject, out var responseCode)) throw new ModelException($"Response code is not a number {receivedMessage.Properties.Subject}"); switch (responseCode) { case Code409: - throw new PreconditionFailException($"Precondition Fail. Message: {receivedMessage.Body}"); + throw new PreconditionFailedException($"Precondition Fail. Message: {receivedMessage.Body}"); } + // Check if the correlationId is the same as the messageId if (sentMessage.Properties.MessageId != receivedMessage.Properties.CorrelationId) throw new ModelException( $"CorrelationId does not match, expected {sentMessage.Properties.MessageId} but got {receivedMessage.Properties.CorrelationId}"); @@ -294,14 +330,14 @@ protected virtual async Task InternalSendAsync(Message message) public async Task CloseAsync() { - Status = Status.Closed; + State = State.Closed; if (_managementSession is { IsClosed: false }) { await _managementSession.CloseAsync(); } } - public event IClosable.ChangeStatusCallBack? ChangeStatus; + public event IClosable.LifeCycleCallBack? ChangeState; } public class InvalidCodeException(string message) : Exception(message); \ No newline at end of file diff --git a/RabbitMQ.AMQP.Client/Impl/AmqpQueueSpecification.cs b/RabbitMQ.AMQP.Client/Impl/AmqpQueueSpecification.cs index aaf6667a..e5783ff7 100644 --- a/RabbitMQ.AMQP.Client/Impl/AmqpQueueSpecification.cs +++ b/RabbitMQ.AMQP.Client/Impl/AmqpQueueSpecification.cs @@ -138,7 +138,6 @@ public IQueueSpecification AutoDelete(bool autoDelete) } - public IQueueSpecification Arguments(Dictionary arguments) { foreach (var (key, value) in arguments) @@ -183,6 +182,10 @@ public async Task Declare() if (string.IsNullOrEmpty(_name) || _name.Trim() == "") { + // If the name is not set, generate a random name + // client side generated names are supported by the server + // but here we generate a name to make easier to track the queue + // and remove it later _name = Utils.GenerateQueueName(); } diff --git a/RabbitMQ.AMQP.Client/Impl/ConnectionSettings.cs b/RabbitMQ.AMQP.Client/Impl/ConnectionSettings.cs index 1e1d8a26..3eab8944 100644 --- a/RabbitMQ.AMQP.Client/Impl/ConnectionSettings.cs +++ b/RabbitMQ.AMQP.Client/Impl/ConnectionSettings.cs @@ -2,8 +2,11 @@ namespace RabbitMQ.AMQP.Client.Impl; + public class ConnectionSettingBuilder { + + // TODO: maybe add the event "LifeCycle" to the builder private string _host = "localhost"; private int _port = 5672; private string _user = "guest"; @@ -13,17 +16,17 @@ public class ConnectionSettingBuilder private string _virtualHost = "/"; private IRecoveryConfiguration _recoveryConfiguration = Impl.RecoveryConfiguration.Create(); - + private ConnectionSettingBuilder() { } - + public static ConnectionSettingBuilder Create() { return new ConnectionSettingBuilder(); } - - + + public ConnectionSettingBuilder Host(string host) { _host = host; @@ -153,10 +156,15 @@ public string ConnectionName() public override string ToString() { - return - $"Address{{host='{Address.Host}', port={Address.Port}, path='{Address.Path}', username='{Address.User}', password='{Address.Password}'}}"; + var i = + $"Address" + + $"host='{Address.Host}', " + + $"port={Address.Port}, VirtualHost='{_virtualHost}', path='{Address.Path}', " + + $"username='{Address.User}', ConnectionName='{_connectionName}'"; + return i; } + public override bool Equals(object? obj) { if (obj == null || GetType() != obj.GetType()) @@ -186,6 +194,13 @@ public override int GetHashCode() public RecoveryConfiguration RecoveryConfiguration { get; set; } = RecoveryConfiguration.Create(); } +/// +/// RecoveryConfiguration is a class that represents the configuration of the recovery of the topology. +/// It is used to configure the recovery of the topology of the server after a connection is established in case of a reconnection +/// The RecoveryConfiguration can be disabled or enabled. +/// If RecoveryConfiguration._active is disabled, the reconnect mechanism will not be activated. +/// If RecoveryConfiguration._topology is disabled, the recovery of the topology will not be activated. +/// public class RecoveryConfiguration : IRecoveryConfiguration { public static RecoveryConfiguration Create() @@ -197,9 +212,14 @@ private RecoveryConfiguration() { } + // Activate the reconnect mechanism private bool _active = true; + + // Activate the recovery of the topology private bool _topology = false; + private IBackOffDelayPolicy _backOffDelayPolicy = Impl.BackOffDelayPolicy.Create(); + public IRecoveryConfiguration Activated(bool activated) { _active = activated; @@ -211,6 +231,17 @@ public bool IsActivate() return _active; } + public IRecoveryConfiguration BackOffDelayPolicy(IBackOffDelayPolicy backOffDelayPolicy) + { + _backOffDelayPolicy = backOffDelayPolicy; + return this; + } + + public IBackOffDelayPolicy GetBackOffDelayPolicy() + { + return _backOffDelayPolicy; + } + public IRecoveryConfiguration Topology(bool activated) { @@ -222,4 +253,55 @@ public bool IsTopologyActive() { return _topology; } + + public override string ToString() + { + return + $"RecoveryConfiguration{{ Active={_active}, Topology={_topology}, BackOffDelayPolicy={_backOffDelayPolicy} }}"; + } +} + +public class BackOffDelayPolicy : IBackOffDelayPolicy +{ + public static BackOffDelayPolicy Create() + { + return new BackOffDelayPolicy(); + } + + private BackOffDelayPolicy() + { + } + + private const int StartRandomMilliseconds = 500; + private const int EndRandomMilliseconds = 1500; + + private int _attempt = 1; + private int _totalAttempt = 0; + + private void ResetAfterMaxAttempt() + { + if (_attempt > 5) + _attempt = 1; + } + + public int Delay() + { + _attempt++; + _totalAttempt++; + ResetAfterMaxAttempt(); + return Random.Shared.Next(StartRandomMilliseconds, EndRandomMilliseconds) * _attempt; + } + + public void Reset() + { + _attempt = 1; + _totalAttempt = 0; + } + + public bool IsActive => _totalAttempt < 12; + + public override string ToString() + { + return $"BackOffDelayPolicy{{ Attempt={_attempt}, TotalAttempt={_totalAttempt}, IsActive={IsActive} }}"; + } } \ No newline at end of file diff --git a/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs b/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs index 0832d2bc..132b0b06 100644 --- a/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs +++ b/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs @@ -4,9 +4,15 @@ namespace RabbitMQ.AMQP.Client.Impl; public interface IVisitor { - void VisitQueues(List queueSpec); + Task VisitQueues(List queueSpec); } +/// +/// RecordingTopologyListener is a concrete implementation of +/// It is used to record the topology of the entities declared in the AMQP server ( like queues, exchanges, etc) +/// It is used to recover the topology of the server after a connection is established in case of a reconnection +/// Each time am entity is declared or deleted, the listener will record the event +/// public class RecordingTopologyListener : ITopologyListener { private readonly ConcurrentDictionary _queueSpecifications = new(); @@ -21,14 +27,19 @@ public void QueueDeleted(string name) _queueSpecifications.TryRemove(name, out _); } + public void Clear() + { + _queueSpecifications.Clear(); + } + public int QueueCount() { return _queueSpecifications.Count; } - public void Accept(IVisitor visitor) + public async Task Accept(IVisitor visitor) { - visitor.VisitQueues(_queueSpecifications.Values.ToList()); + await visitor.VisitQueues(_queueSpecifications.Values.ToList()); } } diff --git a/Tests/ConnectionRecoverTests.cs b/Tests/ConnectionRecoverTests.cs index 2dacd75f..74b0a364 100644 --- a/Tests/ConnectionRecoverTests.cs +++ b/Tests/ConnectionRecoverTests.cs @@ -9,8 +9,42 @@ namespace Tests; using Xunit; +internal class FakeBackOffDelayPolicyDisabled : IBackOffDelayPolicy +{ + public int Delay() + { + return 1; + } + + public void Reset() + { + } + + public bool IsActive => false; +} + +internal class FakeFastBackOffDelay : IBackOffDelayPolicy +{ + public int Delay() + { + return 100; + } + + public void Reset() + { + } + + public bool IsActive => true; +} + public class ConnectionRecoverTests { + /// + /// The normal close the status should be correct and error null + /// The test records the status change when the connection is closed normally. + /// The error _must_ be null when the connection is closed normally even the recovery is activated. + /// + /// If the recovery is enabled. [Theory] [InlineData(true)] [InlineData(false)] @@ -22,28 +56,43 @@ public async void NormalCloseTheStatusShouldBeCorrectAndErrorNull(bool activeRec RecoveryConfiguration.Create().Activated(activeRecovery).Topology(false)).Build()); var completion = new TaskCompletionSource(); - var listFromStatus = new List(); - var listToStatus = new List(); + var listFromStatus = new List(); + var listToStatus = new List(); var listError = new List(); - connection.ChangeStatus += (sender, from, to, error) => + connection.ChangeState += (sender, from, to, error) => { listFromStatus.Add(from); listToStatus.Add(to); listError.Add(error); - if (to == Status.Closed) + if (to == State.Closed) completion.SetResult(); }; await connection.ConnectAsync(); - Assert.Equal(Status.Open, connection.Status); + Assert.Equal(State.Open, connection.State); await connection.CloseAsync(); - Assert.Equal(Status.Closed, connection.Status); + Assert.Equal(State.Closed, connection.State); await completion.Task.WaitAsync(TimeSpan.FromSeconds(5)); - Assert.Equal(Status.Open, listFromStatus[0]); - Assert.Equal(Status.Closed, listToStatus[0]); + Assert.Equal(State.Open, listFromStatus[0]); + Assert.Equal(State.Closing, listToStatus[0]); Assert.Null(listError[0]); + + Assert.Equal(State.Closing, listFromStatus[1]); + Assert.Equal(State.Closed, listToStatus[1]); + Assert.Null(listError[1]); } + + /// + /// The unexpected close the status should be correct and error not null. + /// The connection is closed unexpectedly using HTTP API. + /// The test validates the different status changes: + /// - From Open to Reconnecting (With error) + /// - From Reconnecting to Open + /// + /// then the connection is closed normally. so the status should be: + /// - From Open to Closed + /// [Fact] public async void UnexpectedCloseTheStatusShouldBeCorrectAndErrorNotNull() { @@ -52,35 +101,131 @@ public async void UnexpectedCloseTheStatusShouldBeCorrectAndErrorNotNull() ConnectionSettingBuilder.Create().ConnectionName(connectionName).RecoveryConfiguration( RecoveryConfiguration.Create().Activated(true).Topology(false)).Build()); var resetEvent = new ManualResetEvent(false); - var listFromStatus = new List(); - var listToStatus = new List(); + var listFromStatus = new List(); + var listToStatus = new List(); var listError = new List(); - connection.ChangeStatus += (sender, from, to, error) => + connection.ChangeState += (sender, previousState, currentState, error) => { - listFromStatus.Add(from); - listToStatus.Add(to); + listFromStatus.Add(previousState); + listToStatus.Add(currentState); listError.Add(error); - if (listError.Count >= 3) + if (listError.Count >= 4) resetEvent.Set(); }; await connection.ConnectAsync(); - Assert.Equal(Status.Open, connection.Status); - SystemUtils.WaitUntilConnectionIsKilled(connectionName); + Assert.Equal(State.Open, connection.State); + await SystemUtils.WaitUntilConnectionIsKilled(connectionName); resetEvent.WaitOne(TimeSpan.FromSeconds(5)); SystemUtils.WaitUntil(() => (listFromStatus.Count >= 2)); - Assert.Equal(Status.Open, listFromStatus[0]); - Assert.Equal(Status.Reconneting, listToStatus[0]); + Assert.Equal(State.Open, listFromStatus[0]); + Assert.Equal(State.Reconnecting, listToStatus[0]); Assert.NotNull(listError[0]); - Assert.Equal(Status.Reconneting, listFromStatus[1]); - Assert.Equal(Status.Open, listToStatus[1]); + Assert.Equal(State.Reconnecting, listFromStatus[1]); + Assert.Equal(State.Open, listToStatus[1]); Assert.Null(listError[1]); resetEvent.Reset(); resetEvent.Set(); await connection.CloseAsync(); resetEvent.WaitOne(TimeSpan.FromSeconds(5)); - Assert.Equal(Status.Open, listFromStatus[2]); - Assert.Equal(Status.Closed, listToStatus[2]); + Assert.Equal(State.Open, listFromStatus[2]); + Assert.Equal(State.Closing, listToStatus[2]); Assert.Null(listError[2]); + Assert.Equal(State.Closing, listFromStatus[3]); + Assert.Equal(State.Closed, listToStatus[3]); + Assert.Null(listError[3]); + } + + + /// + /// Test when the connection is closed unexpectedly and the recovery is enabled + /// but the backoff is not active. + /// The backoff can be disabled by the user due of some condition. + /// By default, the backoff is active it will be disabled after some failed attempts. + /// See the BackOffDelayPolicy class for more details. + /// + [Fact] + public async void OverrideTheBackOffWithBackOffDisabled() + { + var connectionName = Guid.NewGuid().ToString(); + var connection = await AmqpConnection.CreateAsync( + ConnectionSettingBuilder.Create().ConnectionName(connectionName).RecoveryConfiguration( + RecoveryConfiguration.Create().Activated(true).Topology(false).BackOffDelayPolicy( + new FakeBackOffDelayPolicyDisabled())).Build()); + var resetEvent = new ManualResetEvent(false); + var listFromStatus = new List(); + var listToStatus = new List(); + var listError = new List(); + connection.ChangeState += (sender, previousState, currentState, error) => + { + listFromStatus.Add(previousState); + listToStatus.Add(currentState); + listError.Add(error); + if (listError.Count >= 4) + resetEvent.Set(); + }; + + await connection.ConnectAsync(); + Assert.Equal(State.Open, connection.State); + await SystemUtils.WaitUntilConnectionIsKilled(connectionName); + resetEvent.WaitOne(TimeSpan.FromSeconds(5)); + SystemUtils.WaitUntil(() => (listFromStatus.Count >= 2)); + Assert.Equal(State.Open, listFromStatus[0]); + Assert.Equal(State.Reconnecting, listToStatus[0]); + Assert.NotNull(listError[0]); + Assert.Equal(State.Reconnecting, listFromStatus[1]); + Assert.Equal(State.Closed, listToStatus[1]); + Assert.NotNull(listError[1]); + Assert.Equal("CONNECTION_NOT_RECOVERED", listError[1].ErrorCode); + await connection.CloseAsync(); + Assert.Equal(State.Closed, connection.State); + } + + + /// + /// Test when the connection is closed unexpectedly and the recovery is enabled and the topology-recover can be: + /// - Enabled + /// - Disabled + /// + /// When the topology-recover is Enabled the temp queues should be recovered. + /// When the topology-recover is Disabled the temp queues should not be recovered. + /// the Queue is a temp queue with the Auto-Delete and Exclusive flag set to true. + /// + /// enable/disable topology-recover + /// the number of the events expected on the ChangeState event + [Theory] + [InlineData(true, 2)] + [InlineData(false, 1)] + public async void RecoveryTopologyShouldRecoverTheTempQueues(bool topologyRecoveryEnabled, int events) + { + var queueName = $"temp-queue-should-recover-{topologyRecoveryEnabled}"; + var connectionName = Guid.NewGuid().ToString(); + var connection = await AmqpConnection.CreateAsync( + ConnectionSettingBuilder.Create() + .RecoveryConfiguration(RecoveryConfiguration.Create() + .BackOffDelayPolicy(new FakeFastBackOffDelay()) + .Topology(topologyRecoveryEnabled)) + .ConnectionName(connectionName) + .Build()); + TaskCompletionSource completion = new(TaskCreationOptions.RunContinuationsAsynchronously); + var recoveryEvents = 0; + connection.ChangeState += (sender, from, to, error) => + { + recoveryEvents++; + if (recoveryEvents == events) + completion.SetResult(true); + }; + var management = connection.Management(); + await management.Queue().Name(queueName).AutoDelete(true).Exclusive(true).Declare(); + Assert.Equal(1, management.TopologyListener().QueueCount()); + + + await SystemUtils.WaitUntilConnectionIsKilled(connectionName); + await completion.Task.WaitAsync(TimeSpan.FromSeconds(2)); + SystemUtils.WaitUntil(() => SystemUtils.QueueExists(queueName) == topologyRecoveryEnabled); + + + await connection.CloseAsync(); + Assert.Equal(0, management.TopologyListener().QueueCount()); } } \ No newline at end of file diff --git a/Tests/ManagementTests.cs b/Tests/ManagementTests.cs index c3d182ff..48edddb3 100644 --- a/Tests/ManagementTests.cs +++ b/Tests/ManagementTests.cs @@ -30,7 +30,7 @@ public void TestHandleResponseMessage(Message msg) HandleResponseMessage(msg); } - public override Status Status { get; protected set; } = Status.Open; + public override State State { get; protected set; } = State.Open; } public class ManagementTests() @@ -51,13 +51,21 @@ await Assert.ThrowsAsync(async () => await management.CloseAsync(); } + /// + /// Test to raise a ModelException based on checking the response + /// the message _must_ respect the following rules: + /// - id and correlation id should match + /// - subject _must_ be a number + /// The test validate the following cases: + /// - subject is not a number + /// - code is not in the expected list + /// - correlation id is not the same as the message id + /// [Fact] public void RaiseModelException() { var management = new TestAmqpManagement(); - const string messageId = "my_id"; - var sent = new Message() { Properties = new Properties() @@ -124,10 +132,16 @@ public async Task RaiseManagementClosedException() var management = new TestAmqpManagement(); await Assert.ThrowsAsync(async () => await management.Request(new Message(), [200])); - Assert.Equal(Status.Closed, management.Status); + Assert.Equal(State.Closed, management.State); } + /// + /// Test to validate the queue declaration with the auto generated name. + /// The auto generated name is a client side generated. + /// The test validates all the queue types. + /// + /// queues type [Theory] [InlineData(QueueType.QUORUM)] [InlineData(QueueType.CLASSIC)] @@ -141,9 +155,13 @@ public async void DeclareQueueWithNoNameShouldGenerateClientSideName(QueueType t Assert.Contains("client.gen-", queueInfo.Name()); await management.QueueDeletion().Delete(queueInfo.Name()); await connection.CloseAsync(); - Assert.Equal(Status.Closed, management.Status); + Assert.Equal(State.Closed, management.State); } + /// + /// Validate the queue declaration. + /// The queue-info response should match the queue declaration. + /// [Theory] [InlineData(true, false, false, QueueType.QUORUM)] [InlineData(true, false, false, QueueType.CLASSIC)] @@ -170,9 +188,47 @@ public async void DeclareQueueWithQueueInfoValidation( Assert.Equal(queueInfo.Exclusive(), exclusive); await management.QueueDeletion().Delete("validate_queue_info"); await connection.CloseAsync(); - Assert.Equal(Status.Closed, management.Status); + Assert.Equal(State.Closed, management.State); } + + + [Fact] + public async void DeclareQueueWithPreconditionFailedException() + { + var connection = await AmqpConnection.CreateAsync(ConnectionSettingBuilder.Create().Build()); + await connection.ConnectAsync(); + var management = connection.Management(); + await management.Queue().Name("precondition_queue").AutoDelete(false).Declare(); + await Assert.ThrowsAsync(async () => + await management.Queue().Name("precondition_queue").AutoDelete(true).Declare()); + await management.QueueDeletion().Delete("precondition_queue"); + await connection.CloseAsync(); + } + + + [Fact] + public async void DeclareAndDeleteTwoTimesShouldNotRaiseErrors() + { + var connection = await AmqpConnection.CreateAsync(ConnectionSettingBuilder.Create().Build()); + await connection.ConnectAsync(); + var management = connection.Management(); + await management.Queue().Name("DeleteTwoTimes").AutoDelete(false).Declare(); + await management.Queue().Name("DeleteTwoTimes").AutoDelete(false).Declare(); + await management.QueueDeletion().Delete("DeleteTwoTimes"); + await management.QueueDeletion().Delete("DeleteTwoTimes"); + await connection.CloseAsync(); + } + + + + ////////////// ----------------- Topology TESTS ----------------- ////////////// + + /// + /// Validate the topology listener. + /// The listener should be able to record the queue declaration. + /// creation and deletion. + /// [Fact] public async void TopologyCountShouldFollowTheQueueDeclaration() { @@ -188,7 +244,7 @@ public async void TopologyCountShouldFollowTheQueueDeclaration() for (var i = 1; i < 7; i++) { await management.QueueDeletion().Delete($"Q_{i}"); - Assert.Equal(((RecordingTopologyListener)management.TopologyListener()).QueueCount(), 6 - i); + Assert.Equal((management.TopologyListener()).QueueCount(), 6 - i); } await connection.CloseAsync(); diff --git a/Tests/Utils.cs b/Tests/Utils.cs index 3d97f99b..f6ddb3f4 100644 --- a/Tests/Utils.cs +++ b/Tests/Utils.cs @@ -66,7 +66,7 @@ public static void WaitUntil(Func func, ushort retries = 40) } } - public static async void WaitUntilAsync(Func> func, ushort retries = 10) + public static async Task WaitUntilAsync(Func> func, ushort retries = 10) { Wait(); while (!await func()) @@ -193,11 +193,11 @@ public static async Task HttpKillConnections(string connectionName) return killed; } - public static int WaitUntilConnectionIsKilled(string connectionName) + public static async Task WaitUntilConnectionIsKilled(string connectionName) { + await WaitUntilAsync(async () => await IsConnectionOpen(connectionName) ); Wait(); - WaitUntilAsync(async () => await HttpKillConnections(connectionName) == 1); - return 1; + await WaitUntilAsync(async () => await HttpKillConnections(connectionName) == 1); } private static HttpClient CreateHttpClient() diff --git a/docs/Examples/GettingStarted/Program.cs b/docs/Examples/GettingStarted/Program.cs index d93328ad..5cead9f0 100644 --- a/docs/Examples/GettingStarted/Program.cs +++ b/docs/Examples/GettingStarted/Program.cs @@ -3,27 +3,34 @@ using Trace = Amqp.Trace; using TraceLevel = Amqp.TraceLevel; -Trace.TraceLevel = TraceLevel.Information; +Trace.TraceLevel = TraceLevel.Verbose; ConsoleTraceListener consoleListener = new(); Trace.TraceListener = (l, f, a) => - consoleListener.WriteLine(DateTime.Now.ToString("[hh:mm:ss.fff]") + " " + string.Format(f, a)); + consoleListener.WriteLine($"[{DateTime.Now}] [{l}] - {f}"); Trace.WriteLine(TraceLevel.Information, "Starting"); -var connectionName = Guid.NewGuid().ToString(); +const string connectionName = "Hello-Connection"; var connection = await AmqpConnection.CreateAsync( - ConnectionSettingBuilder.Create().ConnectionName(connectionName) - .RecoveryConfiguration(RecoveryConfiguration.Create().Activated(true).Topology(true)).Build()); + ConnectionSettingBuilder.Create(). + ConnectionName(connectionName) + .RecoveryConfiguration( + RecoveryConfiguration.Create(). + Activated(true).Topology(true) + ).Build()); Trace.WriteLine(TraceLevel.Information, "Connected"); + var management = connection.Management(); -await management.Queue($"my-first-queue").Declare(); +await management.Queue($"my-first-queue"). + AutoDelete(true).Exclusive(true).Declare(); + Trace.WriteLine(TraceLevel.Information, "Queue Created"); +Console.WriteLine("Press any key to delete the queue and close the connection."); +Console.ReadKey(); await management.QueueDeletion().Delete("my-first-queue"); Trace.WriteLine(TraceLevel.Information, "Queue Deleted"); -Console.WriteLine("Press any key to close connection"); -Console.ReadKey(); await connection.CloseAsync(); Trace.WriteLine(TraceLevel.Information, "Closed"); \ No newline at end of file