Remove DurableClientContext, allow for direct binding of DurableTaskClient (#2372)

Author: jviau

src/Worker.Extensions.DurableTask/DefaultDurableClientContext.cs

@@ -0,0 +1,66 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System;
+using System.Text.Json;
+using System.Threading.Tasks;
+using Microsoft.Azure.Functions.Worker.Converters;
+using Microsoft.DurableTask.Client;
+
+namespace Microsoft.Azure.Functions.Worker.Extensions.DurableTask;
+
+internal sealed partial class DurableTaskClientConverter : IInputConverter
+{
+    private readonly FunctionsDurableClientProvider clientProvider;
+
+    // Constructor parameters are optional DI-injected services.
+    public DurableTaskClientConverter(FunctionsDurableClientProvider clientProvider)
+    {
+        this.clientProvider = clientProvider ?? throw new ArgumentNullException(nameof(clientProvider));
+    }
+
+    public ValueTask<ConversionResult> ConvertAsync(ConverterContext context)
+    {
+        if (context is null)
+        {
+            throw new ArgumentNullException(nameof(context));
+        }
+
+        if (context.TargetType != typeof(DurableTaskClient))
+        {
+            return new ValueTask<ConversionResult>(ConversionResult.Unhandled());
+        }
+
+        // The exact format of the expected JSON string data is controlled by the Durable Task WebJobs client binding logic.
+        // It's never expected to be wrong, but we code defensively just in case.
+        if (context.Source is not string clientConfigText)
+        {
+            return new ValueTask<ConversionResult>(ConversionResult.Failed(new InvalidOperationException(
+                $"Expected the Durable Task WebJobs SDK extension to send a string payload for {nameof(DurableClientAttribute)}.")));
+        }
+
+        try
+        {
+            DurableClientInputData? inputData = JsonSerializer.Deserialize<DurableClientInputData>(clientConfigText);
+            if (!Uri.TryCreate(inputData?.rpcBaseUrl, UriKind.Absolute, out Uri? endpoint))
+            {
+                return new ValueTask<ConversionResult>(ConversionResult.Failed(
+                    new InvalidOperationException("Failed to parse the input binding payload data")));
+            }
+
+            DurableTaskClient client = this.clientProvider.GetClient(endpoint, inputData?.taskHubName, inputData?.connectionName);
+            client = new FunctionsDurableTaskClient(client, inputData!.requiredQueryStringParameters);
+            return new ValueTask<ConversionResult>(ConversionResult.Success(client));
+        }
+        catch (Exception innerException)
+        {
+            InvalidOperationException exception = new(
+                $"Failed to convert the input binding context data into a {nameof(DurableTaskClient)} object. The data may have been delivered in an invalid format.",
+                innerException);
+            return new ValueTask<ConversionResult>(ConversionResult.Failed(exception));
+        }
+    }
+
+    // Serializer is case-sensitive and incoming JSON properties are camel-cased.
+    private record DurableClientInputData(string rpcBaseUrl, string taskHubName, string connectionName, string requiredQueryStringParameters);
+}

src/Worker.Extensions.DurableTask/DurableClientContext.cs

@@ -0,0 +1,111 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System;
+using System.Net;
+using System.Threading;
+using Azure.Core.Serialization;
+using Microsoft.Azure.Functions.Worker.Http;
+using Microsoft.DurableTask.Client;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Options;
+
+namespace Microsoft.Azure.Functions.Worker;
+
+/// <summary>
+/// Extensions for <see cref="DurableTaskClient"/>
+/// </summary>
+public static class DurableTaskClientExtensions
+{
+    /// <summary>
+    /// Creates an HTTP response that is useful for checking the status of the specified instance.
+    /// </summary>
+    /// <param name="client">The <see cref="DurableTaskClient"/>.</param>
+    /// <param name="request">The HTTP request that this response is for.</param>
+    /// <param name="instanceId">The ID of the orchestration instance to check.</param>
+    /// <param name="cancellation">The cancellation token.</param>
+    /// <returns>An HTTP 202 response with a Location header and a payload containing instance control URLs.</returns>
+    public static HttpResponseData CreateCheckStatusResponse(
+        this DurableTaskClient client,
+        HttpRequestData request,
+        string instanceId,
+        CancellationToken cancellation = default)
+    {
+        return client.CreateCheckStatusResponse(request, instanceId, HttpStatusCode.Accepted, cancellation);
+    }
+
+    /// <summary>
+    /// Creates an HTTP response that is useful for checking the status of the specified instance.
+    /// </summary>
+    /// <param name="client">The <see cref="DurableTaskClient"/>.</param>
+    /// <param name="request">The HTTP request that this response is for.</param>
+    /// <param name="instanceId">The ID of the orchestration instance to check.</param>
+    /// <param name="statusCode">The status code.</param>
+    /// <param name="cancellation">The cancellation token.</param>
+    /// <returns>An HTTP response with a Location header and a payload containing instance control URLs.</returns>
+    public static HttpResponseData CreateCheckStatusResponse(
+        this DurableTaskClient client,
+        HttpRequestData request,
+        string instanceId,
+        HttpStatusCode statusCode,
+        CancellationToken cancellation = default)
+    {
+        if (client is null)
+        {
+            throw new ArgumentNullException(nameof(client));
+        }
+
+        static string BuildUrl(string url, params string?[] queryValues)
+        {
+            bool appended = false;
+            foreach (string? query in queryValues)
+            {
+                if (!string.IsNullOrEmpty(query))
+                {
+                    url = url + (appended ? "&" : "?") + query;
+                    appended = true;
+                }
+            }
+
+            return url;
+        }
+
+        // TODO: To better support scenarios involving proxies or application gateways, this
+        //       code should take the X-Forwarded-Host, X-Forwarded-Proto, and Forwarded HTTP
+        //       request headers into consideration and generate the base URL accordingly.
+        //       More info: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Forwarded.
+        //       One potential workaround is to set ASPNETCORE_FORWARDEDHEADERS_ENABLED to true.
+        string baseUrl = request.Url.GetLeftPart(UriPartial.Authority);
+        string formattedInstanceId = Uri.EscapeDataString(instanceId);
+        string instanceUrl = $"{baseUrl}/runtime/webhooks/durabletask/instances/{formattedInstanceId}";
+        string? commonQueryParameters = GetQueryParams(client);
+
+        HttpResponseData response = request.CreateResponse(statusCode);
+        response.Headers.Add("Location", BuildUrl(instanceUrl, commonQueryParameters));
+        response.Headers.Add("Content-Type", "application/json");
+
+        ObjectSerializer serializer = GetObjectSerializer(response);
+        var payload = new
+        {
+            id = instanceId,
+            purgeHistoryDeleteUri = BuildUrl(instanceUrl, commonQueryParameters),
+            sendEventPostUri = BuildUrl($"{instanceUrl}/raiseEvent/{{eventName}}", commonQueryParameters),
+            statusQueryGetUri = BuildUrl(instanceUrl, commonQueryParameters),
+            terminatePostUri = BuildUrl($"{instanceUrl}/terminate", "reason={{text}}}", commonQueryParameters),
+        };
+
+        serializer.Serialize(response.Body, payload, payload.GetType(), cancellation);
+        return response;
+    }
+
+    private static ObjectSerializer GetObjectSerializer(HttpResponseData response)
+    {
+        return response.FunctionContext.InstanceServices.GetService<IOptions<WorkerOptions>>()?.Value?.Serializer
+            ?? throw new InvalidOperationException("A serializer is not configured for the worker.");
+    }
+
+    private static string? GetQueryParams(DurableTaskClient client)
+    {
+        return client is FunctionsDurableTaskClient functions ? functions.QueryString : null;
+    }
+}

src/Worker.Extensions.DurableTask/DurableTaskClientConverter.cs

@@ -54,7 +54,12 @@ public override void Configure(IFunctionsWorkerApplicationBuilder applicationBui
             return new DurableTaskShimFactory(options, factory); // For GrpcOrchestrationRunner
         });
 
-        applicationBuilder.Services.Configure<WorkerOptions>(o => o.InputConverters.Register<OrchestrationInputConverter>());
+        applicationBuilder.Services.Configure<WorkerOptions>(o =>
+        {
+            o.InputConverters.RegisterAt<DurableTaskClientConverter>(0);
+            o.InputConverters.Register<OrchestrationInputConverter>();
+        });
+
         applicationBuilder.UseMiddleware<DurableTaskFunctionsMiddleware>();
     }
 

src/Worker.Extensions.DurableTask/DurableTaskClientExtensions.cs

@@ -23,6 +23,9 @@ namespace Microsoft.Azure.Functions.Worker.Extensions.DurableTask;
 /// <summary>
 /// The functions implementation of the durable task client provider.
 /// </summary>
+/// <remarks>
+/// This class does NOT provide <see cref="FunctionsDurableTaskClient" /> is meant as a per-binding wrapper.
+/// </remarks>
 internal partial class FunctionsDurableClientProvider : IAsyncDisposable
 {
     private readonly ReaderWriterLockSlim sync = new();
@@ -127,7 +130,7 @@ public DurableTaskClient GetClient(Uri endpoint, string? taskHub, string? connec
             };
 
             ILogger logger = this.loggerFactory.CreateLogger<GrpcDurableTaskClient>();
-            GrpcDurableTaskClient client = new(string.Empty, options, logger);
+            GrpcDurableTaskClient client = new(taskHub, options, logger);
             holder = new(client, channel);
             this.clients[key] = holder;
             return client;