GlobalEvents tests. Not finished yet

Farfetch · Oct 10, 2023 · 4cffda0 · 4cffda0
1 parent 1cb2e06
commit 4cffda0
Show file tree

Hide file tree

Showing 6 changed files with 72 additions and 3 deletions.
diff --git a/src/KafkaFlow.Admin.Dashboard/Properties/launchSettings.json b/src/KafkaFlow.Admin.Dashboard/Properties/launchSettings.json
@@ -0,0 +1,12 @@
+{
+  "profiles": {
+    "KafkaFlow.Admin.Dashboard": {
+      "commandName": "Project",
+      "launchBrowser": true,
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      },
+      "applicationUrl": "https://localhost:52566;http://localhost:52567"
+    }
+  }
+}
diff --git a/src/KafkaFlow.IntegrationTests/ConsumerTest.cs b/src/KafkaFlow.IntegrationTests/ConsumerTest.cs
@@ -22,6 +22,8 @@ public class ConsumerTest
         [TestInitialize]
         public void Setup()
         {
+            Bootstrapper.GlobalEvents = observers => { };
+
             this.provider = Bootstrapper.GetServiceProvider();
             MessageStorage.Clear();
         }

diff --git a/src/KafkaFlow.IntegrationTests/Core/Bootstrapper.cs b/src/KafkaFlow.IntegrationTests/Core/Bootstrapper.cs
@@ -3,6 +3,7 @@ namespace KafkaFlow.IntegrationTests.Core
     using System;
     using System.IO;
     using System.Threading;
+    using System.Threading.Tasks;
     using Confluent.Kafka;
     using Confluent.SchemaRegistry;
     using Confluent.SchemaRegistry.Serdes;
@@ -11,6 +12,7 @@ namespace KafkaFlow.IntegrationTests.Core
     using global::Microsoft.Extensions.Hosting;
     using KafkaFlow.Compressor;
     using KafkaFlow.Compressor.Gzip;
+    using KafkaFlow.Configuration;
     using KafkaFlow.IntegrationTests.Core.Handlers;
     using KafkaFlow.IntegrationTests.Core.Messages;
     using KafkaFlow.IntegrationTests.Core.Middlewares;
@@ -42,9 +44,28 @@ internal static class Bootstrapper
         private const string ProtobufGzipTopicName = "test-protobuf-gzip";
         private const string ProtobufGzipTopicName2 = "test-protobuf-gzip-2";
         private const string AvroTopicName = "test-avro";
+        private static Action<IGlobalEvents> globalEvents;
 
         private static readonly Lazy<IServiceProvider> LazyProvider = new(SetupProvider);
 
+        public static Action<IGlobalEvents> GlobalEvents
+        {
+            get
+            {
+                if(globalEvents == null)
+                {
+                    return _ => { };
+                }
+
+                return globalEvents;
+            }
+
+            set
+            {
+                globalEvents = value;
+            }
+        }
+
         public static IServiceProvider GetServiceProvider() => LazyProvider.Value;
 
         private static IServiceProvider SetupProvider()
@@ -87,7 +108,7 @@ private static void SetupServices(HostBuilderContext context, IServiceCollection
 
             ConsumerConfig defaultConfig = new()
             {
-                Acks = Confluent.Kafka.Acks.All,
+                Acks = Acks.All,
                 AllowAutoCreateTopics = false,
                 AutoCommitIntervalMs = 5000,
                 AutoOffsetReset = Confluent.Kafka.AutoOffsetReset.Latest,
@@ -332,7 +353,8 @@ private static void SetupServices(HostBuilderContext context, IServiceCollection
                                     .DefaultTopic(GzipTopicName)
                                     .AddMiddlewares(
                                         middlewares => middlewares
-                                            .AddCompressor<GzipMessageCompressor>()))));
+                                            .AddCompressor<GzipMessageCompressor>())))
+                    .SubscribeGlobalEvents(GlobalEvents));
 
             services.AddSingleton<JsonProducer>();
             services.AddSingleton<JsonGzipProducer>();

diff --git a/src/KafkaFlow.IntegrationTests/ProducerTest.cs b/src/KafkaFlow.IntegrationTests/ProducerTest.cs
@@ -1,10 +1,13 @@
 namespace KafkaFlow.IntegrationTests
 {
     using System;
+    using System.Linq;
     using System.Threading.Tasks;
     using AutoFixture;
+    using KafkaFlow.Consumers;
     using KafkaFlow.IntegrationTests.Core;
     using KafkaFlow.IntegrationTests.Core.Handlers;
+    using KafkaFlow.IntegrationTests.Core.Messages;
     using KafkaFlow.IntegrationTests.Core.Producers;
     using Microsoft.Extensions.DependencyInjection;
     using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -16,9 +19,19 @@ public class ProducerTest
 
         private IServiceProvider provider;
 
+        private IMessageContext messageContext;
+
         [TestInitialize]
         public void Setup()
         {
+            Bootstrapper.GlobalEvents = observers =>
+            {
+                observers.MessageProduceStarted.Subscribe(eventContext =>
+                {
+                    this.messageContext = eventContext.MessageContext;
+                    return Task.CompletedTask;
+                });
+            };
             this.provider = Bootstrapper.GetServiceProvider();
             MessageStorage.Clear();
         }
@@ -36,5 +49,22 @@ public async Task ProduceNullKeyTest()
             // Assert
             await MessageStorage.AssertMessageAsync(message);
         }
+
+        [TestMethod]
+        public void SubscribeGlobalEventsTest()
+        {
+            var consumer = this.provider.GetRequiredService<IConsumerAccessor>().All.First();
+            var producer = this.provider.GetRequiredService<IMessageProducer<ProtobufGzipProducer>>();
+            var messages = this.fixture.CreateMany<TestMessage1>(1).ToList();
+
+            // Act
+            messages.ForEach(m => producer.Produce(m.Id.ToString(), m));
+
+            var x = messages.FirstOrDefault();
+
+            // Assert
+            Assert.IsNotNull(this.messageContext);
+            Assert.AreEqual(this.messageContext.Message.Key, messages.FirstOrDefault().Id.ToString());
+        }
     }
 }
diff --git a/src/KafkaFlow/Configuration/KafkaConfigurationBuilder.cs b/src/KafkaFlow/Configuration/KafkaConfigurationBuilder.cs
@@ -47,7 +47,7 @@ public KafkaConfiguration Build()
             .AddSingleton<IConsumerAccessor>(new ConsumerAccessor())
             .AddSingleton<IConsumerManagerFactory>(new ConsumerManagerFactory())
             .AddSingleton<IClusterManagerAccessor, ClusterManagerAccessor>()
-            .AddSingleton<IGlobalEvents>(r =>
+            .AddSingleton(r =>
             {
                 var logHandler = r.Resolve<ILogHandler>();
 

diff --git a/src/KafkaFlow/Consumers/ConsumerWorker.cs b/src/KafkaFlow/Consumers/ConsumerWorker.cs
@@ -41,6 +41,9 @@ public ConsumerWorker(
 
             middlewareContext.Worker = this;
             middlewareContext.Consumer = consumer;
+            this.workerStoppingEvent = new Event(logHandler);
+            this.workerStoppedEvent = new Event(logHandler);
+            this.workerProcessingEnded = new Event<IMessageContext>(logHandler);
         }
 
         public int Id { get; }