Update to use interface for DI

Signed-off-by: Hal Spang <halspang@microsoft.com>

Author: halspang

src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs

@@ -114,16 +114,25 @@ public static IDurableTaskWorkerBuilder UseVersioning(this IDurableTaskWorkerBui
     /// Adds an orchestration filter to the specified <see cref="IDurableTaskWorkerBuilder"/>.
     /// </summary>
     /// <param name="builder">The builder to set the builder target for.</param>
-    /// <param name="orchestrationFilter">The filter function that determines whether an orchestration should be processed.</param>
+    /// <typeparam name="TOrchestrationFilter">The implementation of a <see cref="IOrchestrationFilter"/> that will be bound.</typeparam>
     /// <returns>The same <see cref="IDurableTaskWorkerBuilder"/> instance, allowing for method chaining.</returns>
-    public static IDurableTaskWorkerBuilder UseOrchestrationFilter(this IDurableTaskWorkerBuilder builder, Func<OrchestrationInfo, bool> orchestrationFilter)
+    public static IDurableTaskWorkerBuilder UseOrchestrationFilter<TOrchestrationFilter>(this IDurableTaskWorkerBuilder builder) where TOrchestrationFilter : class, IOrchestrationFilter
     {
         Check.NotNull(builder);
-        Check.NotNull(orchestrationFilter);
-        builder.Services.Configure<DurableTaskWorkerOptions>(builder.Name, options =>
-        {
-            options.OrchestrationFilter = orchestrationFilter;
-        });
+        builder.Services.AddSingleton<IOrchestrationFilter, TOrchestrationFilter>();
+        return builder;
+    }
+
+    /// <summary>
+    /// Adds an orchestration filter to the specified <see cref="IDurableTaskWorkerBuilder"/>.
+    /// </summary>
+    /// <param name="builder">The builder to set the builder target for.</param>
+    /// <param name="filter">The instance of an <see cref="IOrchestrationFilter"/> to use.</param>
+    /// <returns>The same <see cref="IDurableTaskWorkerBuilder"/> instance, allowing for method chaining.</returns>
+    public static IDurableTaskWorkerBuilder UseOrchestrationFilter(this IDurableTaskWorkerBuilder builder, IOrchestrationFilter filter)
+    {
+        Check.NotNull(builder);
+        builder.Services.AddSingleton(filter);
         return builder;
     }
 }

src/Worker/Core/DurableTaskWorkerOptions.cs

@@ -49,22 +49,6 @@ public enum VersionFailureStrategy
         Fail = 1,
     }
 
-    /// <summary>
-    /// Struct representation of orchestration information.
-    /// </summary>
-    public struct OrchestrationInfo
-    {
-        /// <summary>
-        /// Gets the name of the orchestration.
-        /// </summary>
-        public string Name { get; init; }
-
-        /// <summary>
-        /// Gets the tags associated with the orchestration.
-        /// </summary>
-        public Dictionary<string, string> Tags { get; init; }
-    }
-
     /// <summary>
     /// Gets or sets the data converter. Default value is <see cref="JsonDataConverter.Default" />.
     /// </summary>
@@ -161,6 +145,11 @@ public DataConverter DataConverter
     /// </summary>
     public bool IsVersioningSet { get; internal set; }
 
+    /// <summary>
+    /// Gets or sets a callback function that determines whether an orchestration should be accepted for work.
+    /// </summary>
+    public IOrchestrationFilter? OrchestrationFilter { get; set; }
+
     /// <summary>
     /// Gets a value indicating whether <see cref="DataConverter" /> was explicitly set or not.
     /// </summary>
@@ -172,10 +161,6 @@ public DataConverter DataConverter
     /// </remarks>
     internal bool DataConverterExplicitlySet { get; private set; }
 
-    /// <summary>
-    /// Gets or sets a callback function that determines whether an orchestration should be accepted for work.
-    /// </summary>
-    public Func<OrchestrationInfo, bool>? OrchestrationFilter { get; set; }
 
     /// <summary>
     /// Applies these option values to another.

src/Worker/Core/IOrchestrationFilter.cs

@@ -0,0 +1,34 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace Microsoft.DurableTask.Worker;
+
+/// <summary>
+/// Defines a filter for validating orchestrations.
+/// </summary>
+public interface IOrchestrationFilter
+{
+    /// <summary>
+    /// Validate the orchestration against the filter represented by this interface.
+    /// </summary>
+    /// <param name="info">The information on the orchestration to validate.</param>
+    /// <param name="cancellationToken">The cancellation token for the request to timeout.</param>
+    /// <returns><code>true</code> if the orchestration is valid <code>false</code> otherwise.</returns>
+    Task<bool> IsOrchestrationValidAsync(OrchestrationInfo info, CancellationToken cancellationToken = default);
+}
+
+/// <summary>
+/// Struct representation of orchestration information.
+/// </summary>
+public struct OrchestrationInfo
+{
+    /// <summary>
+    /// Gets the name of the orchestration.
+    /// </summary>
+    public string Name { get; init; }
+
+    /// <summary>
+    /// Gets the tags associated with the orchestration.
+    /// </summary>
+    public Dictionary<string, string> Tags { get; init; }
+}

src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs

@@ -13,7 +13,6 @@
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using static Microsoft.DurableTask.Protobuf.TaskHubSidecarService;
-using static Microsoft.DurableTask.Worker.DurableTaskWorkerOptions;
 using DTCore = DurableTask.Core;
 using P = Microsoft.DurableTask.Protobuf;
 
@@ -32,13 +31,15 @@ class Processor
         readonly TaskHubSidecarServiceClient client;
         readonly DurableTaskShimFactory shimFactory;
         readonly GrpcDurableTaskWorkerOptions.InternalOptions internalOptions;
+        readonly IOrchestrationFilter? orchestrationFilter = null;
 
-        public Processor(GrpcDurableTaskWorker worker, TaskHubSidecarServiceClient client)
+        public Processor(GrpcDurableTaskWorker worker, TaskHubSidecarServiceClient client, IOrchestrationFilter? orchestrationFilter = null)
         {
             this.worker = worker;
             this.client = client;
             this.shimFactory = new DurableTaskShimFactory(this.worker.grpcOptions, this.worker.loggerFactory);
             this.internalOptions = this.worker.grpcOptions.Internal;
+            this.orchestrationFilter = orchestrationFilter;
         }
 
         ILogger Logger => this.worker.logger;
@@ -376,11 +377,17 @@ async Task OnRunOrchestratorAsync(
                     entityConversionState,
                     cancellationToken);
 
-                bool filterPassed = this.worker.workerOptions.OrchestrationFilter?.Invoke(new OrchestrationInfo
+                bool filterPassed = true;
+                if (this.orchestrationFilter != null)
                 {
-                    Name = runtimeState.Name,
-                    Tags = new(runtimeState.Tags ?? ImmutableDictionary<string, string>.Empty),
-                }) ?? true;
+                    filterPassed = await this.orchestrationFilter.IsOrchestrationValidAsync(
+                        new OrchestrationInfo
+                        {
+                            Name = runtimeState.Name,
+                            Tags = new Dictionary<string, string>(runtimeState.Tags ?? ImmutableDictionary<string, string>.Empty),
+                        },
+                        cancellationToken);
+                }
 
                 if (!filterPassed)
                 {

src/Worker/Grpc/GrpcDurableTaskWorker.cs

@@ -2,6 +2,7 @@
 // Licensed under the MIT License.
 
 using Microsoft.DurableTask.Worker.Hosting;
+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 
@@ -48,7 +49,8 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
         await using AsyncDisposable disposable = this.GetCallInvoker(out CallInvoker callInvoker, out string address);
         this.logger.StartingTaskHubWorker(address);
-        await new Processor(this, new(callInvoker)).ExecuteAsync(stoppingToken);
+        IOrchestrationFilter? filter = this.services.GetService<IOrchestrationFilter>();
+        await new Processor(this, new(callInvoker), filter).ExecuteAsync(stoppingToken);
     }
 
 #if NET6_0_OR_GREATER

test/Grpc.IntegrationTests/OrchestrationPatterns.cs

@@ -984,14 +984,11 @@ public async Task FilterOrchestrationsByName()
     {
         // Setup a worker with an Orchestration Filter.
         TaskName orchestratorName = nameof(EmptyOrchestration);
-        ISet<string> orchestratorNames = new HashSet<string>();
+        var orchestrationFilter = new OrchestrationFilter();
         await using HostTestLifetime server = await this.StartWorkerAsync(b =>
         {
             b.AddTasks(tasks => tasks.AddOrchestratorFunc(orchestratorName, ctx => Task.FromResult<object?>(null)));
-            b.UseOrchestrationFilter((info) =>
-            {
-                return !orchestratorNames.Contains(info.Name);
-            });
+            b.UseOrchestrationFilter(orchestrationFilter);
         });
 
         // Nothing in the filter set, the orchestration should complete.
@@ -1004,7 +1001,7 @@ public async Task FilterOrchestrationsByName()
         Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus);
 
         // Update the filter and re-enqueue. We should see the orchestration denied.
-        orchestratorNames.Add(orchestratorName);
+        orchestrationFilter.NameDenySet.Add(orchestratorName);
 
         // This should throw as the work is denied.
         instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName);
@@ -1020,14 +1017,11 @@ public async Task FilterOrchestrationsByTag()
         {
             { "test", "true" }
         };
-        IDictionary<string, string> orchestratorTagFilter = new Dictionary<string, string>();
+        var orchestrationFilter = new OrchestrationFilter();
         await using HostTestLifetime server = await this.StartWorkerAsync(b =>
         {
             b.AddTasks(tasks => tasks.AddOrchestratorFunc(orchestratorName, ctx => Task.FromResult<object?>(null)));
-            b.UseOrchestrationFilter((info) =>
-            {
-                return !orchestratorTagFilter.Any(kvp => info.Tags.ContainsKey(kvp.Key) && info.Tags[kvp.Key] == kvp.Value);
-            });
+            b.UseOrchestrationFilter(orchestrationFilter);
         });
 
         // Nothing in the filter set, the orchestration should complete.
@@ -1043,7 +1037,7 @@ public async Task FilterOrchestrationsByTag()
         Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus);
 
         // Update the filter and re-enqueue. We should see the orchestration denied.
-        orchestratorTagFilter.Add("test", "true");
+        orchestrationFilter.TagDenyDict.Add("test", "true");
 
         // This should throw as the work is denied.
         instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName, new StartOrchestrationOptions
@@ -1053,6 +1047,19 @@ public async Task FilterOrchestrationsByTag()
         await Assert.ThrowsAsync<OperationCanceledException>(async () => await server.Client.WaitForInstanceCompletionAsync(instanceId, new CancellationTokenSource(TimeSpan.FromSeconds(5)).Token));
     }
 
+    class OrchestrationFilter : IOrchestrationFilter
+    {
+        public ISet<string> NameDenySet { get; set; } = new HashSet<string>();
+        public IDictionary<string, string> TagDenyDict = new Dictionary<string, string>();
+
+        public Task<bool> IsOrchestrationValidAsync(OrchestrationInfo info, CancellationToken cancellationToken = default)
+        {
+            return Task.FromResult(
+                !this.NameDenySet.Contains(info.Name)
+                && !this.TagDenyDict.Any(kvp => info.Tags.ContainsKey(kvp.Key) && info.Tags[kvp.Key] == kvp.Value));
+        }
+    }
+
     // TODO: Test for multiple external events with the same name
     // TODO: Test for ContinueAsNew with external events that carry over
     // TODO: Test for catching activity exceptions of specific types