Merge branch 'main' into misc

Author: Gsantomaggio

.ci/ubuntu/gha-setup.sh

@@ -46,18 +46,18 @@ else
     readonly docker_pull_args=''
 fi
 
-set -o nounset
-
-declare -r rabbitmq_docker_name="$docker_name_prefix-rabbitmq"
-declare -r toxiproxy_docker_name="$docker_name_prefix-toxiproxy"
-
 if [[ $1 == 'stop' ]]
 then
     docker stop "$rabbitmq_docker_name"
     docker stop "$toxiproxy_docker_name"
     exit 0
 fi
 
+set -o nounset
+
+declare -r rabbitmq_docker_name="$docker_name_prefix-rabbitmq"
+declare -r toxiproxy_docker_name="$docker_name_prefix-toxiproxy"
+
 function start_toxiproxy
 {
     if [[ $run_toxiproxy == 'true' ]]

.github/workflows/build-test.yaml

@@ -26,14 +26,17 @@ jobs:
         run: dotnet format ${{ github.workspace }}/Build.csproj --no-restore --verify-no-changes
       - name: Start RabbitMQ
         id: start-rabbitmq
-        run: ${{ github.workspace }}/.ci/ubuntu/gha-setup.sh toxiproxy
+        # Note: not using toxiproxy yet
+        # run: ${{ github.workspace }}/.ci/ubuntu/gha-setup.sh toxiproxy
+        run: ${{ github.workspace }}/.ci/ubuntu/gha-setup.sh
       - name: Test
-        run: dotnet test ${{ github.workspace }}/Build.csproj --no-restore --no-build --logger "console;verbosity=detailed" /p:AltCover=true
+        run: dotnet test ${{ github.workspace }}/Build.csproj --no-restore --no-build --logger "console;verbosity=detailed" /p:AltCover=true /p:AltCoverStrongNameKey=${{github.workspace}}/rabbit.snk
       - name: Check for errors in RabbitMQ logs
         run: ${{ github.workspace}}/.ci/ubuntu/gha-log-check.sh
-      - name: Maybe collect toxiproxy logs
-        if: failure()
-        run: docker logs rabbitmq-amqp-dotnet-client-toxiproxy > ${{ github.workspace }}/.ci/ubuntu/log/toxiproxy.log
+      # Note: not using toxiproxy yet
+      # - name: Maybe collect toxiproxy logs
+      #   if: failure()
+      #   run: docker logs rabbitmq-amqp-dotnet-client-toxiproxy > ${{ github.workspace }}/.ci/ubuntu/log/toxiproxy.log
       - name: Maybe upload RabbitMQ logs
         if: failure()
         uses: actions/upload-artifact@v4

Directory.Build.targets

@@ -1,5 +1,9 @@
 <!-- vim: set ft=xml: -->
 <Project>
+  <PropertyGroup>
+    <PublicKey>00240000048000009400000006020000002400005253413100040000010001005d8c30a2f8706bf952e03a51ced8b44b6e13c8a643db9ef030032b25431ec6e67b73c570beae14d63e6bd36f248c51800525e33a3fd7749bb877cfac0ad0b6544c9b385c007efb1b283327e6284396807be6300c13bbd14fed3d4c878dc33d1d832303daa9799a7f90e79aa2e5ded80143361e4849879b19a3886544780bb3bc</PublicKey>
+    <TargetFrameworkMonikerAssemblyAttributesPath>$([System.IO.Path]::Combine('$(IntermediateOutputPath)','$(TargetFrameworkMoniker).AssemblyAttributes$(DefaultLanguageSourceExtension)'))</TargetFrameworkMonikerAssemblyAttributesPath>    
+  </PropertyGroup>
   <ItemGroup>
     <EmbeddedFiles Include="$(GeneratedAssemblyInfoFile)"/>
   </ItemGroup>

Makefile

@@ -7,7 +7,7 @@ build:
 	dotnet build $(CURDIR)/Build.csproj
 
 test: build
-	dotnet test -c Debug $(CURDIR)/Tests/Tests.csproj --no-build --logger:"console;verbosity=detailed" /p:AltCover=true
+	dotnet test -c Debug $(CURDIR)/Tests/Tests.csproj --no-build --logger:"console;verbosity=detailed" /p:AltCover=true /p:AltCoverStrongNameKey=$(CURDIR)/rabbit.snk
 
 rabbitmq-server-start-arm:
 	 ./.ci/ubuntu/gha-setup.sh start pull arm 

RabbitMQ.AMQP.Client/IConnectionSettings.cs

@@ -15,6 +15,7 @@ public interface IConnectionSettings : IEquatable<IConnectionSettings>
     string ConnectionName { get; }
     string Path { get; }
     bool UseSsl { get; }
+    uint MaxFrameSize { get; }
     SaslMechanism SaslMechanism { get; }
     ITlsSettings? TlsSettings { get; }
 
@@ -35,20 +36,25 @@ public interface ITlsSettings
     /// <summary>
     /// Supported protocols to use.
     /// </summary>
-    SslProtocols Protocols { get; }
+    SslProtocols Protocols { get; set; }
+
+    /// <summary>
+    /// Acceptable TLS/SSL errors when connecting.
+    /// </summary>
+    SslPolicyErrors AcceptablePolicyErrors { get; set; }
 
     /// <summary>
     /// Specifies whether certificate revocation should be performed during handshake.
     /// </summary>
-    bool CheckCertificateRevocation { get; }
+    bool CheckCertificateRevocation { get; set; }
 
     /// <summary>
     /// Gets or sets a certificate validation callback to validate remote certificate.
     /// </summary>
-    RemoteCertificateValidationCallback? RemoteCertificateValidationCallback { get; }
+    RemoteCertificateValidationCallback? RemoteCertificateValidationCallback { get; set; }
 
     /// <summary>
     /// Gets or sets a local certificate selection callback to select the certificate which should be used for authentication.
     /// </summary>
-    LocalCertificateSelectionCallback? LocalCertificateSelectionCallback { get; }
+    LocalCertificateSelectionCallback? LocalCertificateSelectionCallback { get; set; }
 }

RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs

@@ -7,29 +7,6 @@
 
 namespace RabbitMQ.AMQP.Client.Impl;
 
-internal class Visitor(AmqpManagement management) : IVisitor
-{
-    private AmqpManagement Management { get; } = management;
-
-    public async Task VisitQueues(IEnumerable<QueueSpec> queueSpec)
-    {
-        foreach (QueueSpec spec in queueSpec)
-        {
-            Trace.WriteLine(TraceLevel.Information, $"Recovering queue {spec.Name}");
-            try
-            {
-                await Management.Queue(spec).Declare()
-                    .ConfigureAwait(false);
-            }
-            catch (Exception e)
-            {
-                Trace.WriteLine(TraceLevel.Error,
-                    $"Error recovering queue {spec.Name}. Error: {e}. Management Status: {Management}");
-            }
-        }
-    }
-}
-
 /// <summary>
 /// AmqpConnection is the concrete implementation of <see cref="IConnection"/>
 /// It is a wrapper around the AMQP.Net Lite <see cref="Connection"/> class
@@ -38,7 +15,7 @@ public class AmqpConnection : AbstractLifeCycle, IConnection
 {
     private const string ConnectionNotRecoveredCode = "CONNECTION_NOT_RECOVERED";
     private const string ConnectionNotRecoveredMessage = "Connection not recovered";
-    private readonly SemaphoreSlim _semaphoreClose = new(1, 1);
+    private readonly SemaphoreSlim _semaphoreClose = new(1, 1); // TODO this needs to be Disposed
 
     // The native AMQP.Net Lite connection
     private Connection? _nativeConnection;
@@ -55,7 +32,7 @@ private void ChangeEntitiesStatus(State state, Error? error)
 
     private void ChangePublishersStatus(State state, Error? error)
     {
-        foreach (var publisher1 in Publishers.Values)
+        foreach (IPublisher publisher1 in Publishers.Values)
         {
             var publisher = (AmqpPublisher)publisher1;
             publisher.ChangeStatus(state, error);
@@ -64,7 +41,7 @@ private void ChangePublishersStatus(State state, Error? error)
 
     private void ChangeConsumersStatus(State state, Error? error)
     {
-        foreach (var consumer1 in Consumers.Values)
+        foreach (IConsumer consumer1 in Consumers.Values)
         {
             var consumer = (AmqpConsumer)consumer1;
             consumer.ChangeStatus(state, error);
@@ -79,7 +56,7 @@ private async Task ReconnectEntities()
 
     private async Task ReconnectPublishers()
     {
-        foreach (var publisher1 in Publishers.Values)
+        foreach (IPublisher publisher1 in Publishers.Values)
         {
             var publisher = (AmqpPublisher)publisher1;
             await publisher.Reconnect().ConfigureAwait(false);
@@ -88,7 +65,7 @@ private async Task ReconnectPublishers()
 
     private async Task ReconnectConsumers()
     {
-        foreach (var consumer1 in Consumers.Values)
+        foreach (IConsumer consumer1 in Consumers.Values)
         {
             var consumer = (AmqpConsumer)consumer1;
             await consumer.Reconnect().ConfigureAwait(false);
@@ -205,10 +182,19 @@ private async Task EnsureConnection()
             var open = new Open
             {
                 HostName = $"vhost:{_connectionSettings.VirtualHost}",
-                Properties = new Fields() { [new Symbol("connection_name")] = _connectionSettings.ConnectionName, }
+                Properties = new Fields()
+                {
+                    [new Symbol("connection_name")] = _connectionSettings.ConnectionName,
+                }
             };
 
-            void OnOpened(Amqp.IConnection connection, Open open1)
+            if (_connectionSettings.MaxFrameSize > uint.MinValue)
+            {
+                // Note: when set here, there is no need to set cf.AMQP.MaxFrameSize
+                open.MaxFrameSize = _connectionSettings.MaxFrameSize;
+            }
+
+            void onOpened(Amqp.IConnection connection, Open open1)
             {
                 Trace.WriteLine(TraceLevel.Verbose, $"{ToString()} is open");
                 OnNewStatus(State.Open, null);
@@ -418,7 +404,6 @@ public IPublisherBuilder PublisherBuilder()
         return publisherBuilder;
     }
 
-
     public override async Task CloseAsync()
     {
         await _semaphoreClose.WaitAsync()
@@ -458,10 +443,32 @@ await ConnectionCloseTaskCompletionSource.Task.WaitAsync(TimeSpan.FromSeconds(10
         OnNewStatus(State.Closed, null);
     }
 
-
     public override string ToString()
     {
         string info = $"AmqpConnection{{ConnectionSettings='{_connectionSettings}', Status='{State.ToString()}'}}";
         return info;
     }
 }
+
+internal class Visitor(AmqpManagement management) : IVisitor
+{
+    private AmqpManagement Management { get; } = management;
+
+    public async Task VisitQueues(IEnumerable<QueueSpec> queueSpec)
+    {
+        foreach (QueueSpec spec in queueSpec)
+        {
+            Trace.WriteLine(TraceLevel.Information, $"Recovering queue {spec.Name}");
+            try
+            {
+                await Management.Queue(spec).Declare()
+                    .ConfigureAwait(false);
+            }
+            catch (Exception e)
+            {
+                Trace.WriteLine(TraceLevel.Error,
+                    $"Error recovering queue {spec.Name}. Error: {e}. Management Status: {Management}");
+            }
+        }
+    }
+}

RabbitMQ.AMQP.Client/Impl/ConnectionSettings.cs

@@ -15,6 +15,7 @@ public class ConnectionSettingBuilder
     private string _scheme = "AMQP";
     private string _connectionName = "AMQP.NET";
     private string _virtualHost = "/";
+    private uint _maxFrameSize = Consts.DefaultMaxFrameSize;
     private SaslMechanism _saslMechanism = Client.SaslMechanism.Plain;
     private IRecoveryConfiguration _recoveryConfiguration = Impl.RecoveryConfiguration.Create();
 
@@ -69,6 +70,17 @@ public ConnectionSettingBuilder VirtualHost(string virtualHost)
         return this;
     }
 
+    public ConnectionSettingBuilder MaxFrameSize(uint maxFrameSize)
+    {
+        _maxFrameSize = maxFrameSize;
+        if (_maxFrameSize != uint.MinValue && _maxFrameSize < 512)
+        {
+            throw new ArgumentOutOfRangeException(nameof(maxFrameSize),
+                "maxFrameSize must be greater or equal to 512");
+        }
+        return this;
+    }
+
     public ConnectionSettingBuilder SaslMechanism(SaslMechanism saslMechanism)
     {
         _saslMechanism = saslMechanism;
@@ -89,9 +101,9 @@ public ConnectionSettingBuilder RecoveryConfiguration(IRecoveryConfiguration rec
 
     public ConnectionSettings Build()
     {
-        var c = new ConnectionSettings(_host, _port, _user,
+        var c = new ConnectionSettings(_scheme, _host, _port, _user,
             _password, _virtualHost,
-            _scheme, _connectionName, _saslMechanism)
+            _connectionName, _saslMechanism, _maxFrameSize)
         {
             Recovery = (RecoveryConfiguration)_recoveryConfiguration
         };
@@ -106,8 +118,9 @@ public ConnectionSettings Build()
 public class ConnectionSettings : IConnectionSettings
 {
     private readonly Address _address;
-    private readonly string _connectionName = "";
     private readonly string _virtualHost = "/";
+    private readonly string _connectionName = "";
+    private readonly uint _maxFrameSize = Consts.DefaultMaxFrameSize;
     private readonly ITlsSettings? _tlsSettings;
     private readonly SaslMechanism _saslMechanism = SaslMechanism.Plain;
 
@@ -122,17 +135,27 @@ public ConnectionSettings(string address, ITlsSettings? tlsSettings = null)
         }
     }
 
-    public ConnectionSettings(string host, int port,
+    public ConnectionSettings(string scheme, string host, int port,
         string? user, string? password,
-        string virtualHost, string scheme, string connectionName,
-        SaslMechanism saslMechanism, ITlsSettings? tlsSettings = null)
+        string virtualHost, string connectionName,
+        SaslMechanism saslMechanism,
+        uint maxFrameSize = Consts.DefaultMaxFrameSize,
+        ITlsSettings? tlsSettings = null)
     {
         _address = new Address(host: host, port: port,
             user: user, password: password,
             path: "/", scheme: scheme);
         _connectionName = connectionName;
         _virtualHost = virtualHost;
         _saslMechanism = saslMechanism;
+
+        _maxFrameSize = maxFrameSize;
+        if (_maxFrameSize != uint.MinValue && _maxFrameSize < 512)
+        {
+            throw new ArgumentOutOfRangeException(nameof(maxFrameSize),
+                "maxFrameSize must be greater or equal to 512");
+        }
+
         _tlsSettings = tlsSettings;
 
         if (_address.UseSsl && _tlsSettings == null)
@@ -150,8 +173,8 @@ public ConnectionSettings(string host, int port,
     public string ConnectionName => _connectionName;
     public string Path => _address.Path;
     public bool UseSsl => _address.UseSsl;
+    public uint MaxFrameSize => _maxFrameSize;
     public SaslMechanism SaslMechanism => _saslMechanism;
-
     public ITlsSettings? TlsSettings => _tlsSettings;
     public IRecoveryConfiguration Recovery { get; init; } = RecoveryConfiguration.Create();
 
@@ -358,40 +381,34 @@ public override string ToString()
 public class TlsSettings : ITlsSettings
 {
     internal const SslProtocols DefaultSslProtocols = SslProtocols.None;
-
-    private readonly SslProtocols _protocols;
-    private readonly X509CertificateCollection _clientCertificates;
-    private readonly bool _checkCertificateRevocation = false;
-    private readonly RemoteCertificateValidationCallback? _remoteCertificateValidationCallback;
-    private readonly LocalCertificateSelectionCallback? _localCertificateSelectionCallback;
+    private readonly X509CertificateCollection _clientCertificates = new X509CertificateCollection();
 
     public TlsSettings() : this(DefaultSslProtocols)
     {
     }
 
     public TlsSettings(SslProtocols protocols)
     {
-        _protocols = protocols;
-        _clientCertificates = new X509CertificateCollection();
-        _remoteCertificateValidationCallback = trustEverythingCertValidationCallback;
-        _localCertificateSelectionCallback = null;
+        Protocols = protocols;
+        RemoteCertificateValidationCallback = trustEverythingCertValidationCallback;
+        LocalCertificateSelectionCallback = null;
     }
 
-    public SslProtocols Protocols => _protocols;
+    public SslProtocols Protocols { get; set; }
+
+    public SslPolicyErrors AcceptablePolicyErrors { get; set; } = SslPolicyErrors.None;
 
     public X509CertificateCollection ClientCertificates => _clientCertificates;
 
-    public bool CheckCertificateRevocation => _checkCertificateRevocation;
+    public bool CheckCertificateRevocation { get; set; } = false;
 
-    public RemoteCertificateValidationCallback? RemoteCertificateValidationCallback
-        => _remoteCertificateValidationCallback;
+    public RemoteCertificateValidationCallback? RemoteCertificateValidationCallback { get; set; }
 
-    public LocalCertificateSelectionCallback? LocalCertificateSelectionCallback
-        => _localCertificateSelectionCallback;
+    public LocalCertificateSelectionCallback? LocalCertificateSelectionCallback { get; set; }
 
-    private static bool trustEverythingCertValidationCallback(object sender, X509Certificate? certificate,
+    private bool trustEverythingCertValidationCallback(object sender, X509Certificate? certificate,
         X509Chain? chain, SslPolicyErrors sslPolicyErrors)
     {
-        return true;
+        return (sslPolicyErrors & ~AcceptablePolicyErrors) == SslPolicyErrors.None;
     }
 }

RabbitMQ.AMQP.Client/Impl/Consts.cs

@@ -7,4 +7,8 @@ public class Consts
     public const string Queues = "queues";
     public const string Bindings = "bindings";
 
+    /// <summary>
+    /// <code>uint.MinValue</code> means "no limit"
+    /// </summary>
+    public const uint DefaultMaxFrameSize = uint.MinValue; // NOTE: Azure/amqpnetlite uses 256 * 1024
 }