Include EventHub and option to SubscribeGlobalEvents #445

simaoribeiro · 2023-10-04T15:41:19Z

Description

Includes EventHub and option to SubscribeGlobalEvents in KafkaConfigurationBuilder

Checklist

My code follows the style guidelines of this project
I have performed a self-review of my own code
I have added tests to cover my changes
I have made corresponding changes to the documentation

fix: change dependency injection lifetime management for worker and consumer/producer fix: create AdminClient without authentication perf: memory optimizations on MessageContext

filipeesch · 2023-10-04T15:46:11Z

src/KafkaFlow.Abstractions/MessageEventContext.cs

+    {
+        public IMessageContext MessageContext { get; set; }
+
+        public IDependencyResolver DependencyResolver { get; set; }


no setters and initialization from ctor. Or just use a record, now we can :)

I tried to use a record but my VS kept saying that I couldn't use C#9 features in .netstandard2.0 projects 🤔

Done, please review

filipeesch · 2023-10-04T15:47:43Z

src/KafkaFlow/Consumers/ConsumerWorker.cs

            var middlewareContext = this.workerDependencyResolverScope.Resolver.Resolve<ConsumerMiddlewareContext>();
+            this.eventHub = this.workerDependencyResolverScope.Resolver.Resolve<IEventHub>() as EventHub;


this.workerDependencyResolverScope.Resolver.Resolve<EventHub>()

Done, please review

filipeesch · 2023-10-04T15:49:55Z

src/KafkaFlow/Consumers/ConsumerWorker.cs

+                await this.eventHub.FireMessageConsumeStartedAsync(
+                    new MessageEventContext
+                    {
+                        DependencyResolver = this.workerDependencyResolverScope.Resolver,


maybe we don't need the Resolver here, since we have it in the IMessageContext

Should we remove the IDependencyResolver from the MessageEventContext class given that, currently, it only contains the IMessageContext and IDependencyResolver objects? 🤔 And only include it later if needed?

Yes, that's it 😃

I ended up keeping as it was and included a constructor with only the IMessageContext

filipeesch · 2023-10-04T15:50:31Z

src/KafkaFlow/Event.cs


        public Event(ILogHandler logHandler)
        {
            this.logHandler = logHandler;
        }

-        public IEventSubscription Subscribe(Func<TArg, Task> handler)
+        public void Subscribe(Func<TArg, Task> handler)


Why the IEventSubscription was removed?

Error in the rebase process. Solved now

src/KafkaFlow/EventHub.cs

ruiqbarbosa

Should some tests be added to cover the new SubscribeGlobalEvents configuration?

joelfoliveira · 2023-10-10T09:05:41Z

src/KafkaFlow.Abstractions/IEvent.cs

@@ -20,7 +20,7 @@ public interface IEvent
    /// Represents an Event to be subscribed.
    /// </summary>
    /// <typeparam name="TArg">The argument expected by the event.</typeparam>
-    public interface IEvent<TArg>
+    public interface IEvent<out TArg>


Why do we need a covariant generic here?

joelfoliveira · 2023-10-10T09:16:35Z

src/KafkaFlow/Configuration/KafkaConfiguration.cs

-        public void AddClusters(IEnumerable<ClusterConfiguration> configurations) =>
-            this.clusters.AddRange(configurations);
-    }
+    internal void AddClusters(IEnumerable<ClusterConfiguration> configurations) =>


Is a reason to move this to internal?

There is an issue in KafkaFlow where the AddClusters method was suggested as a workaround. This change could break those use cases.

joelfoliveira · 2023-10-10T09:22:11Z

src/KafkaFlow/Configuration/KafkaConfigurationBuilder.cs

+        return this;
+    }
+
+    public IKafkaConfigurationBuilder SubscribeGlobalEvents(Action<IGlobalEvents> builder)


Please rename the parameter to observers to be consistent with the interface.

JoaoRodriguesGithub · 2023-10-09T16:34:17Z

src/KafkaFlow.Abstractions/IConsumerContext.cs

@@ -91,7 +91,7 @@ public interface IConsumerContext
        /// By default, this method is automatically invoked when message processing concludes, unless
        /// the consumer is configured for manual message completion or the <see cref="AutoMessageCompletion"/> flag is set to false.
        /// </summary>


Please consider adding the following suggestion:

Suggested change

/// </summary>

/// </summary>

/// <param name="context">The message context</param>

JoaoRodriguesGithub · 2023-10-10T08:28:47Z

src/KafkaFlow/Consumers/ConsumerWorker.cs

@@ -4,14 +4,15 @@ namespace KafkaFlow.Consumers
    using System.Threading;
    using System.Threading.Channels;
    using System.Threading.Tasks;
+    using KafkaFlow.Configuration;


Please consider removing the unnecessary using

erik-catalao

LGTM

simaoribeiro · 2023-10-18T16:31:05Z

src/KafkaFlow/GlobalEvents.cs

+        public Task FireMessageConsumeStartedAsync(MessageEventContext context)
+            => this.messageConsumeStarted.FireAsync(context);
+
+        public Task FireMessageConsumeCompletedAsync(MessageEventContext context)


Ailton asked:
"Why not to expose the FireAsync in the IEvent interface and reduce the complexity for calling the event?"
#451 (comment)

jose-sousa-8 and others added 13 commits September 19, 2023 10:57

chore!: upgrade to .net 6

cf040de

fix!: update dashboard and client app routes to kafkaflow

4e37f85

feat!: async support on message type and schema registry resolvers

0057d79

feat: enable workers count calculation in runtime

66a47a6

fix: change dependency injection lifetime management for worker and consumer/producer fix: create AdminClient without authentication perf: memory optimizations on MessageContext

refactor: core simplification and better thread synchronization

ae2fc2e

feat: creates worker context and worker events

75a09ad

feat: creates the ConsumerLagWorkerBalancer

3b63159

refactor: rename StoreOffset to Complete

2c03fea

refactor: remove unused namespaces

49a9d67

docs: add dynamic workers documentation

356865e

feat: add event notification feature

87cd205

chore: ailton suggestions

3d43c1c

feat: include event hub

d33c769

simaoribeiro changed the base branch from master to release/3.0 October 4, 2023 15:43

filipeesch reviewed Oct 4, 2023

View reviewed changes

src/KafkaFlow/EventHub.cs Outdated Show resolved Hide resolved

code review discussions

73396aa

simaoribeiro marked this pull request as ready for review October 9, 2023 10:14

ruiqbarbosa reviewed Oct 9, 2023

View reviewed changes

ruiqbarbosa approved these changes Oct 9, 2023

View reviewed changes

simaoribeiro added 3 commits October 9, 2023 15:25

code review sugestions

5e86a51

Include FireMessageConsumeCompletedAsync in complete method

2cf0fab

Include FireMessageProduceStartedAsync in InternalProduce

1cb2e06

joelfoliveira approved these changes Oct 10, 2023

View reviewed changes

JoaoRodriguesGithub approved these changes Oct 10, 2023

View reviewed changes

GlobalEvents tests. Not finished yet

4cffda0

erik-catalao approved these changes Oct 10, 2023

View reviewed changes

simaoribeiro commented Oct 18, 2023

View reviewed changes

filipeesch force-pushed the release/3.0 branch from 87cd205 to d5a9c21 Compare October 23, 2023 12:05

simaoribeiro closed this Oct 23, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Include EventHub and option to SubscribeGlobalEvents #445

Include EventHub and option to SubscribeGlobalEvents #445

simaoribeiro commented Oct 4, 2023

filipeesch Oct 4, 2023

simaoribeiro Oct 6, 2023

simaoribeiro Oct 9, 2023

filipeesch Oct 4, 2023

simaoribeiro Oct 9, 2023

filipeesch Oct 4, 2023

simaoribeiro Oct 6, 2023

filipeesch Oct 9, 2023

simaoribeiro Oct 9, 2023

filipeesch Oct 4, 2023

simaoribeiro Oct 6, 2023

ruiqbarbosa left a comment

joelfoliveira Oct 10, 2023

joelfoliveira Oct 10, 2023

joelfoliveira Oct 10, 2023

JoaoRodriguesGithub Oct 9, 2023

JoaoRodriguesGithub Oct 10, 2023

erik-catalao left a comment

simaoribeiro Oct 18, 2023 •

edited

Loading

		var middlewareContext = this.workerDependencyResolverScope.Resolver.Resolve<ConsumerMiddlewareContext>();
		this.eventHub = this.workerDependencyResolverScope.Resolver.Resolve<IEventHub>() as EventHub;

	/// </summary>
	/// </summary>
	/// <param name="context">The message context</param>

Include EventHub and option to SubscribeGlobalEvents #445

Include EventHub and option to SubscribeGlobalEvents #445

Conversation

simaoribeiro commented Oct 4, 2023

Description

Checklist

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ruiqbarbosa left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

erik-catalao left a comment

Choose a reason for hiding this comment

simaoribeiro Oct 18, 2023 • edited Loading

Choose a reason for hiding this comment

simaoribeiro Oct 18, 2023 •

edited

Loading