Merge pull request #49 from neuroglia-io/fix-subscription-filter

Fixed the SubscriptionHandler
neuroglia-io · Jul 31, 2023 · acf5db2 · acf5db2
2 parents 618daf6 + 14789b8
commit acf5db2
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/src/broker/CloudStreams.Broker.Application/Services/SubscriptionHandler.cs b/src/broker/CloudStreams.Broker.Application/Services/SubscriptionHandler.cs
@@ -273,7 +273,7 @@ protected virtual bool Filters(CloudEventRecord e, IDictionary<string, string?>
     {
         if (e == null) throw new ArgumentNullException(nameof(e));
         if (attributeFilters == null) throw new ArgumentNullException(nameof(attributeFilters));
-        var attributes = e.ToDictionary()!.ToDictionary(kvp => kvp.Key, kvp => kvp.Value.ToString());
+        var attributes = e.Metadata.ContextAttributes.ToDictionary(kvp => kvp.Key, kvp => kvp.Value.ToString());
         foreach (var attributeFilter in attributeFilters)
         {
             if (!attributes.TryGetValue(attributeFilter.Key, out var attributeValue) || string.IsNullOrWhiteSpace(attributeValue)) return false;
@@ -374,6 +374,7 @@ protected virtual async Task DispatchAsync(CloudEventRecord e, bool retryOnError
     {
         if (e == null) throw new ArgumentNullException(nameof(e));
         var cloudEvent = e.ToCloudEvent(this.Broker.Resource.Spec.Dispatch?.Sequencing);
+        if (!this.Filters(e)) return;
         cloudEvent = await this.MutateAsync(cloudEvent).ConfigureAwait(false);
         await this.DispatchAsync(cloudEvent, e.Sequence, retryOnError, catchUpWhenAvailable).ConfigureAwait(false);
     }