From 36c66b9ed92c09f558a027a0dc642aae8d05b72e Mon Sep 17 00:00:00 2001 From: Stuart Ferguson Date: Thu, 26 Feb 2026 22:43:06 +0000 Subject: [PATCH] Refactor PersistentSubscription for better observability EventAppeared is now an instance method, enabling access to instance members. Added EventHandlerMissing event for improved traceability when no handlers are configured. Updated constructor to use instance method, removed region comments, and enhanced logging. These changes make the class more object-oriented and easier to monitor. --- .../PersistentSubscription.cs | 122 +++++++----------- 1 file changed, 49 insertions(+), 73 deletions(-) diff --git a/Shared.EventStore/SubscriptionWorker/PersistentSubscription.cs b/Shared.EventStore/SubscriptionWorker/PersistentSubscription.cs index 0b07f93d..1530dfcb 100644 --- a/Shared.EventStore/SubscriptionWorker/PersistentSubscription.cs +++ b/Shared.EventStore/SubscriptionWorker/PersistentSubscription.cs @@ -1,4 +1,5 @@ -using KurrentDB.Client; +using System.Runtime.CompilerServices; +using KurrentDB.Client; using SimpleResults; namespace Shared.EventStore.SubscriptionWorker; @@ -21,15 +22,11 @@ namespace Shared.EventStore.SubscriptionWorker; [ExcludeFromCodeCoverage] public class PersistentSubscription { - #region Fields - public EventHandler SubscriptionHasDropped; - private readonly Func> Subscribe; - - #endregion + public EventHandler EventHandlerMissing; - #region Constructors + private readonly Func> Subscribe; private void SubscriptionDropped(KurrentDB.Client.PersistentSubscription arg1, SubscriptionDroppedReason arg2, @@ -41,8 +38,7 @@ public void SubscriptionDropped(String reason) this.Connected = false; Logger.Logger.LogWarning($"Subscription dropped - {reason}"); - if (this.SubscriptionHasDropped != null) - { + if (this.SubscriptionHasDropped != null) { //Broadcast to owner this.SubscriptionHasDropped(this, reason); } @@ -58,7 +54,7 @@ private PersistentSubscription(IPersistentSubscriptionsClient persistentSubscrip this.PersistentSubscriptionDetails = persistentSubscriptionDetails; UserCredentials userCredentials = new(username, password); - Func< KurrentDB.Client.PersistentSubscription, ResolvedEvent, Int32?, CancellationToken, Task> eventAppeared = (ps, re, retryCount, ct) => PersistentSubscription.EventAppeared(ps, re, retryCount, domainEventHandlerResolver, ct); + Func< KurrentDB.Client.PersistentSubscription, ResolvedEvent, Int32?, CancellationToken, Task> eventAppeared = (ps, re, retryCount, ct) => this.EventAppeared(ps, re, retryCount, domainEventHandlerResolver, ct); this.Subscribe = ct => persistentSubscriptionsClient.SubscribeAsync(this.PersistentSubscriptionDetails.StreamName, this.PersistentSubscriptionDetails.GroupName, @@ -69,29 +65,19 @@ private PersistentSubscription(IPersistentSubscriptionsClient persistentSubscrip ct); } - #endregion - - #region Properties - public Boolean Connected { get; private set; } public KurrentDB.Client.PersistentSubscription KurrentDbPersistentSubscription { get; private set; } public IPersistentSubscriptionsClient PersistentSubscriptionsClient { get; } public PersistentSubscriptionDetails PersistentSubscriptionDetails { get; } - - #endregion - - #region Methods - + public async Task ConnectToSubscription(CancellationToken cancellationToken) { - try - { + try { this.KurrentDbPersistentSubscription = await this.Subscribe(cancellationToken); this.Connected = true; } - catch (Exception e) - { + catch (Exception e) { Logger.Logger.LogError(e); } } @@ -131,65 +117,55 @@ private static TenantIdentifiers GetTenantIdentifiersFromDomainEvent(IDomainEven } } - internal static async Task EventAppeared(KurrentDB.Client.PersistentSubscription persistentSubscription, + internal async Task EventAppeared(KurrentDB.Client.PersistentSubscription persistentSubscription, ResolvedEvent resolvedEvent, Int32? retryCount, IDomainEventHandlerResolver domainEventHandlerResolver, CancellationToken cancellationToken) { - try - { - using (CancellationTokenSource cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken)) - { - if (resolvedEvent.SilentlyHandleEvent()) - { - await PersistentSubscriptionsHelper.AckEvent(persistentSubscription, resolvedEvent); - return; - } - IDomainEvent domainEvent = TypeMapConvertor.Convertor(resolvedEvent); - TenantIdentifiers tenantIdentifiers = PersistentSubscription.GetTenantIdentifiersFromDomainEvent(domainEvent); - Boolean.TryParse(ConfigurationReader.GetValueOrDefault("AppSettings","LogsPerTenantEnabled", "false"), out Boolean logPerTenantEnabled); - - TenantContext tenantContext = new(); - tenantContext.Initialise(tenantIdentifiers, logPerTenantEnabled); - TenantContext.CurrentTenant = tenantContext; - - Logger.Logger.LogInformation( - $"EventAppearedFromPersistentSubscription with Event Id {resolvedEvent.Event.EventId} event type {resolvedEvent.Event.EventType}"); - - Result> domainEventHandlersResult = - domainEventHandlerResolver.GetDomainEventHandlers(domainEvent); - - if (domainEventHandlersResult.IsFailed) - { - // Log a warning out - Logger.Logger.LogWarning( - $"No event handlers configured for Event Type [{domainEvent.GetType().Name}]"); - await PersistentSubscriptionsHelper.AckEvent(persistentSubscription, resolvedEvent); - return; - } - - List domainEventHandlers = domainEventHandlersResult.Data; - - Result result = await domainEvent.DispatchToHandlers(domainEventHandlers, cts.Token); - if (result.IsSuccess) { - await PersistentSubscriptionsHelper.AckEvent(persistentSubscription, resolvedEvent); - } - else { - Exception ex = new($"Failed to process the event type {resolvedEvent.Event.EventType} {resolvedEvent.GetResolvedEventDataAsString()} Result was {result.Message}"); - Logger.Logger.LogError(ex); - } + try { + using CancellationTokenSource cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + if (resolvedEvent.SilentlyHandleEvent()) { + await PersistentSubscriptionsHelper.AckEvent(persistentSubscription, resolvedEvent); + return; + } + + IDomainEvent domainEvent = TypeMapConvertor.Convertor(resolvedEvent); + TenantIdentifiers tenantIdentifiers = PersistentSubscription.GetTenantIdentifiersFromDomainEvent(domainEvent); + Boolean.TryParse(ConfigurationReader.GetValueOrDefault("AppSettings", "LogsPerTenantEnabled", "false"), out Boolean logPerTenantEnabled); + + TenantContext tenantContext = new(); + tenantContext.Initialise(tenantIdentifiers, logPerTenantEnabled); + TenantContext.CurrentTenant = tenantContext; + + Logger.Logger.LogInformation($"EventAppearedFromPersistentSubscription with Event Id {resolvedEvent.Event.EventId} event type {resolvedEvent.Event.EventType}"); + + Result> domainEventHandlersResult = domainEventHandlerResolver.GetDomainEventHandlers(domainEvent); + + if (domainEventHandlersResult.IsFailed) { + // Log a line of trace out + if (this.EventHandlerMissing != null) + this.EventHandlerMissing(this, $"No event handlers configured for Event Type [{domainEvent.GetType().Name}]"); + + await PersistentSubscriptionsHelper.AckEvent(persistentSubscription, resolvedEvent); + return; + } + + List domainEventHandlers = domainEventHandlersResult.Data; + + Result result = await domainEvent.DispatchToHandlers(domainEventHandlers, cts.Token); + if (result.IsSuccess) { + await PersistentSubscriptionsHelper.AckEvent(persistentSubscription, resolvedEvent); + } + else { + Exception ex = new($"Failed to process the event type {resolvedEvent.Event.EventType} {resolvedEvent.GetResolvedEventDataAsString()} Result was {result.Message}"); + Logger.Logger.LogError(ex); } } - catch (Exception e) - { + catch (Exception e) { Exception ex = new($"Failed to process the event type {resolvedEvent.Event.EventType} {resolvedEvent.GetResolvedEventDataAsString()}", e); - + Logger.Logger.LogError(ex); } } - - - - #endregion } \ No newline at end of file