set consumer timeout

stormaref · Apr 17, 2024 · 61e9c93 · 61e9c93
1 parent 8bc9c64
commit 61e9c93
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 2 deletions.
diff --git a/KafkaStorm.Test/TestBase.cs b/KafkaStorm.Test/TestBase.cs
@@ -39,6 +39,8 @@ private IServiceCollection ConfigureServices()
 
             factory.AddConsumers(crf =>
             {
+                crf.SetConsumingPeriod(5);
+
                 crf.AddConsumer<HelloConsumer, HelloEvent>(
                     new ConsumerConfig {BootstrapServers = "localhost:29092", GroupId = "TestGroup"}, "my-topic");
             });

diff --git a/KafkaStorm/KafkaStorm.csproj b/KafkaStorm/KafkaStorm.csproj
@@ -5,7 +5,7 @@
         <Nullable>enable</Nullable>
         <LangVersion>latest</LangVersion>
         <Authors>ArefAzizian</Authors>
-        <Version>8.0.4</Version>
+        <Version>8.0.5</Version>
         <PackageId>KafKaStorm</PackageId>
         <LangVersion>latest</LangVersion>
         <GeneratePackageOnBuild>true</GeneratePackageOnBuild>

diff --git a/KafkaStorm/Registration/ConsumerRegistrationFactory.cs b/KafkaStorm/Registration/ConsumerRegistrationFactory.cs
@@ -11,6 +11,7 @@ public class ConsumerRegistrationFactory
 {
     internal static Dictionary<string, ConsumerConfig> ConsumerConfigs = null!;
     internal static Dictionary<string, string> ConsumerTopics = null!;
+    internal static int ConsumingPeriod = 10;
     private readonly IServiceCollection _serviceCollection;
 
     public ConsumerRegistrationFactory(IServiceCollection serviceCollection)
@@ -42,4 +43,13 @@ public void AddConsumer<TConsumer, TMessage>(ConsumerConfig config, string? topi
         _serviceCollection.AddTransient<IConsumer<TMessage>, TConsumer>();
         _serviceCollection.AddHostedService<ConsumerHostedService<TMessage>>();
     }
+
+    /// <summary>
+    /// Set consuming timeout (period for checking for new messages)
+    /// </summary>
+    /// <param name="period">period in milliseconds</param>
+    public void SetConsumingPeriod(int period = 10)
+    {
+        ConsumingPeriod = period;
+    }
 }
diff --git a/KafkaStorm/Services/ConsumerHostedService.cs b/KafkaStorm/Services/ConsumerHostedService.cs
@@ -50,7 +50,7 @@ public Task StartAsync(CancellationToken cancellationToken)
     private void Handle(CancellationToken cancellationToken)
     {
         if (cancellationToken.IsCancellationRequested) return;
-        var result = _consumer.Consume(1);
+        var result = _consumer.Consume(ConsumerRegistrationFactory.ConsumingPeriod);
         var message = JsonSerializer.Deserialize<TMessage>(result.Message.Value) ??
                       throw new MessageNullException<TMessage>();
         _myConsumer.Handle(message, cancellationToken);