fix: use only dotnet 8 ImeterFactory for implementation

Author: aygalinc

.ci/ubuntu/one-node/gha-setup.sh

@@ -14,10 +14,6 @@ readonly rabbitmq_image="${RABBITMQ_IMAGE:-rabbitmq:4-management}"
 readonly docker_name_prefix='rabbitmq-amqp-dotnet-client'
 readonly docker_network_name="$docker_name_prefix-network"
 
-if [[ ! -v GITHUB_ACTIONS ]]
-then
-    GITHUB_ACTIONS='false'
-fi
 
 if [[ -d $GITHUB_WORKSPACE ]]
 then
@@ -126,14 +122,7 @@ function get_rabbitmq_id
     local rabbitmq_docker_id
     rabbitmq_docker_id="$(docker inspect --format='{{.Id}}' "$rabbitmq_docker_name")"
     echo "[INFO] '$rabbitmq_docker_name' docker id is '$rabbitmq_docker_id'"
-    if [[ -v GITHUB_OUTPUT ]]
-    then
-        if [[ -f $GITHUB_OUTPUT ]]
-        then
-            echo "[INFO] GITHUB_OUTPUT file: '$GITHUB_OUTPUT'"
-        fi
-        echo "id=$rabbitmq_docker_id" >> "$GITHUB_OUTPUT"
-    fi
+ 
 }
 
 function install_ca_certificate
@@ -146,15 +135,9 @@ function install_ca_certificate
     openssl version -d
     set -o errexit
 
-    if [[ $GITHUB_ACTIONS == 'true' ]]
-    then
-        readonly openssl_store_dir='/usr/lib/ssl/certs'
-        sudo cp -vf "$GITHUB_WORKSPACE/.ci/certs/ca_certificate.pem" "$openssl_store_dir"
-        sudo ln -vsf "$openssl_store_dir/ca_certificate.pem" "$openssl_store_dir/$(openssl x509 -hash -noout -in $openssl_store_dir/ca_certificate.pem).0"
-    else
+  
         echo "[WARNING] you must install '$GITHUB_WORKSPACE/.ci/certs/ca_certificate.pem' manually into your trusted root store"
-    fi
-
+   
     openssl s_client -connect localhost:5671 \
         -CAfile "$GITHUB_WORKSPACE/.ci/certs/ca_certificate.pem" \
         -cert "$GITHUB_WORKSPACE/.ci/certs/client_localhost_certificate.pem" \

Directory.Packages.props

@@ -29,6 +29,7 @@
   </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" />

RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs

@@ -36,8 +36,9 @@ public class AmqpConnection : AbstractLifeCycle, IConnection
         private readonly RecordingTopologyListener _recordingTopologyListener = new();
 
         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.
@@ -90,10 +91,12 @@ public ReadOnlyCollection<IConsumer> GetConsumers()
         ///  - 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;
@@ -106,7 +109,7 @@ public IManagement Management()
 
         public IConsumerBuilder ConsumerBuilder()
         {
-            return new AmqpConsumerBuilder(this);
+            return new AmqpConsumerBuilder(this, _metricsReporter);
         }
 
         // TODO cancellation token
@@ -122,7 +125,7 @@ await base.OpenAsync()
         public IPublisherBuilder PublisherBuilder()
         {
             ThrowIfClosed();
-            var publisherBuilder = new AmqpPublisherBuilder(this);
+            var publisherBuilder = new AmqpPublisherBuilder(this, _metricsReporter);
             return publisherBuilder;
         }
 
@@ -218,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));

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;
         }
 
@@ -77,13 +79,8 @@ 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, metricsReporter);
+            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) =>

RabbitMQ.AMQP.Client/Impl/AmqpPublisherBuilder.cs

@@ -15,10 +15,12 @@ public class AmqpPublisherBuilder : IPublisherBuilder
         private string? _key = null;
         private string? _queue = null;
         private TimeSpan _timeout = TimeSpan.FromSeconds(10);
+        private readonly IMetricsReporter _metricsReporter;
 
-        public AmqpPublisherBuilder(AmqpConnection connection)
+        public AmqpPublisherBuilder(AmqpConnection connection, IMetricsReporter metricsReporter)
         {
             _connection = connection;
+            _metricsReporter = metricsReporter;
         }
 
         public IPublisherBuilder Exchange(IExchangeSpecification exchangeSpec)
@@ -67,12 +69,8 @@ public async Task<IPublisher> BuildAsync(CancellationToken cancellationToken = d
             {
                 address = AddressBuilderHelper.AddressBuilder().Exchange(_exchange).Queue(_queue).Key(_key).Address();
             }
-#if NET6_0_OR_GREATER
-            IMetricsReporter metricsReporter = new MetricsReporter();
-#else
-            IMetricsReporter metricsReporter = new NoOpMetricsReporter();
-#endif
-            AmqpPublisher publisher = new(_connection, address, metricsReporter, _timeout);
+            
+            AmqpPublisher publisher = new(_connection, address, _metricsReporter, _timeout);
 
             // TODO pass cancellationToken
             await publisher.OpenAsync()

RabbitMQ.AMQP.Client/Impl/MetricsReporter.cs

@@ -1,4 +1,4 @@
-#if NET6_0_OR_GREATER
+#if NET8_0_OR_GREATER
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
@@ -8,18 +8,18 @@
 
 namespace RabbitMQ.AMQP.Client.Impl
 {
-#if NET6_0_OR_GREATER
+#if NET8_0_OR_GREATER
     // .NET docs on metric instrumentation: https://learn.microsoft.com/en-us/dotnet/core/diagnostics/metrics-instrumentation
     // OpenTelemetry semantic conventions for messaging metric: https://opentelemetry.io/docs/specs/semconv/messaging/messaging-metrics
     internal sealed class MetricsReporter : IMetricsReporter
     {
         const string Version = "0.1.0";
 
-        static readonly Counter<int> s_messagingClientSentMessages;
-        static readonly Histogram<double> s_messagingClientOperationDuration;
+        readonly Counter<int> _messagingClientSentMessages;
+        readonly Histogram<double> _messagingClientOperationDuration;
 
-        static readonly Counter<int> s_messagingClientConsumedMessages;
-        static readonly Histogram<double> s_messagingProcessDuration;
+        readonly Counter<int> _messagingClientConsumedMessages;
+        readonly Histogram<double> _messagingProcessDuration;
 
         readonly KeyValuePair<string, object?>
             _messagingOperationSystemTag = new(MessagingSystem, MessagingSystemValue);
@@ -44,29 +44,29 @@ internal sealed class MetricsReporter : IMetricsReporter
         private const string SendOperation = "send";
         private const string MessagingSystemValue = "rabbitmq";
 
-        static MetricsReporter()
+        public MetricsReporter(IMeterFactory meterFactory)
         {
-            Meter meter = new("RabbitMQ.Amqp", Version);
+            Meter meter = meterFactory.Create("RabbitMQ.Amqp", Version);
 
-            s_messagingClientSentMessages = meter.CreateCounter<int>(
+            _messagingClientSentMessages = meter.CreateCounter<int>(
                 "messaging.client.sent.messages",
                 unit: "{message}",
                 description:
                 "Number of messages producer attempted to send to the broker.");
 
-            s_messagingClientOperationDuration = meter.CreateHistogram<double>(
+            _messagingClientOperationDuration = meter.CreateHistogram<double>(
                 "messaging.client.operation.duration",
                 unit: "s",
                 description:
                 "Duration of messaging operation initiated by a producer or consumer client.");
 
-            s_messagingClientConsumedMessages = meter.CreateCounter<int>(
+            _messagingClientConsumedMessages = meter.CreateCounter<int>(
                 "messaging.client.consumed.messages",
                 unit: "{message}",
                 description:
                 "Number of messages that were delivered to the application. ");
 
-            s_messagingProcessDuration = meter.CreateHistogram<double>(
+            _messagingProcessDuration = meter.CreateHistogram<double>(
                 "messaging.process.duration",
                 unit: "s",
                 description:
@@ -79,17 +79,12 @@ public void ReportMessageSendSuccess(IMetricsReporter.PublisherContext context,
             var serverPort = new KeyValuePair<string, object?>(ServerPort, context.ServerPort);
             var destination = new KeyValuePair<string, object?>(MessageDestinationName, context.Destination);
 
-            s_messagingClientSentMessages.Add(1, serverAddress, serverPort, destination, _messagingOperationSystemTag,
+            _messagingClientSentMessages.Add(1, serverAddress, serverPort, destination, _messagingOperationSystemTag,
                 _sendOperationType, _publishOperationName);
             if (startTimestamp > 0)
             {
-#if NET7_0_OR_GREATER
-            var duration = Stopwatch.GetElapsedTime(startTimestamp);
-#else
-                var duration =
-                    new TimeSpan((long)((Stopwatch.GetTimestamp() - startTimestamp) * s_stopWatchTickFrequency));
-#endif
-                s_messagingClientOperationDuration.Record(duration.TotalSeconds, serverAddress, serverPort, destination,
+                var duration = Stopwatch.GetElapsedTime(startTimestamp);
+                _messagingClientOperationDuration.Record(duration.TotalSeconds, serverAddress, serverPort, destination,
                     _messagingOperationSystemTag, _sendOperationType, _publishOperationName);
             }
         }
@@ -101,18 +96,13 @@ public void ReportMessageSendFailure(IMetricsReporter.PublisherContext context,
             var serverAddress = new KeyValuePair<string, object?>(ServerAddress, context.ServerAddress);
             var serverPort = new KeyValuePair<string, object?>(ServerPort, context.ServerPort);
             var destination = new KeyValuePair<string, object?>(MessageDestinationName, context.Destination);
-            s_messagingClientSentMessages.Add(1, errorType, serverAddress, serverPort, destination,
+            _messagingClientSentMessages.Add(1, errorType, serverAddress, serverPort, destination,
                 _messagingOperationSystemTag, _sendOperationType, _publishOperationName);
 
             if (startTimestamp > 0)
             {
-#if NET7_0_OR_GREATER
-            var duration = Stopwatch.GetElapsedTime(startTimestamp);
-#else
-                var duration =
-                    new TimeSpan((long)((Stopwatch.GetTimestamp() - startTimestamp) * s_stopWatchTickFrequency));
-#endif
-                s_messagingClientOperationDuration.Record(duration.TotalSeconds, errorType, serverAddress, serverPort,
+                var duration = Stopwatch.GetElapsedTime(startTimestamp);
+                _messagingClientOperationDuration.Record(duration.TotalSeconds, errorType, serverAddress, serverPort,
                     destination,
                     _messagingOperationSystemTag, _sendOperationType, _publishOperationName);
             }
@@ -123,20 +113,18 @@ public void ReportMessageDeliverSuccess(IMetricsReporter.ConsumerContext context
             var serverAddress = new KeyValuePair<string, object?>(ServerAddress, context.ServerAddress);
             var serverPort = new KeyValuePair<string, object?>(ServerPort, context.ServerPort);
             var destination = new KeyValuePair<string, object?>(MessageDestinationName, context.Destination);
-            s_messagingClientConsumedMessages.Add(1, serverAddress, serverPort, destination, _messagingOperationSystemTag,
+            _messagingClientConsumedMessages.Add(1, serverAddress, serverPort, destination,
+                _messagingOperationSystemTag,
                 _processOperationType, _deliverOperationName);
-            if (startTimestamp > 0)
+            if (startTimestamp <= 0)
             {
-#if NET7_0_OR_GREATER
-            var duration = Stopwatch.GetElapsedTime(startTimestamp);
-#else
-                var duration =
-                    new TimeSpan((long)((Stopwatch.GetTimestamp() - startTimestamp) * s_stopWatchTickFrequency));
-#endif
-                s_messagingProcessDuration.Record(duration.TotalSeconds, serverAddress, serverPort,
-                    destination,
-                    _messagingOperationSystemTag, _processOperationType, _deliverOperationName);
+                return;
             }
+
+            var duration = Stopwatch.GetElapsedTime(startTimestamp);
+            _messagingProcessDuration.Record(duration.TotalSeconds, serverAddress, serverPort,
+                destination,
+                _messagingOperationSystemTag, _processOperationType, _deliverOperationName);
         }
 
         public void ReportMessageDeliverFailure(IMetricsReporter.ConsumerContext context, long startTimestamp,
@@ -146,28 +134,19 @@ public void ReportMessageDeliverFailure(IMetricsReporter.ConsumerContext context
             var serverAddress = new KeyValuePair<string, object?>(ServerAddress, context.ServerAddress);
             var serverPort = new KeyValuePair<string, object?>(ServerPort, context.ServerPort);
             var destination = new KeyValuePair<string, object?>(MessageDestinationName, context.Destination);
-            s_messagingClientConsumedMessages.Add(1, errorType, serverAddress, serverPort, destination,
+            _messagingClientConsumedMessages.Add(1, errorType, serverAddress, serverPort, destination,
                 _messagingOperationSystemTag,
                 _processOperationType, _deliverOperationName);
-            if (startTimestamp > 0)
+            if (startTimestamp <= 0)
             {
-#if NET7_0_OR_GREATER
-            var duration = Stopwatch.GetElapsedTime(startTimestamp);
-#else
-                var duration =
-                    new TimeSpan((long)((Stopwatch.GetTimestamp() - startTimestamp) * s_stopWatchTickFrequency));
-#endif
-                s_messagingProcessDuration.Record(duration.TotalSeconds, errorType, serverAddress, serverPort,
-                    destination,
-                    _messagingOperationSystemTag, _processOperationType, _deliverOperationName);
+                return;
             }
+
+            var duration = Stopwatch.GetElapsedTime(startTimestamp);
+            _messagingProcessDuration.Record(duration.TotalSeconds, errorType, serverAddress, serverPort,
+                destination,
+                _messagingOperationSystemTag, _processOperationType, _deliverOperationName);
         }
-#if !NET7_0_OR_GREATER
-        const long TicksPerMicrosecond = 10;
-        const long TicksPerMillisecond = TicksPerMicrosecond * 1000;
-        const long TicksPerSecond = TicksPerMillisecond * 1000; // 10,000,000
-        static readonly double s_stopWatchTickFrequency = (double)TicksPerSecond / Stopwatch.Frequency;
-#endif
     }
 #else
 #endif