Skip to content

sample: sample for workflow ExecuteByQS(http) #99

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
81 changes: 79 additions & 2 deletions samples/DtmSample/Controllers/WfTestController.cs
Original file line number Diff line number Diff line change
Expand Up @@ -7,12 +7,10 @@
using Microsoft.Extensions.Options;
using System;
using System.IO;
using System.Diagnostics;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Text;
using System.Text.Json;
using System.Text.Unicode;
using System.Threading;
using System.Threading.Tasks;
using Exception = System.Exception;
Expand Down Expand Up @@ -257,5 +255,84 @@ public async Task<IActionResult> TccRollBack(CancellationToken cancellationToken
return Ok(TransResponse.BuildFailureResponse());
}
}


private static readonly string wfNameForResume = "wfNameForResume";

/// <summary>
///
/// </summary>
/// <param name="cancellationToken"></param>
/// <returns></returns>
[HttpPost("wf-crash")]
public async Task<IActionResult> Crash(CancellationToken cancellationToken)
{
if (!_globalTransaction.Exists(wfNameForResume))
{
_globalTransaction.Register(wfNameForResume, async (wf, data) =>
{
var content = new ByteArrayContent(data);
content.Headers.ContentType = new MediaTypeHeaderValue("application/json");

var outClient = wf.NewBranch().NewRequest();
await outClient.PostAsync(_settings.BusiUrl + "/TransOut", content);

// the first branch succeed, then crashed, the dtm server will call back the flowing wf-call-back
// manual stop application
Environment.Exit(0);

var inClient = wf.NewBranch().NewRequest();
await inClient.PostAsync(_settings.BusiUrl + "/TransIn", content);

return null;
});
}

var req = JsonSerializer.Serialize(new TransRequest("1", -30));
await _globalTransaction.Execute(wfNameForResume, Guid.NewGuid().ToString("N"), Encoding.UTF8.GetBytes(req), true);

return Ok(TransResponse.BuildSucceedResponse());
}

[HttpPost("wf-resume")]
public async Task<IActionResult> WfResume(CancellationToken cancellationToken)
{
try
{
if (!_globalTransaction.Exists(wfNameForResume))
{
// register again after manual crash by Environment.Exit(0);
_globalTransaction.Register(wfNameForResume, async (wf, data) =>
{
var content = new ByteArrayContent(data);
content.Headers.ContentType = new MediaTypeHeaderValue("application/json");

var outClient = wf.NewBranch().NewRequest();
await outClient.PostAsync(_settings.BusiUrl + "/TransOut", content);

var inClient = wf.NewBranch().NewRequest();
await inClient.PostAsync(_settings.BusiUrl + "/TransIn", content);

return null;
});
}

// prepared call ExecuteByQS
using var bodyMemoryStream = new MemoryStream();
await Request.Body.CopyToAsync(bodyMemoryStream, cancellationToken);
byte[] bytes = bodyMemoryStream.ToArray();
string body = Encoding.UTF8.GetString(bytes);
_logger.LogDebug($"body: {body}");

await _globalTransaction.ExecuteByQS(Request.Query, bodyMemoryStream.ToArray());

return Ok(TransResponse.BuildSucceedResponse());
}
catch (Exception ex)
{
_logger.LogError(ex, "Workflow Error");
return Ok(TransResponse.BuildFailureResponse());
}
}
}
}
Loading