fix: don't use CancellationToken for kafka message handler

Informatievlaanderen · Dec 5, 2024 · 47482e2 · 47482e2
1 parent c32fd65
commit 47482e2
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 10 deletions.
diff --git a/src/ParcelRegistry.Consumer.Address.Console/appsettings.json b/src/ParcelRegistry.Consumer.Address.Console/appsettings.json
@@ -17,8 +17,6 @@
 
   "GroupSuffix": "",
   "AddressTopic": "dev.address",
-  "AddressTopicOffset": "",
-  "IgnoreAddressTopicOffsetDataCheck": false,
 
   "BaseUrl": "https://api.staging-basisregisters.vlaanderen/",
 

diff --git a/src/ParcelRegistry.Consumer.Address/BackOfficeConsumer.cs b/src/ParcelRegistry.Consumer.Address/BackOfficeConsumer.cs
@@ -57,14 +57,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
             {
                 await _kafkaIdemIdompotencyConsumer.ConsumeContinuously(async (message, context) =>
                 {
-                    _logger.LogInformation("Handling next message");
-
-                    await commandHandlingProjector.ProjectAsync(commandHandler, message, stoppingToken).ConfigureAwait(false);
-                    await backOfficeProjector.ProjectAsync(context, message, stoppingToken).ConfigureAwait(false);
-
-                    //CancellationToken.None to prevent halfway consumption
-                    await context.SaveChangesAsync(CancellationToken.None);
-
+                    await ConsumeHandler(commandHandlingProjector, backOfficeProjector, commandHandler, message, context);
                 }, stoppingToken);
             }
             catch (Exception)
@@ -73,5 +66,20 @@ await _kafkaIdemIdompotencyConsumer.ConsumeContinuously(async (message, context)
                 throw;
             }
         }
+
+        private async Task ConsumeHandler(
+            ConnectedProjector<CommandHandler> commandHandlingProjector,
+            ConnectedProjector<ConsumerAddressContext> backOfficeProjector,
+            CommandHandler commandHandler,
+            object message,
+            ConsumerAddressContext context)
+        {
+            _logger.LogInformation("Handling next message");
+
+            await commandHandlingProjector.ProjectAsync(commandHandler, message, CancellationToken.None).ConfigureAwait(false);
+            await backOfficeProjector.ProjectAsync(context, message, CancellationToken.None).ConfigureAwait(false);
+
+            await context.SaveChangesAsync(CancellationToken.None);
+        }
     }
 }