From 76e336c0fe2c7c702e269fca7c8fda157ca93b84 Mon Sep 17 00:00:00 2001 From: Luke Bakken Date: Mon, 16 Dec 2024 15:47:59 -0800 Subject: [PATCH] `IManagement` XML docs * Also add docs to implementing classes. * Re-order methods in classes to be public, protected, internal, private. * Add XML docs to management exceptions. --- RabbitMQ.AMQP.Client/IManagement.cs | 63 ++- RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs | 470 ++++++++++--------- RabbitMQ.AMQP.Client/ManagementExceptions.cs | 64 +++ RabbitMQ.AMQP.Client/PublicAPI.Unshipped.txt | 4 +- Tests/Management/MockManagementTests.cs | 4 +- 5 files changed, 342 insertions(+), 263 deletions(-) create mode 100644 RabbitMQ.AMQP.Client/ManagementExceptions.cs diff --git a/RabbitMQ.AMQP.Client/IManagement.cs b/RabbitMQ.AMQP.Client/IManagement.cs index f036434..38cd6ec 100644 --- a/RabbitMQ.AMQP.Client/IManagement.cs +++ b/RabbitMQ.AMQP.Client/IManagement.cs @@ -2,51 +2,64 @@ // and the Mozilla Public License, version 2.0. // Copyright (c) 2017-2024 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. -using System; using System.Threading; using System.Threading.Tasks; namespace RabbitMQ.AMQP.Client { - public class ModelException : Exception - { - public ModelException(string message) : base(message) - { - } - } - - public class PreconditionFailedException : Exception - { - public PreconditionFailedException(string message) : base(message) - { - } - } - - public class BadRequestException : Exception - { - public BadRequestException(string message) : base(message) - { - } - } - /// - /// IManagement interface and is responsible for managing the AMQP resources. - /// RabbitMQ uses AMQP end point: "/management" to manage the resources like queues, exchanges, and bindings. - /// The management endpoint works like an HTTP RPC endpoint where the client sends a request to the server. + /// The interface is used to manage the + /// AMQP 0.9.1 model + /// topology (exchanges, queues, and bindings). /// public interface IManagement : ILifeCycle { + /// + /// Create an , with an auto-generated name. + /// + /// A builder for IQueueSpecification Queue(); + + /// + /// Create an , with the given name. + /// + /// A builder for IQueueSpecification Queue(string name); + /// + /// Get the for the given queue specification. + /// + /// The + /// The + /// The for the given spec. Task GetQueueInfoAsync(IQueueSpecification queueSpec, CancellationToken cancellationToken = default); + + /// + /// Get the for the given queue name. + /// + /// The queue name + /// The + /// The for the given spec. Task GetQueueInfoAsync(string queueName, CancellationToken cancellationToken = default); + /// + /// Create an , with an auto-generated name. + /// + /// A builder for IExchangeSpecification Exchange(); + + /// + /// Create an , with the given name. + /// + /// A builder for IExchangeSpecification Exchange(string name); + /// + /// Create an . + /// + /// A builder for IBindingSpecification Binding(); } diff --git a/RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs b/RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs index b7bb4e5..43d5f84 100644 --- a/RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs +++ b/RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs @@ -16,8 +16,9 @@ namespace RabbitMQ.AMQP.Client.Impl { /// - /// AmqpManagement implements the IManagement - /// See for more information + /// AmqpManagement implements the interface + /// to manage the AMQP 0.9.1 model + /// topology (exchanges, queues, and bindings). /// public class AmqpManagement : AbstractLifeCycle, IManagement, IManagementTopology { @@ -55,10 +56,9 @@ internal AmqpManagement(AmqpManagementParameters amqpManagementParameters) } /// - /// Create a new queue specification - /// See for more information + /// Create an , with an auto-generated name. /// - /// A builder for IQueueSpecification + /// A builder for public IQueueSpecification Queue() { ThrowIfClosed(); @@ -66,20 +66,20 @@ public IQueueSpecification Queue() } /// - /// Create a new queue specification with the given name - /// See for more information + /// Create an , with the given name. /// - /// A builder for IQueueSpecification + /// A builder for public IQueueSpecification Queue(string name) { return Queue().Name(name); } /// - /// Get the queue info for the given queue specification - /// See for more information + /// Get the for the given queue specification /// - /// Queue Information + /// The + /// The + /// The for the given spec. public Task GetQueueInfoAsync(IQueueSpecification queueSpec, CancellationToken cancellationToken = default) { @@ -87,10 +87,11 @@ public Task GetQueueInfoAsync(IQueueSpecification queueSpec, } /// - /// Get the queue info for the given queue name - /// See for more information + /// Get the for the given queue name. /// - /// Queue Information + /// The queue name + /// The + /// The for the given spec. public async Task GetQueueInfoAsync(string queueName, CancellationToken cancellationToken = default) { @@ -105,19 +106,10 @@ public async Task GetQueueInfoAsync(string queueName, return new DefaultQueueInfo((Map)response.Body); } - internal IQueueSpecification Queue(QueueSpec spec) - { - return Queue().Name(spec.QueueName) - .AutoDelete(spec.IsAutoDelete) - .Exclusive(spec.IsExclusive) - .Arguments(spec.QueueArguments); - } - /// - /// Create a new AMQPExchange specification - /// See for more information + /// Create an , with an auto-generated name. /// - /// A builder for IExchangeSpecification + /// A builder for public IExchangeSpecification Exchange() { ThrowIfClosed(); @@ -125,46 +117,25 @@ public IExchangeSpecification Exchange() } /// - /// - /// Create a new AMQPExchange specification with the given name - /// See for more information + /// Create an , with the given name. /// - /// A builder for IExchangeSpecification + /// A builder for public IExchangeSpecification Exchange(string name) { return Exchange().Name(name); } - internal IExchangeSpecification Exchange(ExchangeSpec spec) - { - return Exchange().Name(spec.ExchangeName) - .AutoDelete(spec.IsAutoDelete) - .Type(spec.ExchangeType) - .Arguments(spec.ExchangeArguments); - } - + /// + /// Create an . + /// + /// A builder for public IBindingSpecification Binding() { return new AmqpBindingSpecification(this); } - internal IBindingSpecification Binding(BindingSpec spec) - { - return Binding() - .SourceExchange(spec.SourceExchangeName) - .DestinationQueue(spec.DestinationQueueName) - .DestinationExchange(spec.DestinationExchangeName) - .Key(spec.BindingKey) - .Arguments(spec.BindingArguments); - } - - ITopologyListener IManagementTopology.TopologyListener() - { - return _amqpManagementParameters.TopologyListener(); - } - /// - /// Open the management session to RabbitMQ + /// Open the management session. /// public override async Task OpenAsync() { @@ -193,169 +164,67 @@ await base.OpenAsync() .ConfigureAwait(false); } - private void OnManagementSessionClosed(IAmqpObject sender, Amqp.Framing.Error error) + /// + /// Close the management session. + /// + public override async Task CloseAsync() { - if (State != State.Closed && error != null) - { - Trace.WriteLine(TraceLevel.Warning, $"Management session closed " + - $"with error: {Utils.ConvertError(error)} " + - $" AmqpManagement: {ToString()}"); - } + // TODO 10 seconds seems too long + TimeSpan closeSpan = TimeSpan.FromSeconds(10); - OnNewStatus(State.Closed, Utils.ConvertError(error)); + if (_managementSession is { IsClosed: false }) + { + OnNewStatus(State.Closing, null); - // Note: TrySetResult *must* be used here - _managementSessionClosedTcs.TrySetResult(true); - } + await _managementSession.CloseAsync(closeSpan) + .ConfigureAwait(false); - private async Task ProcessResponses() - { - try - { - while (_managementSession?.IsClosed == false && - _amqpManagementParameters.IsNativeConnectionClosed == false) - { - if (_receiverLink == null) - { - continue; - } + await _managementSessionClosedTcs.Task.WaitAsync(closeSpan) + .ConfigureAwait(false); - TimeSpan timeout = TimeSpan.FromSeconds(59); - using (Message msg = await _receiverLink.ReceiveAsync(timeout).ConfigureAwait(false)) - { - if (msg == null) - { - // this is not a problem, it is just a timeout. - // the timeout is set to 60 seconds. - // For the moment I'd trace it at some point we can remove it - Trace.WriteLine(TraceLevel.Verbose, - $"{ToString()} - Timeout {timeout.Seconds} s.. waiting for message."); - continue; - } + _managementSession = null; + _senderLink = null; + _receiverLink = null; - _receiverLink.Accept(msg); - HandleResponseMessage(msg); - } - } - } - catch (Exception e) - { - // TODO this is a serious situation that should be thrown - // up to the client application - if (_receiverLink?.IsClosed == false) - { - Trace.WriteLine(TraceLevel.Error, - $"Receiver link error in management session {e}. Receiver link closed: {_receiverLink?.IsClosed}"); - } + // this is actually a double set of the status, but it is needed to ensure that the status is set to closed + // but the `OnNewStatus` is idempotent + OnNewStatus(State.Closed, null); } - - Trace.WriteLine(TraceLevel.Verbose, "ProcessResponses Task closed"); } - private async Task EnsureReceiverLinkAsync() + /// + /// Convert this instance to a string. + /// + /// The string representation of this instance. + public override string ToString() { - if (_receiverLink == null || _receiverLink.IsClosed) - { - var receiveAttach = new Attach() - { - SndSettleMode = SenderSettleMode.Settled, - RcvSettleMode = ReceiverSettleMode.First, - Properties = new Fields { { new Symbol("paired"), true } }, - LinkName = LinkPairName, - Source = - new Source() { Address = ManagementNodeAddress, ExpiryPolicy = new Symbol("LINK_DETACH"), }, - Handle = 1, - Target = - new Target() { Address = ManagementNodeAddress, ExpiryPolicy = new Symbol("SESSION_END"), }, - }; - - TaskCompletionSource tcs = Utils.CreateTaskCompletionSource(); - var tmpReceiverLink = new ReceiverLink( - _managementSession, LinkPairName, receiveAttach, (ILink link, Attach attach) => - { - if (link is ReceiverLink receiverLink) - { - tcs.SetResult(receiverLink); - } - else - { - // TODO create "internal bug" exception type? - var ex = new InvalidOperationException( - "invalid link in OnAttached, report via https://github.com/rabbitmq/rabbitmq-amqp-dotnet-client/issues"); - tcs.SetException(ex); - } - }); - - _receiverLink = await tcs.Task - .ConfigureAwait(false); + string info = $"AmqpManagement{{" + + $"AmqpConnection='{_amqpManagementParameters.Connection}', " + + $"Status='{State.ToString()}'" + + $"ReceiverLink closed: {_receiverLink?.IsClosed} " + + $"}}"; - if (false == Object.ReferenceEquals(_receiverLink, tmpReceiverLink)) - { - // TODO log this case? - } + return info; + } - // TODO - // using a credit of 1 can result in AmqpExceptions in ProcessResponses - _receiverLink.SetCredit(100); - } + ITopologyListener IManagementTopology.TopologyListener() + { + return _amqpManagementParameters.TopologyListener(); } - private async Task EnsureSenderLinkAsync() + internal protected virtual Task InternalSendAsync(Message message, TimeSpan timeout) { - if (_senderLink == null || _senderLink.IsClosed) + if (_senderLink is null) { - var senderAttach = new Attach - { - SndSettleMode = SenderSettleMode.Settled, - RcvSettleMode = ReceiverSettleMode.First, - Properties = new Fields { { new Symbol("paired"), true } }, - LinkName = LinkPairName, - Source = new Source() - { - Address = ManagementNodeAddress, - ExpiryPolicy = new Symbol("LINK_DETACH"), - Timeout = 0, - Dynamic = false, - Durable = 0 - }, - Handle = 0, - Target = new Target() - { - Address = ManagementNodeAddress, - ExpiryPolicy = new Symbol("SESSION_END"), - Timeout = 0, - Dynamic = false, - }, - }; - - TaskCompletionSource tcs = Utils.CreateTaskCompletionSource(); - var tmpSenderLink = new SenderLink( - _managementSession, LinkPairName, senderAttach, (ILink link, Attach attach) => - { - if (link is SenderLink senderLink) - { - tcs.SetResult(senderLink); - } - else - { - // TODO create "internal bug" exception type? - var ex = new InvalidOperationException( - "invalid link in OnAttached, report via https://github.com/rabbitmq/rabbitmq-amqp-dotnet-client/issues"); - tcs.SetException(ex); - } - }); - - _senderLink = await tcs.Task - .ConfigureAwait(false); - - if (false == Object.ReferenceEquals(_senderLink, tmpSenderLink)) - { - // TODO log this case? - } + // TODO create "internal bug" exception type? + throw new InvalidOperationException( + "_senderLink is null, report via https://github.com/rabbitmq/rabbitmq-amqp-dotnet-client/issues"); } + + return _senderLink.SendAsync(message, timeout); } - protected void HandleResponseMessage(Message msg) + internal protected void HandleResponseMessage(Message msg) { if (msg.Properties.CorrelationId != null && _requests.TryRemove(msg.Properties.CorrelationId, out TaskCompletionSource? mre)) @@ -371,6 +240,32 @@ protected void HandleResponseMessage(Message msg) } } + internal IQueueSpecification Queue(QueueSpec spec) + { + return Queue().Name(spec.QueueName) + .AutoDelete(spec.IsAutoDelete) + .Exclusive(spec.IsExclusive) + .Arguments(spec.QueueArguments); + } + + internal IExchangeSpecification Exchange(ExchangeSpec spec) + { + return Exchange().Name(spec.ExchangeName) + .AutoDelete(spec.IsAutoDelete) + .Type(spec.ExchangeType) + .Arguments(spec.ExchangeArguments); + } + + internal IBindingSpecification Binding(BindingSpec spec) + { + return Binding() + .SourceExchange(spec.SourceExchangeName) + .DestinationQueue(spec.DestinationQueueName) + .DestinationExchange(spec.DestinationExchangeName) + .Key(spec.BindingKey) + .Arguments(spec.BindingArguments); + } + internal Task RequestAsync(string path, string method, int[] expectedResponseCodes, TimeSpan? timeout = null, @@ -473,6 +368,11 @@ await sendTask.WaitAsync(linkedCts.Token) return result; } + internal void ChangeStatus(State newState, Error? error) + { + OnNewStatus(newState, error); + } + /// /// Check the response of a request and throw exceptions if needed /// @@ -515,64 +415,166 @@ internal void CheckResponse(Message sentMessage, int[] expectedResponseCodes, Me } } - protected virtual Task InternalSendAsync(Message message, TimeSpan timeout) + private void OnManagementSessionClosed(IAmqpObject sender, Amqp.Framing.Error error) { - if (_senderLink is null) + if (State != State.Closed && error != null) { - // TODO create "internal bug" exception type? - throw new InvalidOperationException( - "_senderLink is null, report via https://github.com/rabbitmq/rabbitmq-amqp-dotnet-client/issues"); + Trace.WriteLine(TraceLevel.Warning, $"Management session closed " + + $"with error: {Utils.ConvertError(error)} " + + $" AmqpManagement: {ToString()}"); } - return _senderLink.SendAsync(message, timeout); + OnNewStatus(State.Closed, Utils.ConvertError(error)); + + // Note: TrySetResult *must* be used here + _managementSessionClosedTcs.TrySetResult(true); } - public override async Task CloseAsync() + private async Task ProcessResponses() { - // TODO 10 seconds seems too long - TimeSpan closeSpan = TimeSpan.FromSeconds(10); + try + { + while (_managementSession?.IsClosed == false && + _amqpManagementParameters.IsNativeConnectionClosed == false) + { + if (_receiverLink == null) + { + continue; + } - if (_managementSession is { IsClosed: false }) + TimeSpan timeout = TimeSpan.FromSeconds(59); + using (Message msg = await _receiverLink.ReceiveAsync(timeout).ConfigureAwait(false)) + { + if (msg == null) + { + // this is not a problem, it is just a timeout. + // the timeout is set to 60 seconds. + // For the moment I'd trace it at some point we can remove it + Trace.WriteLine(TraceLevel.Verbose, + $"{ToString()} - Timeout {timeout.Seconds} s.. waiting for message."); + continue; + } + + _receiverLink.Accept(msg); + HandleResponseMessage(msg); + } + } + } + catch (Exception e) { - OnNewStatus(State.Closing, null); + // TODO this is a serious situation that should be thrown + // up to the client application + if (_receiverLink?.IsClosed == false) + { + Trace.WriteLine(TraceLevel.Error, + $"Receiver link error in management session {e}. Receiver link closed: {_receiverLink?.IsClosed}"); + } + } - await _managementSession.CloseAsync(closeSpan) - .ConfigureAwait(false); + Trace.WriteLine(TraceLevel.Verbose, "ProcessResponses Task closed"); + } - await _managementSessionClosedTcs.Task.WaitAsync(closeSpan) + private async Task EnsureReceiverLinkAsync() + { + if (_receiverLink == null || _receiverLink.IsClosed) + { + var receiveAttach = new Attach() + { + SndSettleMode = SenderSettleMode.Settled, + RcvSettleMode = ReceiverSettleMode.First, + Properties = new Fields { { new Symbol("paired"), true } }, + LinkName = LinkPairName, + Source = + new Source() { Address = ManagementNodeAddress, ExpiryPolicy = new Symbol("LINK_DETACH"), }, + Handle = 1, + Target = + new Target() { Address = ManagementNodeAddress, ExpiryPolicy = new Symbol("SESSION_END"), }, + }; + + TaskCompletionSource tcs = Utils.CreateTaskCompletionSource(); + var tmpReceiverLink = new ReceiverLink( + _managementSession, LinkPairName, receiveAttach, (ILink link, Attach attach) => + { + if (link is ReceiverLink receiverLink) + { + tcs.SetResult(receiverLink); + } + else + { + // TODO create "internal bug" exception type? + var ex = new InvalidOperationException( + "invalid link in OnAttached, report via https://github.com/rabbitmq/rabbitmq-amqp-dotnet-client/issues"); + tcs.SetException(ex); + } + }); + + _receiverLink = await tcs.Task .ConfigureAwait(false); - _managementSession = null; - _senderLink = null; - _receiverLink = null; + if (false == Object.ReferenceEquals(_receiverLink, tmpReceiverLink)) + { + // TODO log this case? + } - // this is actually a double set of the status, but it is needed to ensure that the status is set to closed - // but the `OnNewStatus` is idempotent - OnNewStatus(State.Closed, null); + // TODO + // using a credit of 1 can result in AmqpExceptions in ProcessResponses + _receiverLink.SetCredit(100); } } - public override string ToString() + private async Task EnsureSenderLinkAsync() { - string info = $"AmqpManagement{{" + - $"AmqpConnection='{_amqpManagementParameters.Connection}', " + - $"Status='{State.ToString()}'" + - $"ReceiverLink closed: {_receiverLink?.IsClosed} " + - $"}}"; + if (_senderLink == null || _senderLink.IsClosed) + { + var senderAttach = new Attach + { + SndSettleMode = SenderSettleMode.Settled, + RcvSettleMode = ReceiverSettleMode.First, + Properties = new Fields { { new Symbol("paired"), true } }, + LinkName = LinkPairName, + Source = new Source() + { + Address = ManagementNodeAddress, + ExpiryPolicy = new Symbol("LINK_DETACH"), + Timeout = 0, + Dynamic = false, + Durable = 0 + }, + Handle = 0, + Target = new Target() + { + Address = ManagementNodeAddress, + ExpiryPolicy = new Symbol("SESSION_END"), + Timeout = 0, + Dynamic = false, + }, + }; - return info; - } + TaskCompletionSource tcs = Utils.CreateTaskCompletionSource(); + var tmpSenderLink = new SenderLink( + _managementSession, LinkPairName, senderAttach, (ILink link, Attach attach) => + { + if (link is SenderLink senderLink) + { + tcs.SetResult(senderLink); + } + else + { + // TODO create "internal bug" exception type? + var ex = new InvalidOperationException( + "invalid link in OnAttached, report via https://github.com/rabbitmq/rabbitmq-amqp-dotnet-client/issues"); + tcs.SetException(ex); + } + }); - internal void ChangeStatus(State newState, Error? error) - { - OnNewStatus(newState, error); - } - } + _senderLink = await tcs.Task + .ConfigureAwait(false); - public class InvalidCodeException : Exception - { - public InvalidCodeException(string message) : base(message) - { + if (false == Object.ReferenceEquals(_senderLink, tmpSenderLink)) + { + // TODO log this case? + } + } } } } diff --git a/RabbitMQ.AMQP.Client/ManagementExceptions.cs b/RabbitMQ.AMQP.Client/ManagementExceptions.cs new file mode 100644 index 0000000..919913e --- /dev/null +++ b/RabbitMQ.AMQP.Client/ManagementExceptions.cs @@ -0,0 +1,64 @@ +// This source code is dual-licensed under the Apache License, version 2.0, +// and the Mozilla Public License, version 2.0. +// Copyright (c) 2017-2024 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. + +using System; + +namespace RabbitMQ.AMQP.Client +{ + /// + /// Thrown when management response code or correlation ID mismatch. + /// + public class ModelException : Exception + { + /// + /// Create a new + /// + /// The exception message. + public ModelException(string message) : base(message) + { + } + } + + /// + /// Thrown when management response is "precondition failed". + /// + public class PreconditionFailedException : Exception + { + /// + /// Create a new + /// + /// The exception message. + public PreconditionFailedException(string message) : base(message) + { + } + } + + /// + /// Thrown when management response is code 400, bad request. + /// + public class BadRequestException : Exception + { + /// + /// Create a new + /// + /// The exception message. + public BadRequestException(string message) : base(message) + { + } + } + + /// + /// Thrown when management response code is not expected. + /// + public class InvalidCodeException : Exception + { + /// + /// Create a new + /// + /// The exception message. + public InvalidCodeException(string message) : base(message) + { + } + } +} diff --git a/RabbitMQ.AMQP.Client/PublicAPI.Unshipped.txt b/RabbitMQ.AMQP.Client/PublicAPI.Unshipped.txt index 6c88407..7b38b66 100644 --- a/RabbitMQ.AMQP.Client/PublicAPI.Unshipped.txt +++ b/RabbitMQ.AMQP.Client/PublicAPI.Unshipped.txt @@ -531,8 +531,6 @@ RabbitMQ.AMQP.Client.Impl.DefaultQueueInfo.Name() -> string! RabbitMQ.AMQP.Client.Impl.DefaultQueueInfo.Type() -> RabbitMQ.AMQP.Client.QueueType RabbitMQ.AMQP.Client.Impl.FieldNotSetException RabbitMQ.AMQP.Client.Impl.FieldNotSetException.FieldNotSetException() -> void -RabbitMQ.AMQP.Client.Impl.InvalidCodeException -RabbitMQ.AMQP.Client.Impl.InvalidCodeException.InvalidCodeException(string! message) -> void RabbitMQ.AMQP.Client.Impl.ListenerStreamOptions RabbitMQ.AMQP.Client.Impl.ListenerStreamOptions.ListenerStreamOptions(Amqp.Types.Map! filters, bool areFilterExpressionsSupported) -> void RabbitMQ.AMQP.Client.Impl.MessageAddressBuilder @@ -601,6 +599,8 @@ RabbitMQ.AMQP.Client.InternalBugException.InternalBugException() -> void RabbitMQ.AMQP.Client.InternalBugException.InternalBugException(string! message) -> void RabbitMQ.AMQP.Client.InvalidAddressException RabbitMQ.AMQP.Client.InvalidAddressException.InvalidAddressException(string! message) -> void +RabbitMQ.AMQP.Client.InvalidCodeException +RabbitMQ.AMQP.Client.InvalidCodeException.InvalidCodeException(string! message) -> void RabbitMQ.AMQP.Client.IPublisher RabbitMQ.AMQP.Client.IPublisher.PublishAsync(RabbitMQ.AMQP.Client.IMessage! message, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! RabbitMQ.AMQP.Client.IPublisherBuilder diff --git a/Tests/Management/MockManagementTests.cs b/Tests/Management/MockManagementTests.cs index 462d920..f1f0472 100644 --- a/Tests/Management/MockManagementTests.cs +++ b/Tests/Management/MockManagementTests.cs @@ -16,7 +16,7 @@ public class MockManagementTests() { public class TestAmqpManagement() : AmqpManagement(new AmqpManagementParameters(null!)) { - protected override async Task InternalSendAsync(Message message, TimeSpan timeout) + internal protected override async Task InternalSendAsync(Message message, TimeSpan timeout) { await Task.Delay(1000); } @@ -29,7 +29,7 @@ public TestAmqpManagementOpen() : base(new AmqpManagementParameters(null!)) State = State.Open; } - protected override async Task InternalSendAsync(Message message, TimeSpan timeout) + internal protected override async Task InternalSendAsync(Message message, TimeSpan timeout) { await Task.Delay(1000); }