Skip to content

Commit 5b306c0

Browse files
committed
sample: sample for workflow ExecuteByQS(http)
* sample: sample for workflow ExecuteByQS - Implement wf-crash endpoint to simulate workflow crash - Implement wf-resume endpoint for workflow http callback
1 parent 8c65784 commit 5b306c0

1 file changed

Lines changed: 79 additions & 2 deletions

File tree

samples/DtmSample/Controllers/WfTestController.cs

Lines changed: 79 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,12 +7,10 @@
77
using Microsoft.Extensions.Options;
88
using System;
99
using System.IO;
10-
using System.Diagnostics;
1110
using System.Net.Http;
1211
using System.Net.Http.Headers;
1312
using System.Text;
1413
using System.Text.Json;
15-
using System.Text.Unicode;
1614
using System.Threading;
1715
using System.Threading.Tasks;
1816
using Exception = System.Exception;
@@ -257,5 +255,84 @@ public async Task<IActionResult> TccRollBack(CancellationToken cancellationToken
257255
return Ok(TransResponse.BuildFailureResponse());
258256
}
259257
}
258+
259+
260+
private static readonly string wfNameForResume = "wfNameForResume";
261+
262+
/// <summary>
263+
///
264+
/// </summary>
265+
/// <param name="cancellationToken"></param>
266+
/// <returns></returns>
267+
[HttpPost("wf-crash")]
268+
public async Task<IActionResult> Crash(CancellationToken cancellationToken)
269+
{
270+
if (!_globalTransaction.Exists(wfNameForResume))
271+
{
272+
_globalTransaction.Register(wfNameForResume, async (wf, data) =>
273+
{
274+
var content = new ByteArrayContent(data);
275+
content.Headers.ContentType = new MediaTypeHeaderValue("application/json");
276+
277+
var outClient = wf.NewBranch().NewRequest();
278+
await outClient.PostAsync(_settings.BusiUrl + "/TransOut", content);
279+
280+
// the first branch succeed, then crashed, the dtm server will call back the flowing wf-call-back
281+
// manual stop application
282+
Environment.Exit(0);
283+
284+
var inClient = wf.NewBranch().NewRequest();
285+
await inClient.PostAsync(_settings.BusiUrl + "/TransIn", content);
286+
287+
return null;
288+
});
289+
}
290+
291+
var req = JsonSerializer.Serialize(new TransRequest("1", -30));
292+
await _globalTransaction.Execute(wfNameForResume, Guid.NewGuid().ToString("N"), Encoding.UTF8.GetBytes(req), true);
293+
294+
return Ok(TransResponse.BuildSucceedResponse());
295+
}
296+
297+
[HttpPost("wf-resume")]
298+
public async Task<IActionResult> WfResume(CancellationToken cancellationToken)
299+
{
300+
try
301+
{
302+
if (!_globalTransaction.Exists(wfNameForResume))
303+
{
304+
// register again after manual crash by Environment.Exit(0);
305+
_globalTransaction.Register(wfNameForResume, async (wf, data) =>
306+
{
307+
var content = new ByteArrayContent(data);
308+
content.Headers.ContentType = new MediaTypeHeaderValue("application/json");
309+
310+
var outClient = wf.NewBranch().NewRequest();
311+
await outClient.PostAsync(_settings.BusiUrl + "/TransOut", content);
312+
313+
var inClient = wf.NewBranch().NewRequest();
314+
await inClient.PostAsync(_settings.BusiUrl + "/TransIn", content);
315+
316+
return null;
317+
});
318+
}
319+
320+
// prepared call ExecuteByQS
321+
using var bodyMemoryStream = new MemoryStream();
322+
await Request.Body.CopyToAsync(bodyMemoryStream, cancellationToken);
323+
byte[] bytes = bodyMemoryStream.ToArray();
324+
string body = Encoding.UTF8.GetString(bytes);
325+
_logger.LogDebug($"body: {body}");
326+
327+
await _globalTransaction.ExecuteByQS(Request.Query, bodyMemoryStream.ToArray());
328+
329+
return Ok(TransResponse.BuildSucceedResponse());
330+
}
331+
catch (Exception ex)
332+
{
333+
_logger.LogError(ex, "Workflow Error");
334+
return Ok(TransResponse.BuildFailureResponse());
335+
}
336+
}
260337
}
261338
}

0 commit comments

Comments
 (0)