-
Notifications
You must be signed in to change notification settings - Fork 21
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* feat: add saga sample * docs: add init readme * simplify saga workflow * fix: use shallow errors * remove throw exception * fix: make exection non retryable * disable CA1031 warning * remove extra check --------- Co-authored-by: Loren☺️ <[email protected]>
- Loading branch information
Showing
7 changed files
with
220 additions
and
23 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| using Microsoft.Extensions.Logging; | ||
| using Temporalio.Activities; | ||
| using Temporalio.Exceptions; | ||
|
|
||
| namespace TemporalioSamples.Saga; | ||
|
|
||
| public record TransferDetails(decimal Amount, string FromAmount, string ToAmount, string ReferenceId); | ||
|
|
||
| public static class Activities | ||
| { | ||
| [Activity] | ||
| public static void Withdraw(TransferDetails d) | ||
| { | ||
| ActivityExecutionContext.Current.Logger.LogInformation("Withdrawing {Amount} from account {FromAmount}. ReferenceId: {ReferenceId}", d.Amount, d.FromAmount, d.ReferenceId); | ||
| } | ||
|
|
||
| [Activity] | ||
| public static void WithdrawCompensation(TransferDetails d) | ||
| { | ||
| ActivityExecutionContext.Current.Logger.LogInformation("Withdrawing Compensation {Amount} from account {FromAmount}. ReferenceId: {ReferenceId}", d.Amount, d.FromAmount, d.ReferenceId); | ||
| } | ||
|
|
||
| [Activity] | ||
| public static void Deposit(TransferDetails d) | ||
| { | ||
| ActivityExecutionContext.Current.Logger.LogInformation("Depositing {Amount} into account {ToAmount}. ReferenceId: {ReferenceId}", d.Amount, d.ToAmount, d.ReferenceId); | ||
| } | ||
|
|
||
| [Activity] | ||
| public static void DepositCompensation(TransferDetails d) | ||
| { | ||
| ActivityExecutionContext.Current.Logger.LogInformation("Depositing Compensation {Amount} int account {ToAmount}. ReferenceId: {ReferenceId}", d.Amount, d.ToAmount, d.ReferenceId); | ||
| } | ||
|
|
||
| [Activity] | ||
| public static void StepWithError(TransferDetails d) | ||
| { | ||
| ActivityExecutionContext.Current.Logger.LogInformation("Simulate failure to trigger compensation. ReferenceId: {ReferenceId}", d.ReferenceId); | ||
| throw new ApplicationFailureException("Simulated failure", nonRetryable: true); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,67 @@ | ||
| using System.Diagnostics; | ||
| using Microsoft.Extensions.Logging; | ||
| using Temporalio.Client; | ||
| using Temporalio.Worker; | ||
| using TemporalioSamples.Saga; | ||
|
|
||
| var client = await TemporalClient.ConnectAsync(new("localhost:7233") | ||
| { | ||
| LoggerFactory = LoggerFactory.Create(builder => | ||
| builder.AddSimpleConsole(options => options.TimestampFormat = "[HH:mm:ss] ").SetMinimumLevel(LogLevel.Information)), | ||
| }); | ||
|
|
||
| async Task RunWorkerAsync() | ||
| { | ||
| // Cancellation token cancelled on ctrl+c | ||
| using var tokenSource = new CancellationTokenSource(); | ||
| Console.CancelKeyPress += (_, eventArgs) => | ||
| { | ||
| tokenSource.Cancel(); | ||
| eventArgs.Cancel = true; | ||
| }; | ||
|
|
||
| // Run worker until cancelled | ||
| Console.WriteLine("Running worker"); | ||
|
|
||
| using var worker = new TemporalWorker( | ||
| client, | ||
| new TemporalWorkerOptions(taskQueue: "workflow-saga-sample") | ||
| .AddAllActivities(typeof(Activities), null) | ||
| .AddWorkflow<SagaWorkflow>()); | ||
| try | ||
| { | ||
| await worker.ExecuteAsync(tokenSource.Token); | ||
| } | ||
| catch (OperationCanceledException) | ||
| { | ||
| Console.WriteLine("Worker cancelled"); | ||
| } | ||
| } | ||
|
|
||
| async Task ExecuteWorkflowAsync() | ||
| { | ||
| var workflowId = "test-" + Guid.NewGuid(); | ||
| Console.WriteLine($"Starting test workflow with id '{workflowId}'."); | ||
|
|
||
| var sw = Stopwatch.StartNew(); | ||
| var handle = await client.StartWorkflowAsync( | ||
| (SagaWorkflow wf) => wf.RunAsync(new TransferDetails(100, "acc1000", "acc2000", "1324")), | ||
| new(workflowId, "workflow-saga-sample")); | ||
|
|
||
| Console.WriteLine($"Test workflow '{workflowId}' started"); | ||
|
|
||
| await handle.GetResultAsync(); | ||
| Console.WriteLine($"Test workflow '{workflowId}' finished after {sw.ElapsedMilliseconds}ms"); | ||
| } | ||
|
|
||
| switch (args.ElementAtOrDefault(0)) | ||
| { | ||
| case "worker": | ||
| await RunWorkerAsync(); | ||
| break; | ||
| case "workflow": | ||
| await ExecuteWorkflowAsync(); | ||
| break; | ||
| default: | ||
| throw new ArgumentException("Must pass 'worker' or 'workflow' as the first argument"); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| # Saga | ||
|
|
||
| This sample demonstrates orchestrating microservices using a very simplistic Saga pattern. | ||
|
|
||
| To run, first see [README.md](../../README.md) for prerequisites. Then, run the following from this directory | ||
| in a separate terminal to start the worker: | ||
|
|
||
| dotnet run worker | ||
|
|
||
| Then in another terminal, run the workflow from this directory: | ||
|
|
||
| dotnet run workflow | ||
|
|
||
| This will show logs in the worker window of the workflow running. |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| using Microsoft.Extensions.Logging; | ||
| using Temporalio.Workflows; | ||
|
|
||
| namespace TemporalioSamples.Saga; | ||
|
|
||
| [Workflow] | ||
| public class SagaWorkflow | ||
| { | ||
| [WorkflowRun] | ||
| public async Task RunAsync(TransferDetails transfer) | ||
| { | ||
| List<Func<Task>> compensations = new(); | ||
| var logger = Workflow.Logger; | ||
|
|
||
| var options = new ActivityOptions() { StartToCloseTimeout = TimeSpan.FromSeconds(90) }; | ||
|
|
||
| try | ||
| { | ||
| await Workflow.ExecuteActivityAsync(() => Activities.Withdraw(transfer), options); | ||
|
|
||
| compensations.Add(async () => await Workflow.ExecuteActivityAsync( | ||
| () => Activities.WithdrawCompensation(transfer), | ||
| options)); | ||
|
|
||
| await Workflow.ExecuteActivityAsync(() => Activities.Deposit(transfer), options); | ||
|
|
||
| compensations.Add(async () => await Workflow.ExecuteActivityAsync( | ||
| () => Activities.DepositCompensation(transfer), | ||
| options)); | ||
|
|
||
| // throw new Exception | ||
| await Workflow.ExecuteActivityAsync(() => Activities.StepWithError(transfer), options); | ||
| } | ||
| catch (Exception) | ||
| { | ||
| logger.LogInformation("Exception caught. Initiating compensation..."); | ||
| await CompensateAsync(compensations); | ||
| throw; | ||
| } | ||
| } | ||
|
|
||
| private async Task CompensateAsync(List<Func<Task>> compensations) | ||
| { | ||
| compensations.Reverse(); | ||
| foreach (var comp in compensations) | ||
| { | ||
| #pragma warning disable CA1031 | ||
| try | ||
| { | ||
| await comp.Invoke(); | ||
| } | ||
| catch (Exception ex) | ||
| { | ||
| Workflow.Logger.LogError(ex, "Failed to compensate"); | ||
| // swallow errors | ||
| } | ||
| #pragma warning restore CA1031 | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,7 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
|
|
||
| <PropertyGroup> | ||
| <OutputType>Exe</OutputType> | ||
| </PropertyGroup> | ||
|
|
||
| </Project> |