fix: fix tests

Author: aygalinc

Tests/Consumer/ConsumerMetricsTests.cs

@@ -3,12 +3,9 @@
 // Copyright (c) 2017-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 
 #if NET8_0_OR_GREATER
-using System.Diagnostics.Metrics;
 using System.Threading.Tasks;
-using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Diagnostics.Metrics.Testing;
 using RabbitMQ.AMQP.Client;
-using RabbitMQ.AMQP.Client.Impl;
 using Xunit;
 using Xunit.Abstractions;
 
@@ -79,16 +76,6 @@ public async Task RecordMessageDeliverySuccessAndDuration()
         await consumer.CloseAsync();
         consumer.Dispose();
     }
-
-    // Setup a new service provider. This example creates the collection explicitly but you might leverage
-    // a host or some other application setup code to do this as well.
-    private ServiceProvider CreateServiceProvider()
-    {
-        return new ServiceCollection()
-            .AddMetrics()
-            .AddSingleton<IMetricsReporter, MetricsReporter>()
-            .BuildServiceProvider();
-    }
 }
 
 #else

Tests/Publisher/PublisherMetricsTests.cs

@@ -23,10 +23,10 @@ public async Task PublisherMetricsShouldBeIncrementedWhenMessageIsSendWithSucces
         Assert.NotNull(_management);
         Assert.NotNull(_meterFactory);
         Assert.NotNull(_metricsReporter);
-        var messageSendCollector =
-            new MetricCollector<int>(_meterFactory, "RabbitMQ.Amqp", "messaging.client.sent.messages");
-        var messageSendDurationCollector =
-            new MetricCollector<double>(_meterFactory, "RabbitMQ.Amqp", "messaging.client.operation.duration");
+        MetricCollector<int> clientSendMessageCounterCollector =
+            new(_meterFactory, "RabbitMQ.Amqp", "messaging.client.sent.messages");
+        MetricCollector<double> clientSendDurationCollector =
+            new(_meterFactory, "RabbitMQ.Amqp", "messaging.client.operation.duration");
 
         IQueueSpecification queueSpecification = _management.Queue(_queueName);
         await queueSpecification.DeclareAsync();
@@ -37,34 +37,122 @@ public async Task PublisherMetricsShouldBeIncrementedWhenMessageIsSendWithSucces
 
         await SystemUtils.WaitUntilQueueMessageCount(queueSpecification, 1);
 
-        var consumedMessagesMeasurements = messageSendCollector.GetMeasurementSnapshot();
-        Assert.NotEmpty(consumedMessagesMeasurements);
-        Assert.Equal(1, consumedMessagesMeasurements[0].Value);
-        Assert.Equal(consumedMessagesMeasurements[0].Tags["messaging.system"], "rabbitmq");
-        Assert.Equal(consumedMessagesMeasurements[0].Tags["messaging.operation.name"], "publish");
-        Assert.Equal(consumedMessagesMeasurements[0].Tags["messaging.operation.type"], "send");
-        Assert.Equal(consumedMessagesMeasurements[0].Tags["messaging.destination.name"],
+        var clientSendMessagesMeasurements =
+            clientSendMessageCounterCollector
+                .GetMeasurementSnapshot();
+        Assert.NotEmpty(clientSendMessagesMeasurements);
+        Assert.Equal(1, clientSendMessagesMeasurements[0].Value);
+        Assert.Equal(clientSendMessagesMeasurements[0].Tags["messaging.system"], "rabbitmq");
+        Assert.Equal(clientSendMessagesMeasurements[0].Tags["messaging.operation.name"], "publish");
+        Assert.Equal(clientSendMessagesMeasurements[0].Tags["messaging.operation.type"], "send");
+        Assert.Equal(clientSendMessagesMeasurements[0].Tags["messaging.destination.name"],
             $"/queues/{Utils.EncodePathSegment(queueSpecification.QueueName)}");
-        Assert.Equal(consumedMessagesMeasurements[0].Tags["server.port"],
+        Assert.Equal(clientSendMessagesMeasurements[0].Tags["server.port"],
             _connectionSettings!.Port);
-        Assert.Equal(consumedMessagesMeasurements[0].Tags["server.address"],
+        Assert.Equal(clientSendMessagesMeasurements[0].Tags["server.address"],
             _connectionSettings!.Host);
 
-        var sendDurationsMeasurements = messageSendDurationCollector.GetMeasurementSnapshot();
-        Assert.NotEmpty(sendDurationsMeasurements);
-        Assert.True(sendDurationsMeasurements[0].Value > 0);
-        Assert.Equal(sendDurationsMeasurements[0].Tags["messaging.system"], "rabbitmq");
-        Assert.Equal(sendDurationsMeasurements[0].Tags["messaging.operation.name"], "publish");
-        Assert.Equal(sendDurationsMeasurements[0].Tags["messaging.operation.type"], "send");
-        Assert.Equal(sendDurationsMeasurements[0].Tags["messaging.destination.name"],
+        var clientSendDurationsMeasurements =
+            clientSendDurationCollector
+                .GetMeasurementSnapshot();
+        Assert.NotEmpty(clientSendDurationsMeasurements);
+        Assert.True(clientSendDurationsMeasurements[0].Value > 0);
+        Assert.Equal(clientSendDurationsMeasurements[0].Tags["messaging.system"], "rabbitmq");
+        Assert.Equal(clientSendDurationsMeasurements[0].Tags["messaging.operation.name"], "publish");
+        Assert.Equal(clientSendDurationsMeasurements[0].Tags["messaging.operation.type"], "send");
+        Assert.Equal(clientSendDurationsMeasurements[0].Tags["messaging.destination.name"],
             $"/queues/{Utils.EncodePathSegment(queueSpecification.QueueName)}");
-        Assert.Equal(sendDurationsMeasurements[0].Tags["server.port"],
+        Assert.Equal(clientSendDurationsMeasurements[0].Tags["server.port"],
             _connectionSettings!.Port);
-        Assert.Equal(sendDurationsMeasurements[0].Tags["server.address"],
+        Assert.Equal(clientSendDurationsMeasurements[0].Tags["server.address"],
             _connectionSettings!.Host);
 
         await publisher.CloseAsync();
         publisher.Dispose();
     }
+
+    [Fact]
+    public async Task PublisherShouldRecordAFailureWhenSendingThrowAnException()
+    {
+        Assert.NotNull(_connection);
+        Assert.NotNull(_management);
+        Assert.NotNull(_meterFactory);
+        Assert.NotNull(_metricsReporter);
+        MetricCollector<int> clientSendMessageCounterCollector =
+            new(_meterFactory, "RabbitMQ.Amqp", "messaging.client.sent.messages");
+        MetricCollector<double> clientSendDurationCollector =
+            new(_meterFactory, "RabbitMQ.Amqp", "messaging.client.operation.duration");
+        
+        IMessage message = new AmqpMessage(Encoding.ASCII.GetBytes("hello"));
+
+        IExchangeSpecification exchangeSpecification = _management.Exchange(_exchangeName).Type(ExchangeType.FANOUT);
+        await exchangeSpecification.DeclareAsync();
+
+        IPublisherBuilder publisherBuilder = _connection.PublisherBuilder();
+        // TODO implement Listeners
+        IPublisher publisher = await publisherBuilder.Exchange(exchangeSpecification).BuildAsync();
+
+        try
+        {
+            IQueueSpecification queueSpecification = _management.Queue().Exclusive(true);
+            IQueueInfo queueInfo = await queueSpecification.DeclareAsync();
+            IBindingSpecification bindingSpecification = _management.Binding()
+                .SourceExchange(_exchangeName)
+                .DestinationQueue(queueInfo.Name());
+            await bindingSpecification.BindAsync();
+
+            PublishResult publishResult = await publisher.PublishAsync(message);
+            Assert.Equal(OutcomeState.Accepted, publishResult.Outcome.State);
+        }
+        finally
+        {
+            await exchangeSpecification.DeleteAsync();
+        }
+
+        for (int i = 0; i < 100; i++)
+        {
+            PublishResult nextPublishResult = await publisher.PublishAsync(message);
+            if (OutcomeState.Rejected == nextPublishResult.Outcome.State)
+            {
+                break;
+            }
+
+            await Task.Delay(TimeSpan.FromMilliseconds(100));
+        }
+
+        var failedSendMeasure =
+            clientSendMessageCounterCollector
+                .LastMeasurement!;
+        Assert.Equal(1, failedSendMeasure.Value);
+        Assert.Equal(failedSendMeasure.Tags["messaging.system"], "rabbitmq");
+        Assert.Equal(failedSendMeasure.Tags["messaging.operation.name"], "publish");
+        Assert.Equal(failedSendMeasure.Tags["messaging.operation.type"], "send");
+        Assert.Equal(failedSendMeasure.Tags["messaging.destination.name"],
+            $"/exchanges/{Utils.EncodePathSegment(exchangeSpecification.ExchangeName)}");
+        Assert.Equal(failedSendMeasure.Tags["server.port"],
+            _connectionSettings!.Port);
+        Assert.Equal(failedSendMeasure.Tags["server.address"],
+            _connectionSettings!.Host);
+        Assert.Equal(failedSendMeasure.Tags["error.type"],
+            "amqp:not-found");
+
+        var failedMessageSendDuration =
+            clientSendDurationCollector
+                .LastMeasurement!;
+        Assert.True(failedMessageSendDuration.Value > 0);
+        Assert.Equal(failedMessageSendDuration.Tags["messaging.system"], "rabbitmq");
+        Assert.Equal(failedMessageSendDuration.Tags["messaging.operation.name"], "publish");
+        Assert.Equal(failedMessageSendDuration.Tags["messaging.operation.type"], "send");
+        Assert.Equal(failedMessageSendDuration.Tags["messaging.destination.name"],
+            $"/exchanges/{Utils.EncodePathSegment(exchangeSpecification.ExchangeName)}");
+        Assert.Equal(failedMessageSendDuration.Tags["server.port"],
+            _connectionSettings!.Port);
+        Assert.Equal(failedMessageSendDuration.Tags["server.address"],
+            _connectionSettings!.Host);
+        Assert.Equal(failedMessageSendDuration.Tags["error.type"],
+            "amqp:not-found");
+        await publisher.CloseAsync();
+        publisher.Dispose();
+    }
 }
 #endif