feat: introduce metrics based on system diagnostics

RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs

@@ -35,7 +35,7 @@ public class AmqpConnection : AbstractLifeCycle, IConnection
         private readonly AmqpManagement _management;
         private readonly RecordingTopologyListener _recordingTopologyListener = new();
 
-        private readonly IConnectionSettings _connectionSettings;
+        internal readonly IConnectionSettings _connectionSettings;
         internal readonly AmqpSessionManagement _nativePubSubSessions;
 
         /// <summary>
@@ -45,6 +45,7 @@ public class AmqpConnection : AbstractLifeCycle, IConnection
         /// See <see cref="AmqpPublisher"/>
         /// </summary>
         internal ConcurrentDictionary<Guid, IPublisher> Publishers { get; } = new();
+
         internal ConcurrentDictionary<Guid, IConsumer> Consumers { get; } = new();
 
         private readonly TaskCompletionSource<bool> _connectionClosedTcs =
@@ -57,7 +58,6 @@ public IRpcServerBuilder RpcServerBuilder()
 
         public IRpcClientBuilder RpcClientBuilder()
         {
-
             return new AmqpRpcClientBuilder(this);
         }
 
@@ -82,16 +82,20 @@ public ReadOnlyCollection<IConsumer> GetConsumers()
         }
 
         public long Id { get; set; }
+
         /// <summary>
         /// Creates a new instance of <see cref="AmqpConnection"/>
         /// Through the Connection is possible to create:
         ///  - Management. See <see cref="AmqpManagement"/>
         ///  - Publishers and Consumers: See <see cref="AmqpPublisherBuilder"/> and <see cref="AmqpConsumerBuilder"/> 
         /// </summary>
         /// <param name="connectionSettings"></param>
+        /// <param name="metricsReporter"></param>
         /// <returns></returns>
-        public static async Task<IConnection> CreateAsync(IConnectionSettings connectionSettings)
+        public static async Task<IConnection> CreateAsync(IConnectionSettings connectionSettings,
+            IMetricsReporter? metricsReporter = null)
         {
+            metricsReporter ??= new NoOpMetricsReporter();
             var connection = new AmqpConnection(connectionSettings);
             await connection.OpenAsync()
                 .ConfigureAwait(false);
@@ -184,6 +188,7 @@ protected override void Dispose(bool disposing)
                 {
                     _nativeConnection.Closed -= _closedCallback;
                 }
+
                 _semaphoreOpen.Dispose();
                 _semaphoreClose.Dispose();
             }
@@ -242,10 +247,7 @@ await _semaphoreOpen.WaitAsync()
                     HostName = $"vhost:{_connectionSettings.VirtualHost}",
                     // Note: no need to set cf.AMQP.ContainerId
                     ContainerId = _connectionSettings.ContainerId,
-                    Properties = new Fields()
-                    {
-                        [new Symbol("connection_name")] = _connectionSettings.ContainerId,
-                    }
+                    Properties = new Fields() { [new Symbol("connection_name")] = _connectionSettings.ContainerId, }
                 };
 
                 if (_connectionSettings.MaxFrameSize > uint.MinValue)
@@ -300,7 +302,8 @@ void OnOpened(Amqp.IConnection connection, Open open1)
                     if (_connectionSettings is null)
                     {
                         // TODO create "internal bug" exception type?
-                        throw new InvalidOperationException("_connectionSettings is null, report via https://github.com/rabbitmq/rabbitmq-amqp-dotnet-client/issues");
+                        throw new InvalidOperationException(
+                            "_connectionSettings is null, report via https://github.com/rabbitmq/rabbitmq-amqp-dotnet-client/issues");
                     }
                     else
                     {
@@ -447,8 +450,8 @@ await OpenConnectionAsync(CancellationToken.None)
                         if (false == connected)
                         {
                             var notRecoveredError = new Error(ConnectionNotRecoveredCode,
-                                    $"{ConnectionNotRecoveredMessage}," +
-                                    $"recover status: {_connectionSettings.Recovery}");
+                                $"{ConnectionNotRecoveredMessage}," +
+                                $"recover status: {_connectionSettings.Recovery}");
                             DoClose(notRecoveredError);
                             return;
                         }

RabbitMQ.AMQP.Client/Impl/AmqpConsumer.cs

@@ -3,12 +3,15 @@
 // Copyright (c) 2017-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 
 using System;
+using System.Diagnostics;
 using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
 using Amqp;
 using Amqp.Framing;
 using Amqp.Types;
+using Trace = Amqp.Trace;
+using TraceLevel = Amqp.TraceLevel;
 
 namespace RabbitMQ.AMQP.Client.Impl
 {
@@ -28,10 +31,12 @@ private enum PauseStatus
         private PauseStatus _pauseStatus = PauseStatus.UNPAUSED;
         private readonly UnsettledMessageCounter _unsettledMessageCounter = new();
         private readonly ConsumerConfiguration _configuration;
+        private readonly IMetricsReporter _metricsReporter;
 
-        public AmqpConsumer(ConsumerConfiguration configuration)
+        public AmqpConsumer(ConsumerConfiguration configuration, IMetricsReporter metricsReporter)
         {
             _configuration = configuration;
+            _metricsReporter = metricsReporter;
             if (false == _configuration.Connection.Consumers.TryAdd(_id, this))
             {
                 // TODO error?
@@ -120,6 +125,11 @@ await base.OpenAsync()
 
         private async Task ProcessMessages()
         {
+            IMetricsReporter.ConsumerContext consumerContext = new(_configuration.Address,
+                _configuration.Connection._connectionSettings.Host,
+                _configuration.Connection._connectionSettings.Port);
+
+            long startTimestamp = 0;
             try
             {
                 if (_receiverLink is null)
@@ -130,6 +140,7 @@ private async Task ProcessMessages()
 
                 while (_receiverLink is { LinkState: LinkState.Attached })
                 {
+                    startTimestamp = Stopwatch.GetTimestamp();
                     // TODO the timeout waiting for messages should be configurable
                     TimeSpan timeout = TimeSpan.FromSeconds(60);
                     Message? nativeMessage = await _receiverLink.ReceiveAsync(timeout).ConfigureAwait(false);
@@ -154,6 +165,8 @@ private async Task ProcessMessages()
                     {
                         await _configuration.Handler(context, amqpMessage).ConfigureAwait(false);
                     }
+
+                    _metricsReporter.ReportMessageDeliverSuccess(consumerContext, startTimestamp);
                 }
             }
             catch (Exception e)
@@ -163,6 +176,7 @@ private async Task ProcessMessages()
                     return;
                 }
 
+                _metricsReporter.ReportMessageDeliverFailure(consumerContext, startTimestamp, e);
                 Trace.WriteLine(TraceLevel.Error, $"{ToString()} Failed to process messages, {e}");
                 // TODO this is where a Listener should get a closed event
                 // See the ConsumerShouldBeClosedWhenQueueIsDeleted test

RabbitMQ.AMQP.Client/Impl/AmqpConsumerBuilder.cs

@@ -77,8 +77,13 @@ public async Task<IConsumer> BuildAndStartAsync(CancellationToken cancellationTo
             {
                 throw new ConsumerException("Message handler is not set");
             }
+#if NET6_0_OR_GREATER
+            IMetricsReporter metricsReporter = new MetricsReporter();
+#else
+            IMetricsReporter metricsReporter = new NoOpMetricsReporter();
+#endif
 
-            AmqpConsumer consumer = new(_configuration);
+            AmqpConsumer consumer = new(_configuration, metricsReporter);
 
             // TODO pass cancellationToken
             await consumer.OpenAsync()

RabbitMQ.AMQP.Client/Impl/AmqpPublisher.cs

@@ -3,6 +3,7 @@
 // Copyright (c) 2017-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 
 using System;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 using Amqp;
@@ -17,14 +18,17 @@ public class AmqpPublisher : AbstractReconnectLifeCycle, IPublisher
         private readonly AmqpConnection _connection;
         private readonly TimeSpan _timeout;
         private readonly string? _address;
+        private readonly IMetricsReporter _metricsReporter;
         private readonly Guid _id = Guid.NewGuid();
 
         private SenderLink? _senderLink = null;
 
-        public AmqpPublisher(AmqpConnection connection, string? address, TimeSpan timeout)
+        public AmqpPublisher(AmqpConnection connection, string? address, IMetricsReporter metricsReporter,
+            TimeSpan timeout)
         {
             _connection = connection;
             _address = address;
+            _metricsReporter = metricsReporter;
             _timeout = timeout;
 
             if (false == _connection.Publishers.TryAdd(_id, this))
@@ -121,6 +125,10 @@ public async Task<PublishResult> PublishAsync(IMessage message, CancellationToke
                     "_senderLink is null, report via https://github.com/rabbitmq/rabbitmq-amqp-dotnet-client/issues");
             }
 
+            IMetricsReporter.PublisherContext context =
+                new(_address, _connection._connectionSettings.Host,
+                    _connection._connectionSettings.Port);
+            long startTimestamp = Stopwatch.GetTimestamp();
             try
             {
                 TaskCompletionSource<PublishOutcome> messagePublishedTcs =
@@ -140,7 +148,8 @@ void OutcomeCallback(ILink sender, Message inMessage, Outcome outcome, object st
                         case Rejected rejectedOutcome:
                             {
                                 const OutcomeState publishState = OutcomeState.Rejected;
-                                publishOutcome = new PublishOutcome(publishState, Utils.ConvertError(rejectedOutcome.Error));
+                                publishOutcome = new PublishOutcome(publishState,
+                                    Utils.ConvertError(rejectedOutcome.Error));
                                 break;
                             }
                         case Released:
@@ -175,11 +184,12 @@ void OutcomeCallback(ILink sender, Message inMessage, Outcome outcome, object st
                 // PublishOutcome publishOutcome = await messagePublishedTcs.Task.WaitAsync(TimeSpan.FromSeconds(5), cancellationToken)
                 PublishOutcome publishOutcome = await messagePublishedTcs.Task.WaitAsync(TimeSpan.FromSeconds(5))
                     .ConfigureAwait(false);
-
+                _metricsReporter.ReportMessageSendSuccess(context, startTimestamp);
                 return new PublishResult(message, publishOutcome);
             }
             catch (AmqpException ex)
             {
+                _metricsReporter.ReportMessageSendFailure(context, startTimestamp, ex);
                 var publishOutcome = new PublishOutcome(OutcomeState.Rejected, Utils.ConvertError(ex.Error));
                 return new PublishResult(message, publishOutcome);
             }

RabbitMQ.AMQP.Client/Impl/AmqpPublisherBuilder.cs

@@ -67,8 +67,12 @@ public async Task<IPublisher> BuildAsync(CancellationToken cancellationToken = d
             {
                 address = AddressBuilderHelper.AddressBuilder().Exchange(_exchange).Queue(_queue).Key(_key).Address();
             }
-
-            AmqpPublisher publisher = new(_connection, address, _timeout);
+#if NET6_0_OR_GREATER
+            IMetricsReporter metricsReporter = new MetricsReporter();
+#else
+            IMetricsReporter metricsReporter = new NoOpMetricsReporter();
+#endif
+            AmqpPublisher publisher = new(_connection, address, metricsReporter, _timeout);
 
             // TODO pass cancellationToken
             await publisher.OpenAsync()

RabbitMQ.AMQP.Client/Impl/IMetricsReporter.cs

@@ -0,0 +1,40 @@
+using System;
+using Amqp;
+
+namespace RabbitMQ.AMQP.Client.Impl
+{
+    public interface IMetricsReporter
+    {
+        void ReportMessageSendSuccess(PublisherContext context, long startTimestamp);
+        void ReportMessageSendFailure(PublisherContext context, long startTimestamp, AmqpException amqpException);
+        public void ReportMessageDeliverSuccess(ConsumerContext context, long startTimestamp);
+        void ReportMessageDeliverFailure(ConsumerContext consumerContext, long startTimestamp, Exception exception);
+        sealed class ConsumerContext
+        {
+            public ConsumerContext(string? destination, string serverAddress, int serverPort)
+            {
+                Destination = destination;
+                ServerAddress = serverAddress;
+                ServerPort = serverPort;
+            }
+
+            public string? Destination { get; }
+            public string ServerAddress { get; }
+            public int ServerPort { get; }
+        }
+
+        sealed class PublisherContext
+        {
+            public PublisherContext(string? destination, string serverAddress, int serverPort)
+            {
+                Destination = destination;
+                ServerAddress = serverAddress;
+                ServerPort = serverPort;
+            }
+
+            public string? Destination { get; }
+            public string ServerAddress { get; }
+            public int ServerPort { get; }
+        }
+    }
+}