Merge pull request #24 from atidev/rewrite-metricsfactory-to-di

metricsFactory from DI
atidev · Mar 28, 2024 · 2fa7a3a · 2fa7a3a
2 parents 030fb63 + 420b6cd
commit 2fa7a3a
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 9 deletions.
diff --git a/ATI.Services.RabbitMQ/ATI.Services.RabbitMQ.csproj b/ATI.Services.RabbitMQ/ATI.Services.RabbitMQ.csproj
@@ -16,7 +16,7 @@
     <NoWarn>1701;1702;CS1591;CS1571;CS1573;CS1574</NoWarn>
   </PropertyGroup>
   <ItemGroup>
-    <PackageReference Include="atisu.services.common" Version="13.7.2" />
+    <PackageReference Include="atisu.services.common" Version="15.0.0" />
     <PackageReference Include="EasyNetQ" Version="7.4.3" />
   </ItemGroup>
 </Project>
diff --git a/ATI.Services.RabbitMQ/EventbusManager.cs b/ATI.Services.RabbitMQ/EventbusManager.cs
@@ -41,8 +41,8 @@ public class EventbusManager : IDisposable, IInitializer
     private readonly JsonSerializer _jsonSerializer;
     private readonly string _connectionString;
 
-    private readonly MetricsFactory _inMetricsFactory = MetricsFactory.CreateRabbitMqMetricsFactory(RabbitMetricsType.Subscribe, nameof(EventbusManager), additionalSummaryLabels: "rmq_app_id");
-    private readonly MetricsFactory _outMetricsFactory = MetricsFactory.CreateRabbitMqMetricsFactory(RabbitMetricsType.Publish, nameof(EventbusManager));
+    private readonly MetricsInstance _inMetrics;
+    private readonly MetricsInstance _outMetrics;
 
     private readonly ILogger _logger = LogManager.GetCurrentClassLogger();
     private ConcurrentBag<SubscriptionInfo> _subscriptions = new();
@@ -52,12 +52,19 @@ public class EventbusManager : IDisposable, IInitializer
     private static readonly UTF8Encoding BodyEncoding = new(false);
     private readonly RmqTopology _rmqTopology;
 
-    public EventbusManager(JsonSerializer jsonSerializer, IOptions<EventbusOptions> options, RmqTopology rmqTopology)
+    public EventbusManager(
+        JsonSerializer jsonSerializer, 
+        IOptions<EventbusOptions> options, 
+        RmqTopology rmqTopology,
+        MetricsFactory metricsFactory)
     {
         _options = options.Value;
         _connectionString = options.Value.ConnectionString;
         _jsonSerializer = jsonSerializer;
         _rmqTopology = rmqTopology;
+
+        _inMetrics = metricsFactory.CreateRabbitMqMetricsFactory(RabbitMetricsType.Subscribe, nameof(EventbusManager), additionalSummaryLabels: "rmq_app_id");
+        _outMetrics = metricsFactory.CreateRabbitMqMetricsFactory(RabbitMetricsType.Publish, nameof(EventbusManager));
 
         _subscribePolicy = Policy.Handle<Exception>()
                                  .WaitAndRetryForeverAsync(_ => _options.RabbitConnectInterval,
@@ -113,7 +120,7 @@ public async Task PublishRawAsync(
             string.IsNullOrWhiteSpace(publishBody))
             return;
 
-        using (_outMetricsFactory.CreateLoggingMetricsTimer(metricEntity, $"{exchangeName}:{routingKey}"))
+        using (_outMetrics.CreateLoggingMetricsTimer(metricEntity, $"{exchangeName}:{routingKey}"))
         {
             var messageProperties = GetProperties(additionalHeaders, withAcceptLang);
             var exchange = new Exchange(exchangeName);
@@ -150,7 +157,7 @@ public async Task PublishAsync<T>(
             publishObject == null)
             return;
 
-        using (_outMetricsFactory.CreateLoggingMetricsTimer(metricEntity, $"{exchangeName}:{routingKey}"))
+        using (_outMetrics.CreateLoggingMetricsTimer(metricEntity, $"{exchangeName}:{routingKey}"))
         {
             var messageProperties = GetProperties(additionalHeaders, withAcceptLang);
             var exchange = new Exchange(exchangeName);
@@ -297,7 +304,7 @@ private async Task<IDisposable> SubscribePrivateAsync(
 
         async Task HandleEventBusMessageWithPolicy(ReadOnlyMemory<byte> body, MessageProperties props, MessageReceivedInfo info)
         {
-            using (_inMetricsFactory.CreateLoggingMetricsTimer(metricEntity ?? "Eventbus",
+            using (_inMetrics.CreateLoggingMetricsTimer(metricEntity ?? "Eventbus",
                                                                $"{info.Exchange}:{info.RoutingKey}",
                                                                additionalLabels: props.AppId ?? "Unknown"))
             {
@@ -329,7 +336,7 @@ private async Task BindConsumerAsync(QueueExchangeBinding mainQueueBinding,
         {
             return async (body, props, info) =>
             {
-                using (_inMetricsFactory.CreateLoggingMetricsTimer(metricEntity ?? "Eventbus",
+                using (_inMetrics.CreateLoggingMetricsTimer(metricEntity ?? "Eventbus",
                            $"{info.Exchange}:{info.RoutingKey}",
                            additionalLabels: props.AppId ?? "Unknown"))
                 {
@@ -362,7 +369,7 @@ Func<ReadOnlyMemory<byte>, MessageProperties, MessageReceivedInfo, Task> HandleP
         {
             return async (body, props, info) =>
             {
-                using (_outMetricsFactory.CreateLoggingMetricsTimer($"{metricEntity ?? "Eventbus"}-Poison", $"{info.Exchange}:{info.RoutingKey}",
+                using (_outMetrics.CreateLoggingMetricsTimer($"{metricEntity ?? "Eventbus"}-Poison", $"{info.Exchange}:{info.RoutingKey}",
                            additionalLabels: props.AppId ?? "Unknown"))
                 {
                     HandleMessageProps(props);