test

Farfetch · Nov 22, 2023 · 6729912 · 6729912
1 parent a45ea76
commit 6729912
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 2 deletions.
diff --git a/src/KafkaFlow/Consumers/ConsumerWorker.cs b/src/KafkaFlow/Consumers/ConsumerWorker.cs
@@ -127,7 +127,7 @@ private async Task ProcessMessageAsync(ConsumeResult<byte[], byte[]> message, Ca
                         message.TopicPartitionOffset,
                         () => scope.Dispose());
 
-                    await this.globalEvents.FireMessageConsumeStartedAsync(new MessageEventContext(context));
+                    this.globalEvents.FireMessageConsumeStarted(new MessageEventContext(context));
 
                     await this.middlewareExecutor
                         .Execute(scope.Resolver, context, _ => Task.CompletedTask)

diff --git a/src/KafkaFlow/Event.cs b/src/KafkaFlow/Event.cs
@@ -29,10 +29,14 @@ public IEventSubscription Subscribe(Action<TArg> handler)
 
         internal Task FireAsync(TArg arg)
         {
-            this.InvokeSyncHandlers(arg);
             return this.InvokeAsyncHandlers(arg);
         }
 
+        internal void Fire(TArg arg)
+        {
+            this.InvokeSyncHandlers(arg);
+        }
+
         private IEventSubscription Subscribe<T>(List<T> handlersList, T handler)
         {
             if (!handlersList.Contains(handler))

diff --git a/src/KafkaFlow/GlobalEvents.cs b/src/KafkaFlow/GlobalEvents.cs
@@ -37,6 +37,9 @@ public GlobalEvents(ILogHandler log)
         public Task FireMessageConsumeStartedAsync(MessageEventContext context)
             => this.messageConsumeStarted.FireAsync(context);
 
+        public void FireMessageConsumeStarted(MessageEventContext context)
+            => this.messageConsumeStarted.Fire(context);
+
         public Task FireMessageConsumeErrorAsync(MessageErrorEventContext context)
             => this.messageConsumeError.FireAsync(context);