add already exists check to pgmq initialization

Author: Alex-Kum

src/Motor.Extensions.Hosting.PgMq/PgMqHostBuilderExtensions.cs

@@ -52,6 +52,7 @@ public static void AddPgMqWithConfig<T>(this IPublisherBuilder<T> builder, IConf
         builder.Configure<PgMqPublisherOptions>(config);
         builder.AddPublisher(sp => new PgMqMessageProducer<T>(
             MSOptions.Create(options),
+            sp.GetRequiredService<ILogger<PgMqMessageProducer<T>>>(),
             sp.GetRequiredService<IOptions<PublisherOptions>>(),
             new NpgmqClient(options.ToConnectionString())
         ));

src/Motor.Extensions.Hosting.PgMq/PgMqMessageConsumer.cs

@@ -8,6 +8,7 @@
 using Motor.Extensions.Hosting.CloudEvents;
 using Motor.Extensions.Hosting.PgMq.Options;
 using Npgmq;
+using Npgsql;
 
 namespace Motor.Extensions.Hosting.PgMq;
 
@@ -83,15 +84,13 @@ public async Task StartAsync(CancellationToken token = default)
         }
         catch (NpgmqException e)
         {
-            // This is possibly a race with another extension initializing in parallel.
-            // Check if the extension is already present.
-            try
+            if (e.InnerException is PostgresException { SqlState: PostgresErrorCodes.UniqueViolation })
             {
-                await _npgmqClient.GetPgmqVersionAsync(token);
+                _logger.LogWarning(e, "pgmq extension already exists, assuming it was created by another instance");
             }
-            catch (NpgmqException)
+            else
             {
-                throw e;
+                throw;
             }
         }
 

src/Motor.Extensions.Hosting.PgMq/PgMqMessageProducer.cs

@@ -2,11 +2,13 @@
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using Motor.Extensions.Hosting.Abstractions;
 using Motor.Extensions.Hosting.CloudEvents;
 using Motor.Extensions.Hosting.PgMq.Options;
 using Npgmq;
+using Npgsql;
 
 namespace Motor.Extensions.Hosting.PgMq;
 
@@ -23,16 +25,19 @@ public class PgMqMessageProducer<TOutput> : IRawMessagePublisher<TOutput>
     where TOutput : notnull
 {
     private readonly PgMqPublisherOptions _options;
+    private readonly ILogger<PgMqMessageProducer<TOutput>> _logger;
     private readonly PublisherOptions _publisherOptions;
     private readonly INpgmqClient _npgmqClient;
 
     public PgMqMessageProducer(
         IOptions<PgMqPublisherOptions> options,
+        ILogger<PgMqMessageProducer<TOutput>> logger,
         IOptions<PublisherOptions> publisherOptions,
         INpgmqClient npgmqClient
     )
     {
         _options = options.Value;
+        _logger = logger;
         _publisherOptions = publisherOptions.Value;
         _npgmqClient = npgmqClient;
     }
@@ -62,15 +67,13 @@ public async Task StartAsync(CancellationToken token = default)
         }
         catch (NpgmqException e)
         {
-            // This is possibly a race with another extension initializing in parallel.
-            // Check if the extension is already present.
-            try
+            if (e.InnerException is PostgresException { SqlState: PostgresErrorCodes.UniqueViolation })
             {
-                await _npgmqClient.GetPgmqVersionAsync(token);
+                _logger.LogWarning(e, "pgmq extension already exists, assuming it was created by another instance");
             }
-            catch (NpgmqException)
+            else
             {
-                throw e;
+                throw;
             }
         }
 

test/Motor.Extensions.Hosting.PgMq_IntegrationTest/PgMqIntegrationTests.cs

@@ -320,7 +320,11 @@ await producer.PublishMessageAsync(
         Assert.NotNull(requeued);
     }
 
-    private PgMqMessageProducer<T> GetProducer<T>(string queueName, CloudEventFormat cloudEventFormat)
+    private PgMqMessageProducer<T> GetProducer<T>(
+        string queueName,
+        CloudEventFormat cloudEventFormat,
+        ILogger<PgMqMessageProducer<T>>? logger = null
+    )
         where T : notnull
     {
         var options = BuildPgOptions(
@@ -334,8 +338,10 @@ private PgMqMessageProducer<T> GetProducer<T>(string queueName, CloudEventFormat
             }
         );
         var producerOptions = MSOptions.Create(new PublisherOptions { CloudEventFormat = cloudEventFormat });
+        logger ??= Mock.Of<ILogger<PgMqMessageProducer<T>>>();
         return new PgMqMessageProducer<T>(
             MSOptions.Create(options),
+            logger,
             producerOptions,
             new NpgmqClient(options.ToConnectionString())
         );