* Fix reconnection due to the use of SendAsync in AmqpPublisher. So bizarre!

* Make all of `Utils` internal

Author: lukebakken

RabbitMQ.AMQP.Client/IConnectionSettings.cs

@@ -15,7 +15,7 @@ public interface IConnectionSettings : IEquatable<IConnectionSettings>
     string ContainerId { get; }
     string Path { get; }
     bool UseSsl { get; }
-    int MaxFrameSize { get; }
+    uint MaxFrameSize { get; }
     SaslMechanism SaslMechanism { get; }
     ITlsSettings? TlsSettings { get; }
     IRecoveryConfiguration Recovery { get; }

RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs

@@ -5,16 +5,17 @@
 using System.Collections.Concurrent;
 using System.Collections.ObjectModel;
 using Amqp;
-using Amqp.Handler;
+using Amqp.Framing;
 using Amqp.Sasl;
+using Amqp.Types;
 
 namespace RabbitMQ.AMQP.Client.Impl;
 
 /// <summary>
 /// AmqpConnection is the concrete implementation of <see cref="IConnection"/>
 /// It is a wrapper around the AMQP.Net Lite <see cref="Connection"/> class
 /// </summary>
-public class AmqpConnection : AbstractLifeCycle, IConnection, IHandler
+public class AmqpConnection : AbstractLifeCycle, IConnection
 {
     private const string ConnectionNotRecoveredCode = "CONNECTION_NOT_RECOVERED";
     private const string ConnectionNotRecoveredMessage = "Connection not recovered";
@@ -24,6 +25,7 @@ public class AmqpConnection : AbstractLifeCycle, IConnection, IHandler
 
     // The native AMQP.Net Lite connection
     private Connection? _nativeConnection;
+    private ClosedCallback? _closedCallback;
 
     private readonly AmqpManagement _management;
     private readonly RecordingTopologyListener _recordingTopologyListener = new();
@@ -133,90 +135,6 @@ await _connectionCloseTaskCompletionSource.Task.WaitAsync(TimeSpan.FromSeconds(1
         OnNewStatus(State.Closed, null);
     }
 
-    public bool CanHandle(EventId id)
-    {
-        Trace.WriteLine(TraceLevel.Verbose, $"EventId: {id}");
-        return id switch
-        {
-            EventId.SocketAccept => false,
-            EventId.SocketConnect => false,
-            EventId.ConnectionLocalOpen => true,
-            EventId.ConnectionRemoteOpen => false,
-            EventId.ConnectionLocalClose => false,
-            EventId.ConnectionRemoteClose => true,
-            EventId.SessionLocalOpen => false,
-            EventId.SessionRemoteOpen => false,
-            EventId.SessionLocalClose => false,
-            EventId.SessionRemoteClose => false,
-            EventId.LinkLocalOpen => false,
-            EventId.LinkRemoteOpen => false,
-            EventId.LinkLocalClose => false,
-            EventId.LinkRemoteClose => false,
-            EventId.SendDelivery => false,
-            EventId.ReceiveDelivery => false,
-            EventId.SslAuthenticate => false,
-            EventId.SslStreamAccept => false,
-            EventId.HttpAccept => false,
-            EventId.WebSocketAccept => false,
-            EventId.ConnectionAccept => false,
-            _ => false,
-        };
-    }
-
-    public void Handle(Event protocolEvent)
-    {
-        Trace.WriteLine(TraceLevel.Verbose, $"protocolEvent: {protocolEvent}");
-        switch (protocolEvent.Id)
-        {
-            case EventId.ConnectionLocalOpen:
-                Trace.WriteLine(TraceLevel.Verbose, $"{ToString()} is open");
-                OnNewStatus(State.Open, null);
-                break;
-            case EventId.ConnectionRemoteOpen:
-                break;
-            case EventId.ConnectionLocalClose:
-                break;
-            case EventId.ConnectionRemoteClose:
-                // TODO oy vey, what to do about the task?
-                _ = Task.Run(() => ReconnectAsync((Amqp.Framing.Close)protocolEvent.Context));
-                break;
-            case EventId.SessionLocalOpen:
-                break;
-            case EventId.SessionRemoteOpen:
-                break;
-            case EventId.SessionLocalClose:
-                break;
-            case EventId.SessionRemoteClose:
-                break;
-            case EventId.LinkLocalOpen:
-                break;
-            case EventId.LinkRemoteOpen:
-                break;
-            case EventId.LinkLocalClose:
-                break;
-            case EventId.LinkRemoteClose:
-                break;
-            case EventId.SendDelivery:
-                break;
-            case EventId.ReceiveDelivery:
-                break;
-            case EventId.SocketConnect:
-                break;
-            case EventId.SslAuthenticate:
-                break;
-            case EventId.SocketAccept:
-                break;
-            case EventId.SslStreamAccept:
-                break;
-            case EventId.HttpAccept:
-                break;
-            case EventId.WebSocketAccept:
-                break;
-            case EventId.ConnectionAccept:
-                break;
-        }
-    }
-
     public override string ToString()
     {
         string info = $"AmqpConnection{{ConnectionSettings='{_connectionSettings}', Status='{State.ToString()}'}}";
@@ -232,6 +150,11 @@ protected override void Dispose(bool disposing)
     {
         if (disposing)
         {
+            if (_nativeConnection is not null &&
+                _closedCallback is not null)
+            {
+                _nativeConnection.Closed -= _closedCallback;
+            }
             _semaphoreOpen.Dispose();
             _semaphoreClose.Dispose();
         }
@@ -283,33 +206,26 @@ await _semaphoreOpen.WaitAsync()
                 return;
             }
 
-            /*
             var open = new Open
             {
+                // Note: no need to set cf.AMQP.HostName
                 HostName = $"vhost:{_connectionSettings.VirtualHost}",
+                // Note: no need to set cf.AMQP.ContainerId
                 ContainerId = _connectionSettings.ContainerId,
                 Properties = new Fields()
                 {
                     [new Symbol("connection_name")] = _connectionSettings.ContainerId,
                 }
             };
 
-            if (_connectionSettings.MaxFrameSize > int.MinValue)
+            if (_connectionSettings.MaxFrameSize > uint.MinValue)
             {
                 // Note: when set here, there is no need to set cf.AMQP.MaxFrameSize
                 open.MaxFrameSize = _connectionSettings.MaxFrameSize;
             }
-            */
 
             var cf = new ConnectionFactory();
 
-            cf.AMQP.ContainerId = _connectionSettings.ContainerId;
-            cf.AMQP.HostName = $"vhost:{_connectionSettings.VirtualHost}";
-            if (_connectionSettings.MaxFrameSize > int.MinValue)
-            {
-                cf.AMQP.MaxFrameSize = _connectionSettings.MaxFrameSize;
-            }
-
             if (_connectionSettings is { UseSsl: true, TlsSettings: not null })
             {
                 cf.SSL.Protocols = _connectionSettings.TlsSettings.Protocols;
@@ -338,13 +254,11 @@ [new Symbol("connection_name")] = _connectionSettings.ContainerId,
                 cf.SASL.Profile = SaslProfile.External;
             }
 
-            /*
             void OnOpened(Amqp.IConnection connection, Open open1)
             {
                 Trace.WriteLine(TraceLevel.Verbose, $"{ToString()} is open");
                 OnNewStatus(State.Open, null);
             }
-            */
 
             try
             {
@@ -358,8 +272,8 @@ void OnOpened(Amqp.IConnection connection, Open open1)
                 {
                     connectionSettings = (ConnectionSettings)_connectionSettings;
                     Address address = connectionSettings.Address;
-                    _nativeConnection = await cf.CreateAsync(address: address, handler: this).ConfigureAwait(false);
-                    // _nativeConnection = await cf.CreateAsync(address: address, open: open, onOpened: OnOpened).ConfigureAwait(false);
+                    _nativeConnection = await cf.CreateAsync(address: address, open: open, onOpened: OnOpened)
+                        .ConfigureAwait(false);
                 }
             }
             catch (Exception ex)
@@ -377,8 +291,8 @@ void OnOpened(Amqp.IConnection connection, Open open1)
             await _management.OpenAsync()
                 .ConfigureAwait(false);
 
-            // _closedCallback = BuildClosedCallback();
-            // _nativeConnection.AddClosedCallback(_closedCallback);
+            _closedCallback = BuildClosedCallback();
+            _nativeConnection.AddClosedCallback(_closedCallback);
         }
         catch (AmqpException e)
         {
@@ -391,98 +305,6 @@ await _management.OpenAsync()
         }
     }
 
-    private void DoClose(Amqp.Framing.Error errorFrame, Error? argError = null)
-    {
-        Error? err = argError ?? Utils.ConvertError(errorFrame);
-        Trace.WriteLine(TraceLevel.Verbose, $"{ToString()} is closed");
-        OnNewStatus(State.Closed, err);
-        ChangeEntitiesStatus(State.Closed, err);
-        _connectionCloseTaskCompletionSource.SetResult(true);
-    }
-
-    private async Task ReconnectAsync(Amqp.Framing.Close closeFrame)
-    {
-        Amqp.Framing.Error errorFrame = closeFrame.Error;
-
-        // change the status for the connection and all the entities
-        // to reconnecting and all the events are fired
-        OnNewStatus(State.Reconnecting, Utils.ConvertError(errorFrame));
-        ChangeEntitiesStatus(State.Reconnecting, Utils.ConvertError(errorFrame));
-
-        IBackOffDelayPolicy backOffDelayPolicy = _connectionSettings.Recovery.GetBackOffDelayPolicy();
-        bool connected = false;
-        // as first step we try to recover the connection
-        // so the connected flag is false
-        while (false == connected &&
-               // 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
-               backOffDelayPolicy.IsActive() &&
-               // even we set the status to reconnecting up, we need to check if the connection is still in the
-               // reconnecting status. The user may close the connection in the meanwhile
-               State == State.Reconnecting)
-        {
-            try
-            {
-                int nextDelayMs = backOffDelayPolicy.Delay();
-
-                Trace.WriteLine(TraceLevel.Information,
-                    $"{ToString()} is trying Recovering connection in {nextDelayMs} milliseconds, " +
-                    $"attempt: {_connectionSettings.Recovery.GetBackOffDelayPolicy().CurrentAttempt}. ");
-
-                await Task.Delay(TimeSpan.FromMilliseconds(nextDelayMs))
-                    .ConfigureAwait(false);
-
-                await OpenConnectionAsync()
-                    .ConfigureAwait(false);
-
-                connected = true;
-            }
-            catch (Exception e)
-            {
-                // TODO this could / should be more visible to the user, perhaps?
-                Trace.WriteLine(TraceLevel.Warning,
-                    $"{ToString()} Error trying to recover connection {e}");
-            }
-        }
-
-        backOffDelayPolicy.Reset();
-        string connectionDescription = connected ? "recovered" : "not recovered";
-        Trace.WriteLine(TraceLevel.Information,
-            $"{ToString()} is {connectionDescription}");
-
-        if (false == connected)
-        {
-            var notRecoveredError = new Error(ConnectionNotRecoveredCode,
-                    $"{ConnectionNotRecoveredMessage}," +
-                    $"recover status: {_connectionSettings.Recovery}");
-            DoClose(errorFrame, notRecoveredError);
-            return;
-        }
-
-        if (_connectionSettings.Recovery.IsTopologyActive())
-        {
-            Trace.WriteLine(TraceLevel.Information, $"{ToString()} Recovering topology");
-            var visitor = new Visitor(_management);
-            await _recordingTopologyListener.Accept(visitor)
-                .ConfigureAwait(false);
-        }
-
-        OnNewStatus(State.Open, null);
-
-        // after the connection is recovered we have to reconnect all the publishers and consumers
-        try
-        {
-            await ReconnectEntitiesAsync()
-                .ConfigureAwait(false);
-        }
-        catch (Exception e)
-        {
-            Trace.WriteLine(TraceLevel.Error, $"{ToString()} error trying to reconnect entities {e}");
-        }
-    }
-
     /// <summary>
     /// Event handler for the connection closed event.
     /// In case the error is null means that the connection is closed by the user.
@@ -494,7 +316,6 @@ await ReconnectEntitiesAsync()
     /// and then kick off a task dedicated to recovery
     /// </summary>
     /// <returns></returns>
-    /*
     private ClosedCallback BuildClosedCallback()
     {
         return async (sender, error) =>
@@ -511,6 +332,15 @@ await _semaphoreClose.WaitAsync()
                 // close all the sessions, if the connection is closed the sessions are not valid anymore
                 _nativePubSubSessions.ClearSessions();
 
+                void DoClose(Error? argError = null)
+                {
+                    Error? err = argError ?? Utils.ConvertError(error);
+                    Trace.WriteLine(TraceLevel.Verbose, $"{ToString()} is closed");
+                    OnNewStatus(State.Closed, err);
+                    ChangeEntitiesStatus(State.Closed, err);
+                    _connectionCloseTaskCompletionSource.SetResult(true);
+                }
+
                 if (error is null)
                 {
                     DoClose();
@@ -631,7 +461,6 @@ await ReconnectEntitiesAsync()
             }
         };
     }
-    */
 
     private void ChangeEntitiesStatus(State state, Error? error)
     {

RabbitMQ.AMQP.Client/Impl/AmqpConsumer.cs

@@ -23,8 +23,8 @@ private enum PauseStatus
     private readonly MessageHandler _messageHandler;
     private readonly int _initialCredits;
     private readonly Map _filters;
+    private readonly Guid _id = Guid.NewGuid();
 
-    private Guid _id = Guid.NewGuid();
     private ReceiverLink? _receiverLink;
 
     private PauseStatus _pauseStatus = PauseStatus.UNPAUSED;
@@ -47,25 +47,11 @@ public AmqpConsumer(AmqpConnection connection, string address,
 
     public override async Task OpenAsync()
     {
-        if (_receiverLink is not null)
-        {
-            if (false == _connection.Consumers.TryRemove(_id, out _))
-            {
-                // TODO error?
-            }
-
-            _id = Guid.NewGuid();
-
-            if (false == _connection.Consumers.TryAdd(_id, this))
-            {
-                // TODO error?
-            }
-        }
         try
         {
             TaskCompletionSource<ReceiverLink> attachCompletedTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
 
-            Attach attach = Utils.CreateAttach(_address, DeliveryMode.AtLeastOnce, _id.ToString(), _filters);
+            Attach attach = Utils.CreateAttach(_address, DeliveryMode.AtLeastOnce, _id, _filters);
 
             void onAttached(ILink argLink, Attach argAttach)
             {