feat(workflow): add support for HTTP branches interceptor in workflows

Author: wooln

src/Dtmcli/Constant.cs

@@ -4,7 +4,8 @@ internal static class Constant
     {
         internal const string DtmClientHttpName = "dtmClient";
         internal const string BranchClientHttpName = "branchClient";
-       
+        internal const string WorkflowBranchClientHttpName = "WF";
+
         internal static class Request
         {
             internal const string CONTENT_TYPE = "application/json";

src/Dtmcli/DtmClient.cs

@@ -148,7 +148,7 @@ public async Task<TransGlobal> Query(string gid, CancellationToken cancellationT
             var client = _httpClientFactory.CreateClient(Constant.DtmClientHttpName);
             var response = await client.GetAsync(url, cancellationToken).ConfigureAwait(false);
             var dtmContent = await response.Content.ReadAsStringAsync().ConfigureAwait(false);
-            DtmImp.Utils.CheckStatus(response.StatusCode, dtmContent);
+            DtmImp.Utils.CheckStatusCode(response.StatusCode);
             return JsonSerializer.Deserialize<TransGlobal>(dtmContent, _jsonOptions);
         }
 
@@ -167,7 +167,7 @@ public async Task<string> QueryStatus(string gid, CancellationToken cancellation
             var client = _httpClientFactory.CreateClient(Constant.DtmClientHttpName);
             var response = await client.GetAsync(url, cancellationToken).ConfigureAwait(false);
             var dtmContent = await response.Content.ReadAsStringAsync().ConfigureAwait(false);
-            DtmImp.Utils.CheckStatus(response.StatusCode, dtmContent);
+            DtmImp.Utils.CheckStatusCode(response.StatusCode);
             var graph = JsonSerializer.Deserialize<TransGlobalForStatus>(dtmContent, _jsonOptions);
             return graph.Transaction == null
                 ? string.Empty

src/Dtmcli/DtmImp/Utils.cs

@@ -40,6 +40,14 @@ public static void CheckStatus(HttpStatusCode status, string dtmResult)
             }
         }
 
+        public static void CheckStatusCode(HttpStatusCode status)
+        {
+            if (status != HttpStatusCode.OK)
+            {
+                throw new DtmException(string.Format(CheckStatusMsgFormat, status.ToString(), string.Empty));
+            }
+        }
+
         /// <summary>
         /// OrString return the first not null or not empty string
         /// </summary>

src/Dtmcli/TransGlobal.cs

@@ -28,7 +28,7 @@ public class DtmTransaction
     {
         [JsonPropertyName("id")] public int Id { get; set; }
 
-        [JsonPropertyName("create_time")] public DateTimeOffset CreateTime { get; set; }
+        [JsonPropertyName("create_time")] public DateTimeOffset? CreateTime { get; set; }
 
         [JsonPropertyName("update_time")] public DateTimeOffset UpdateTime { get; set; }
 
@@ -64,9 +64,9 @@ public class DtmBranch
     {
         [JsonPropertyName("id")] public int Id { get; set; }
 
-        [JsonPropertyName("create_time")] public DateTimeOffset CreateTime { get; set; }
+        [JsonPropertyName("create_time")] public DateTimeOffset? CreateTime { get; set; }
 
-        [JsonPropertyName("update_time")] public DateTimeOffset UpdateTime { get; set; }
+        [JsonPropertyName("update_time")] public DateTimeOffset? UpdateTime { get; set; }
 
         [JsonPropertyName("gid")] public string Gid { get; set; }
 

src/Dtmworkflow/ServiceCollectionExtensions.cs

@@ -23,6 +23,8 @@ public static IServiceCollection AddDtmWorkflow(this IServiceCollection services
             services.TryAddSingleton<IWorkflowFactory, WorkflowFactory>();
             services.TryAddSingleton<WorkflowGlobalTransaction>();
 
+            AddHttpClient(services);
+
             return services;
         }
 
@@ -33,8 +35,22 @@ public static IServiceCollection AddDtmWorkflow(this IServiceCollection services
 
             services.TryAddSingleton<IWorkflowFactory, WorkflowFactory>();
             services.TryAddSingleton<WorkflowGlobalTransaction>();
+                       
+            AddHttpClient(services);
 
             return services;
         }
+
+        private static void AddHttpClient(IServiceCollection services /*, DtmOptions options*/)
+        {
+            services.AddHttpClient(Dtmcli.Constant.WorkflowBranchClientHttpName, client =>
+            {
+                // TODO DtmOptions
+                // client.Timeout = TimeSpan.FromMilliseconds(options.BranchTimeout);                
+            }).AddHttpMessageHandler<WorkflowHttpInterceptor>();
+
+            // TODO how to inject workflow instance?
+            services.AddTransient<WorkflowHttpInterceptor>();
+        }
     }
-}
+}

src/Dtmworkflow/Workflow.Imp.cs

@@ -64,7 +64,7 @@ internal async Task<byte[]> Process(WfFunc2 handler, byte[] data)
             }
 
             err = Utils.GrpcError2DtmError(err);
-
+            
             if (err != null && err is not DtmCommon.DtmFailureException) throw err;
 
             try
@@ -193,7 +193,7 @@ private StepResult StepResultFromGrpc(IMessage reply, Exception err)
             return sr;
         }
 
-        private HttpResponseMessage StepResultToHttp(StepResult r)
+        public HttpResponseMessage StepResultToHttp(StepResult r)
         {
             if (r.Error != null)
             {
@@ -203,7 +203,7 @@ private HttpResponseMessage StepResultToHttp(StepResult r)
             return Utils.NewJSONResponse(HttpStatusCode.OK, r.Data);
         }
 
-        private StepResult StepResultFromHTTP(HttpResponseMessage resp, Exception err)
+        public StepResult StepResultFromHTTP(HttpResponseMessage resp, Exception err)
         {
             var sr = new StepResult
             {
@@ -212,7 +212,7 @@ private StepResult StepResultFromHTTP(HttpResponseMessage resp, Exception err)
 
             if (err == null)
             {
-                // HTTPResp2DtmError
+                (sr.Data, sr.Error) = Utils.HTTPResp2DtmError(resp); // TODO go 使用了 this.Options.HTTPResp2DtmError(resp), 方便定制
                 sr.Status = WfErrorToStatus(sr.Error);
             }
 
@@ -234,9 +234,9 @@ private string WfErrorToStatus(Exception err)
         }
 
 
-        private async Task<StepResult> RecordedDo(Func<DtmCommon.BranchBarrier, Task<StepResult>> fn)
+        public async Task<StepResult> RecordedDo(Func<DtmCommon.BranchBarrier, Task<StepResult>> fn)
         {
-            var sr = await this.RecordedDoInner(fn);
+            StepResult sr = await this.RecordedDoInner(fn);
 
             // do not compensate the failed branch if !CompensateErrorBranch
             if (this.Options.CompensateErrorBranch && sr.Status == DtmCommon.Constant.StatusFailed)

src/Dtmworkflow/Workflow.cs

@@ -3,7 +3,9 @@
 using Microsoft.Extensions.Logging;
 using System;
 using System.Collections.Generic;
+using System.Net.Http;
 using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
 
 namespace Dtmworkflow
 {
@@ -32,7 +34,13 @@ public Workflow(IDtmClient httpClient, IDtmgRPCClient grpcClient, Dtmcli.IBranch
 
         public System.Net.Http.HttpClient NewRequest()
         {
-            return _httpClient.GetHttpClient("WF");
+            if(true)
+                return new HttpClient(new WorkflowHttpInterceptor(this));
+            else
+            {
+                var client =  _httpClient.GetHttpClient("WF");
+                return client;
+            }
         }
 
         /// <summary>

src/Dtmworkflow/WorkflowHttpInterceptor.cs

@@ -0,0 +1,39 @@
+using System;
+using System.Net.Http;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Dtmworkflow;
+
+internal class WorkflowHttpInterceptor : DelegatingHandler
+{
+    private readonly Workflow _wf;
+
+    public WorkflowHttpInterceptor(Workflow wf)
+    {
+        this._wf = wf;
+        InnerHandler = new HttpClientHandler();
+    }
+
+    protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
+    {
+        Func<DtmCommon.BranchBarrier, Task<StepResult>> origin = async (barrier) =>
+        {
+            var response = await base.SendAsync(request, cancellationToken);
+            return _wf.StepResultFromHTTP(response, null);
+        };
+
+        StepResult sr;
+        // in phase 2, do not save, because it is saved outer
+        if (_wf.WorkflowImp.CurrentOp != DtmCommon.Constant.OpAction)
+        {
+            sr = await origin(null);
+        }
+        else
+        {
+            sr = await _wf.RecordedDo(origin);
+        }
+
+        return _wf.StepResultToHttp(sr);
+    }
+}

tests/BusiGrpcService/Program.cs

@@ -8,6 +8,8 @@
 {
     // Setup a HTTP/2 endpoint without TLS.
     options.ListenLocalhost(5005, o => o.Protocols = HttpProtocols.Http2);
+    // test for workflow http branch
+    options.ListenLocalhost(5006, o => o.Protocols = HttpProtocols.Http1);
 });
 
 builder.Services.AddGrpc();
@@ -20,6 +22,15 @@
 
 // Configure the HTTP request pipeline.
 app.MapGrpcService<BusiApiService>();
+
+// test for workflow http branch
+app.MapGet("/test-http-ok1", () => "SUCCESS");
+app.MapGet("/test-http-ok2", () => "SUCCESS");
+app.MapGet("/409", context =>
+{    
+    context.Response.StatusCode = 409;
+    return context.Response.WriteAsync("i am body, the http branch is 409"); // FAILURE
+});
 app.MapGet("/", () => "Communication with gRPC endpoints must be made through a gRPC client. To learn how to create a client, visit: https://go.microsoft.com/fwlink/?linkid=2086909");
 
 app.Run();