feat: introduce metrics based on system diagnostics

Directory.Packages.props

@@ -2,10 +2,10 @@
   <PropertyGroup>
     <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
   </PropertyGroup>
-
   <ItemGroup>
     <!-- RabbitMQ.Amqp.Client -->
     <PackageVersion Include="AMQPNetLite.Core" Version="2.4.11" />
+    <PackageVersion Include="Microsoft.Extensions.Diagnostics.Testing" Version="8.10.0" />
     <!-- Tests -->
     <PackageVersion Include="xunit" Version="2.9.0" />
     <PackageVersion Include="xunit.runner.visualstudio" Version="2.8.2" />
@@ -14,7 +14,6 @@
     <PackageVersion Include="Microsoft.NET.Test.Sdk" Version="17.11.0" />
     <PackageVersion Include="EasyNetQ.Management.Client" Version="3.0.0" />
   </ItemGroup>
-
   <ItemGroup Condition="$(TargetFramework)=='netstandard2.0'">
     <!--
       Note: do NOT upgrade these dependencies unless necessary
@@ -25,22 +24,19 @@
     <PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="6.0.0" />
     <PackageVersion Include="System.Runtime.CompilerServices.Unsafe" Version="6.0.0" />
   </ItemGroup>
-
   <ItemGroup Condition="'$(TargetFramework)'=='net6.0'">
     <PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="6.0.0" />
   </ItemGroup>
-
   <ItemGroup Condition="'$(TargetFramework)'=='net8.0'">
     <PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="8.0.1" />
+    <PackageVersion Include="Microsoft.Extensions.Diagnostics" Version="8.0.1" />
   </ItemGroup>
-
   <ItemGroup Condition="'$(TargetFrameworkIdentifier)'=='.NETFramework'">
     <GlobalPackageReference Include="Microsoft.NETFramework.ReferenceAssemblies" Version="1.0.3" />
   </ItemGroup>
-
   <ItemGroup Condition="'$(IsPackable)'=='true'">
     <GlobalPackageReference Include="Microsoft.CodeAnalysis.PublicApiAnalyzers" Version="3.3.4" />
     <GlobalPackageReference Include="Microsoft.SourceLink.GitHub" Version="8.0.0" />
     <GlobalPackageReference Include="MinVer" Version="5.0.0" />
   </ItemGroup>
-</Project>
+</Project>

RabbitMQ.AMQP.Client/IMetricsReporter.cs

@@ -0,0 +1,38 @@
+using Amqp;
+
+namespace RabbitMQ.AMQP.Client
+{
+    public interface IMetricsReporter
+    {
+        void ReportMessageSendSuccess(PublisherContext context, long startTimestamp);
+        void ReportMessageSendFailure(PublisherContext context, long startTimestamp, AmqpException amqpException);
+        public void ReportMessageDeliverSuccess(ConsumerContext context, long startTimestamp);
+        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; }
+        }
+    }
+}

RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs

@@ -35,16 +35,18 @@ public class AmqpConnection : AbstractLifeCycle, IConnection
         private readonly AmqpManagement _management;
         private readonly RecordingTopologyListener _recordingTopologyListener = new();
 
-        private readonly IConnectionSettings _connectionSettings;
+        internal readonly IConnectionSettings _connectionSettings;
+        private readonly IMetricsReporter _metricsReporter;
         internal readonly AmqpSessionManagement _nativePubSubSessions;
-
+ 
         /// <summary>
         /// Publishers contains all the publishers created by the connection.
         /// Each connection can have multiple publishers.
         /// They key is the publisher Id ( a Guid)  
         /// 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 +59,6 @@ public IRpcServerBuilder RpcServerBuilder()
 
         public IRpcClientBuilder RpcClientBuilder()
         {
-
             return new AmqpRpcClientBuilder(this);
         }
 
@@ -82,17 +83,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 = default)
         {
-            var connection = new AmqpConnection(connectionSettings);
+            var connection = new AmqpConnection(connectionSettings, metricsReporter ?? new NoOpMetricsReporter());
             await connection.OpenAsync()
                 .ConfigureAwait(false);
             return connection;
@@ -105,7 +109,7 @@ public IManagement Management()
 
         public IConsumerBuilder ConsumerBuilder()
         {
-            return new AmqpConsumerBuilder(this);
+            return new AmqpConsumerBuilder(this, _metricsReporter);
         }
 
         // TODO cancellation token
@@ -121,7 +125,7 @@ await base.OpenAsync()
         public IPublisherBuilder PublisherBuilder()
         {
             ThrowIfClosed();
-            var publisherBuilder = new AmqpPublisherBuilder(this);
+            var publisherBuilder = new AmqpPublisherBuilder(this, _metricsReporter);
             return publisherBuilder;
         }
 
@@ -184,6 +188,7 @@ protected override void Dispose(bool disposing)
                 {
                     _nativeConnection.Closed -= _closedCallback;
                 }
+
                 _semaphoreOpen.Dispose();
                 _semaphoreClose.Dispose();
             }
@@ -216,9 +221,10 @@ await consumer.CloseAsync()
             }
         }
 
-        private AmqpConnection(IConnectionSettings connectionSettings)
+        private AmqpConnection(IConnectionSettings connectionSettings, IMetricsReporter metricsReporter)
         {
             _connectionSettings = connectionSettings;
+            _metricsReporter = metricsReporter;
             _nativePubSubSessions = new AmqpSessionManagement(this, 1);
             _management =
                 new AmqpManagement(new AmqpManagementParameters(this).TopologyListener(_recordingTopologyListener));
@@ -242,10 +248,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 +303,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 +451,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);
@@ -143,17 +154,20 @@ private async Task ProcessMessages()
                         continue;
                     }
 
+                    _metricsReporter.ReportMessageDeliverSuccess(consumerContext, startTimestamp);
                     _unsettledMessageCounter.Increment();
 
                     IContext context = new DeliveryContext(_receiverLink, nativeMessage, _unsettledMessageCounter);
                     var amqpMessage = new AmqpMessage(nativeMessage);
-
+                    
                     // TODO catch exceptions thrown by handlers,
                     // then call exception handler?
+
                     if (_configuration.Handler != null)
                     {
                         await _configuration.Handler(context, amqpMessage).ConfigureAwait(false);
                     }
+
                 }
             }
             catch (Exception e)
@@ -162,7 +176,7 @@ private async Task ProcessMessages()
                 {
                     return;
                 }
-
+                
                 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

@@ -30,9 +30,11 @@ public class ConsumerConfiguration
     public class AmqpConsumerBuilder : IConsumerBuilder
     {
         private readonly ConsumerConfiguration _configuration = new();
+        private readonly IMetricsReporter _metricsReporter;
 
-        public AmqpConsumerBuilder(AmqpConnection connection)
+        public AmqpConsumerBuilder(AmqpConnection connection, IMetricsReporter metricsReporter)
         {
+            _metricsReporter = metricsReporter;
             _configuration.Connection = connection;
         }
 
@@ -78,7 +80,7 @@ public async Task<IConsumer> BuildAndStartAsync(CancellationToken cancellationTo
                 throw new ConsumerException("Message handler is not set");
             }
 
-            AmqpConsumer consumer = new(_configuration);
+            AmqpConsumer consumer = new(_configuration, _metricsReporter);
 
             // TODO pass cancellationToken
             await consumer.OpenAsync()

RabbitMQ.AMQP.Client/Impl/AmqpEnvironment.cs

@@ -12,23 +12,27 @@ namespace RabbitMQ.AMQP.Client.Impl
 {
     public class AmqpEnvironment : IEnvironment
     {
-        private IConnectionSettings? ConnectionSettings { get; }
+        private IConnectionSettings ConnectionSettings { get; }
         private long _sequentialId = 0;
         private readonly ConcurrentDictionary<long, IConnection> _connections = new();
+        private readonly IMetricsReporter? _metricsReporter;
 
-        private AmqpEnvironment(IConnectionSettings connectionSettings)
+        private AmqpEnvironment(IConnectionSettings connectionSettings, IMetricsReporter? metricsReporter)
         {
             ConnectionSettings = connectionSettings;
+            _metricsReporter = metricsReporter;
         }
 
-        public static Task<IEnvironment> CreateAsync(IConnectionSettings connectionSettings)
+        public static Task<IEnvironment> CreateAsync(IConnectionSettings connectionSettings,
+            IMetricsReporter? metricsReporter = default)
         {
-            return Task.FromResult<IEnvironment>(new AmqpEnvironment(connectionSettings));
+            return Task.FromResult<IEnvironment>(new AmqpEnvironment(connectionSettings,
+                metricsReporter ?? new NoOpMetricsReporter()));
         }
 
         public async Task<IConnection> CreateConnectionAsync(IConnectionSettings connectionSettings)
         {
-            IConnection c = await AmqpConnection.CreateAsync(connectionSettings).ConfigureAwait(false);
+            IConnection c = await AmqpConnection.CreateAsync(connectionSettings,_metricsReporter).ConfigureAwait(false);
             c.Id = Interlocked.Increment(ref _sequentialId);
             _connections.TryAdd(c.Id, c);
             c.ChangeState += (sender, previousState, currentState, failureCause) =>