diff --git a/src/OpenFeature/EventExecutor.cs b/src/OpenFeature/EventExecutor.cs index ad53a949..b54bd9c0 100644 --- a/src/OpenFeature/EventExecutor.cs +++ b/src/OpenFeature/EventExecutor.cs @@ -1,6 +1,5 @@ using System; using System.Collections.Generic; -using System.Threading; using System.Threading.Channels; using System.Threading.Tasks; using Microsoft.Extensions.Logging; @@ -10,26 +9,23 @@ namespace OpenFeature { - internal delegate Task ShutdownDelegate(CancellationToken cancellationToken); - internal sealed partial class EventExecutor : IAsyncDisposable { - private readonly object _lockObj = new object(); + private readonly object _lockObj = new(); public readonly Channel EventChannel = Channel.CreateBounded(1); private FeatureProvider? _defaultProvider; - private readonly Dictionary _namedProviderReferences = new Dictionary(); - private readonly List _activeSubscriptions = new List(); + private readonly Dictionary _namedProviderReferences = []; + private readonly List _activeSubscriptions = []; - private readonly Dictionary> _apiHandlers = new Dictionary>(); - private readonly Dictionary>> _clientHandlers = new Dictionary>>(); + private readonly Dictionary> _apiHandlers = []; + private readonly Dictionary>> _clientHandlers = []; private ILogger _logger; public EventExecutor() { this._logger = NullLogger.Instance; - var eventProcessing = new Thread(this.ProcessEventAsync); - eventProcessing.Start(); + Task.Run(this.ProcessEventAsync); } public ValueTask DisposeAsync() => new(this.ShutdownAsync()); @@ -42,7 +38,7 @@ internal void AddApiLevelHandler(ProviderEventTypes eventType, EventHandlerDeleg { if (!this._apiHandlers.TryGetValue(eventType, out var eventHandlers)) { - eventHandlers = new List(); + eventHandlers = []; this._apiHandlers[eventType] = eventHandlers; } @@ -70,13 +66,13 @@ internal void AddClientHandler(string client, ProviderEventTypes eventType, Even // check if there is already a list of handlers for the given client and event type if (!this._clientHandlers.TryGetValue(client, out var registry)) { - registry = new Dictionary>(); + registry = []; this._clientHandlers[client] = registry; } if (!this._clientHandlers[client].TryGetValue(eventType, out var eventHandlers)) { - eventHandlers = new List(); + eventHandlers = []; this._clientHandlers[client][eventType] = eventHandlers; } @@ -127,16 +123,15 @@ internal void RegisterClientFeatureProvider(string client, FeatureProvider? prov } lock (this._lockObj) { - var newProvider = provider; FeatureProvider? oldProvider = null; if (this._namedProviderReferences.TryGetValue(client, out var foundOldProvider)) { oldProvider = foundOldProvider; } - this._namedProviderReferences[client] = newProvider; + this._namedProviderReferences[client] = provider; - this.StartListeningAndShutdownOld(newProvider, oldProvider); + this.StartListeningAndShutdownOld(provider, oldProvider); } } @@ -146,8 +141,7 @@ private void StartListeningAndShutdownOld(FeatureProvider newProvider, FeaturePr if (!this.IsProviderActive(newProvider)) { this._activeSubscriptions.Add(newProvider); - var featureProviderEventProcessing = new Thread(this.ProcessFeatureProviderEventsAsync); - featureProviderEventProcessing.Start(newProvider); + Task.Run(() => this.ProcessFeatureProviderEventsAsync(newProvider)); } if (oldProvider != null && !this.IsProviderBound(oldProvider)) @@ -186,42 +180,37 @@ private void EmitOnRegistration(FeatureProvider? provider, ProviderEventTypes ev } var status = provider.Status; - var message = ""; - if (status == ProviderStatus.Ready && eventType == ProviderEventTypes.ProviderReady) - { - message = "Provider is ready"; - } - else if (status == ProviderStatus.Error && eventType == ProviderEventTypes.ProviderError) + var message = status switch { - message = "Provider is in error state"; - } - else if (status == ProviderStatus.Stale && eventType == ProviderEventTypes.ProviderStale) + ProviderStatus.Ready when eventType == ProviderEventTypes.ProviderReady => "Provider is ready", + ProviderStatus.Error when eventType == ProviderEventTypes.ProviderError => "Provider is in error state", + ProviderStatus.Stale when eventType == ProviderEventTypes.ProviderStale => "Provider is in stale state", + _ => string.Empty + }; + + if (string.IsNullOrWhiteSpace(message)) { - message = "Provider is in stale state"; + return; } - if (message != "") + try { - try + handler.Invoke(new ProviderEventPayload { - handler.Invoke(new ProviderEventPayload - { - ProviderName = provider.GetMetadata()?.Name, - Type = eventType, - Message = message - }); - } - catch (Exception exc) - { - this.ErrorRunningHandler(exc); - } + ProviderName = provider.GetMetadata()?.Name, + Type = eventType, + Message = message + }); + } + catch (Exception exc) + { + this.ErrorRunningHandler(exc); } } - private async void ProcessFeatureProviderEventsAsync(object? providerRef) + private async Task ProcessFeatureProviderEventsAsync(FeatureProvider provider) { - var typedProviderRef = (FeatureProvider?)providerRef; - if (typedProviderRef?.GetEventChannel() is not { Reader: { } reader }) + if (provider.GetEventChannel() is not { Reader: { } reader }) { return; } @@ -234,82 +223,92 @@ private async void ProcessFeatureProviderEventsAsync(object? providerRef) switch (item) { case ProviderEventPayload eventPayload: - this.UpdateProviderStatus(typedProviderRef, eventPayload); - await this.EventChannel.Writer.WriteAsync(new Event { Provider = typedProviderRef, EventPayload = eventPayload }).ConfigureAwait(false); + UpdateProviderStatus(provider, eventPayload); + await this.EventChannel.Writer.WriteAsync(new Event { Provider = provider, EventPayload = eventPayload }).ConfigureAwait(false); break; } } } // Method to process events - private async void ProcessEventAsync() + private async Task ProcessEventAsync() { while (await this.EventChannel.Reader.WaitToReadAsync().ConfigureAwait(false)) { if (!this.EventChannel.Reader.TryRead(out var item)) + { continue; + } - switch (item) + if (item is not Event e) { - case Event e: - lock (this._lockObj) - { - if (e.EventPayload?.Type != null && this._apiHandlers.TryGetValue(e.EventPayload.Type, out var eventHandlers)) - { - foreach (var eventHandler in eventHandlers) - { - this.InvokeEventHandler(eventHandler, e); - } - } - - // look for client handlers and call invoke method there - foreach (var keyAndValue in this._namedProviderReferences) - { - if (keyAndValue.Value == e.Provider && keyAndValue.Key != null) - { - if (this._clientHandlers.TryGetValue(keyAndValue.Key, out var clientRegistry)) - { - if (e.EventPayload?.Type != null && clientRegistry.TryGetValue(e.EventPayload.Type, out var clientEventHandlers)) - { - foreach (var eventHandler in clientEventHandlers) - { - this.InvokeEventHandler(eventHandler, e); - } - } - } - } - } - - if (e.Provider != this._defaultProvider) - { - break; - } - // handling the default provider - invoke event handlers for clients which are not bound - // to a particular feature provider - foreach (var keyAndValues in this._clientHandlers) - { - if (this._namedProviderReferences.TryGetValue(keyAndValues.Key, out _)) - { - // if there is an association for the client to a specific feature provider, then continue - continue; - } - if (e.EventPayload?.Type != null && keyAndValues.Value.TryGetValue(e.EventPayload.Type, out var clientEventHandlers)) - { - foreach (var eventHandler in clientEventHandlers) - { - this.InvokeEventHandler(eventHandler, e); - } - } - } - } - break; + continue; + } + + lock (this._lockObj) + { + this.ProcessApiHandlers(e); + this.ProcessClientHandlers(e); + this.ProcessDefaultProviderHandlers(e); + } + } + } + + private void ProcessApiHandlers(Event e) + { + if (e.EventPayload?.Type != null && this._apiHandlers.TryGetValue(e.EventPayload.Type, out var eventHandlers)) + { + foreach (var eventHandler in eventHandlers) + { + this.InvokeEventHandler(eventHandler, e); + } + } + } + + private void ProcessClientHandlers(Event e) + { + foreach (var keyAndValue in this._namedProviderReferences) + { + if (keyAndValue.Value == e.Provider + && this._clientHandlers.TryGetValue(keyAndValue.Key, out var clientRegistry) + && e.EventPayload?.Type != null + && clientRegistry.TryGetValue(e.EventPayload.Type, out var clientEventHandlers)) + { + foreach (var eventHandler in clientEventHandlers) + { + this.InvokeEventHandler(eventHandler, e); + } + } + } + } + + private void ProcessDefaultProviderHandlers(Event e) + { + if (e.Provider != this._defaultProvider) + { + return; + } + + foreach (var keyAndValues in this._clientHandlers) + { + if (this._namedProviderReferences.ContainsKey(keyAndValues.Key)) + { + continue; } + if (e.EventPayload?.Type != null && keyAndValues.Value.TryGetValue(e.EventPayload.Type, out var clientEventHandlers)) + { + foreach (var eventHandler in clientEventHandlers) + { + this.InvokeEventHandler(eventHandler, e); + } + } } } + // map events to provider status as per spec: https://openfeature.dev/specification/sections/events/#requirement-535 - private void UpdateProviderStatus(FeatureProvider provider, ProviderEventPayload eventPayload) + private static void UpdateProviderStatus(FeatureProvider provider, ProviderEventPayload eventPayload) { switch (eventPayload.Type) { diff --git a/src/OpenFeature/Providers/Memory/InMemoryProvider.cs b/src/OpenFeature/Providers/Memory/InMemoryProvider.cs index 3283ea22..4a06dc85 100644 --- a/src/OpenFeature/Providers/Memory/InMemoryProvider.cs +++ b/src/OpenFeature/Providers/Memory/InMemoryProvider.cs @@ -65,6 +65,7 @@ public async Task UpdateFlagsAsync(IDictionary? flags = null) FlagsChanged = changed, // emit all Message = "flags changed", }; + await this.EventChannel.Writer.WriteAsync(@event).ConfigureAwait(false); } diff --git a/test/OpenFeature.Tests/TestImplementations.cs b/test/OpenFeature.Tests/TestImplementations.cs index 724278e8..df738efe 100644 --- a/test/OpenFeature.Tests/TestImplementations.cs +++ b/test/OpenFeature.Tests/TestImplementations.cs @@ -152,10 +152,5 @@ internal ValueTask SendEventAsync(ProviderEventTypes eventType, CancellationToke { return this.EventChannel.Writer.WriteAsync(new ProviderEventPayload { Type = eventType, ProviderName = this.GetMetadata().Name, }, cancellationToken); } - - internal ValueTask SendEventAsync(ProviderEventPayload payload, CancellationToken cancellationToken = default) - { - return this.EventChannel.Writer.WriteAsync(payload, cancellationToken); - } } }