Added RetryFlow sample and retry-option overloads to Flow base-class.

Author: stidsborg

Cleipnir.Flows/Flow.cs

@@ -14,16 +14,48 @@ public abstract class BaseFlow
     public Utilities Utilities => Workflow.Utilities;
     public Messages Messages => Workflow.Messages;
     public Effect Effect => Workflow.Effect;
-    
+
+    #region Capture explicit id with ResiliencyLevel
+
     public Task<T> Capture<T>(string id, Func<Task<T>> work, ResiliencyLevel resiliencyLevel = ResiliencyLevel.AtLeastOnce) 
         => Effect.Capture(id, work, resiliencyLevel);
     public Task<T> Capture<T>(string id, Func<T> work, ResiliencyLevel resiliencyLevel = ResiliencyLevel.AtLeastOnce) 
         => Effect.Capture(id, work, resiliencyLevel);
     public Task Capture(string id, Func<Task> work, ResiliencyLevel resiliencyLevel = ResiliencyLevel.AtLeastOnce) 
         => Effect.Capture(id, work, resiliencyLevel);
     public Task Capture(string id, Action work, ResiliencyLevel resiliencyLevel = ResiliencyLevel.AtLeastOnce) 
-        => Effect.Capture(id, work, resiliencyLevel);
+        => Effect.Capture(id, work, resiliencyLevel); 
+
+    #endregion
+
+    #region Capture explicit id with RetryPolicy
+
+    public Task<T> Capture<T>(string id, Func<Task<T>> work, RetryPolicy retryPolicy, bool flush = true) 
+        => Effect.Capture(id, work, retryPolicy, flush);
+    public Task<T> Capture<T>(string id, Func<T> work, RetryPolicy retryPolicy, bool flush = true) 
+        => Effect.Capture(id, work, retryPolicy, flush);
+    public Task Capture(string id, Func<Task> work, RetryPolicy retryPolicy, bool flush = true) 
+        => Effect.Capture(id, work, retryPolicy, flush);
+    public Task Capture(string id, Action work, RetryPolicy retryPolicy, bool flush = true) 
+        => Effect.Capture(id, work, retryPolicy, flush);
+
+    #endregion
+
+    #region Capture implicit id with RetryPolicy
+
+    public Task<T> Capture<T>(Func<Task<T>> work, RetryPolicy retryPolicy, bool flush = true) 
+        => Effect.Capture(work, retryPolicy, flush);
+    public Task<T> Capture<T>(Func<T> work, RetryPolicy retryPolicy, bool flush = true) 
+        => Effect.Capture(work, retryPolicy, flush);
+    public Task Capture(Func<Task> work, RetryPolicy retryPolicy, bool flush = true) 
+        => Effect.Capture(work, retryPolicy, flush);
+    public Task Capture(Action work, RetryPolicy retryPolicy, bool flush = true) 
+        => Effect.Capture(work, retryPolicy, flush);
+
+    #endregion
 
+    #region Capture implicit id with ResiliencyLevel
+
     public Task<T> Capture<T>(Func<Task<T>> work, ResiliencyLevel resiliencyLevel = ResiliencyLevel.AtLeastOnce) 
         => Effect.Capture(work, resiliencyLevel);
     public Task<T> Capture<T>(Func<T> work, ResiliencyLevel resiliencyLevel = ResiliencyLevel.AtLeastOnce) 
@@ -32,6 +64,8 @@ public Task Capture(Func<Task> work, ResiliencyLevel resiliencyLevel = Resilienc
         => Effect.Capture(work, resiliencyLevel);
     public Task Capture(Action work, ResiliencyLevel resiliencyLevel = ResiliencyLevel.AtLeastOnce) 
         => Effect.Capture(work, resiliencyLevel);
+
+    #endregion
 
     public Task<TMessage> Message<TMessage>() => Workflow.Messages.FirstOfType<TMessage>();
     public Task<Option<TMessage>> Message<TMessage>(string timeoutId, DateTime timesOutAt) => Workflow

Cleipnir.Flows/Options.cs

@@ -31,17 +31,17 @@ public class Options
     /// Configuration options for Cleipnir
     /// </summary>
     /// <param name="unhandledExceptionHandler">Callback handler for unhandled flow exceptions</param>
-    /// <param name="retentionPeriod">Period to keep completed flows before deletion</param>
-    /// <param name="retentionCleanUpFrequency">Retention clean-up check frequency</param>
-    /// <param name="leaseLength">Flow lease-length. Leases are automatically renewed</param>
-    /// <param name="enableWatchdogs">Enable background crashed, interrupted and postponed flow scheduling</param>
-    /// <param name="watchdogCheckFrequency">Check frequency for eligible crashed, interrupted and postponed flows</param>
-    /// <param name="messagesPullFrequency">Pull frequency for active/max-waiting messages</param>
-    /// <param name="messagesDefaultMaxWaitForCompletion">Default wait duration before suspension for messages</param>
-    /// <param name="delayStartup">Delay watchdog start-up</param>
-    /// <param name="maxParallelRetryInvocations">Limit the number of watchdog started invocations</param>
-    /// <param name="serializer">Specify custom serializer</param>
-    /// <param name="utcNow">Provide custom delegate for providing current utc datetime</param>
+    /// <param name="retentionPeriod">Period to keep completed flows before deletion. Default infinite.</param>
+    /// <param name="retentionCleanUpFrequency">Retention clean-up check frequency. Default 1 hour when retention period is not infinite.</param>
+    /// <param name="leaseLength">Flow lease-length. Leases are automatically renewed. Default 60 seconds.</param>
+    /// <param name="enableWatchdogs">Enable background crashed, interrupted and postponed flow scheduling. Default true.</param>
+    /// <param name="watchdogCheckFrequency">Check frequency for eligible crashed, interrupted and postponed flows. Default 1 second.</param>
+    /// <param name="messagesPullFrequency">Pull frequency for active/max-waiting messages. Default: 250ms</param>
+    /// <param name="messagesDefaultMaxWaitForCompletion">Default wait duration before suspension for messages. Defaults to none.</param>
+    /// <param name="delayStartup">Delay watchdog start-up. Defaults to none.</param>
+    /// <param name="maxParallelRetryInvocations">Limit the number of watchdog started invocations. Default: 100.</param>
+    /// <param name="serializer">Specify custom serializer. Default built-in json-serializer.</param>
+    /// <param name="utcNow">Provide custom delegate for providing current utc datetime. Default: () => DateTime.UtcNow</param>
     public Options(
         Action<FrameworkException>? unhandledExceptionHandler = null, 
         TimeSpan? retentionPeriod = null,

Samples/Cleipnir.Flows.Samples.Console/Cleipnir.Flows.Sample.ConsoleApp.csproj

@@ -8,6 +8,7 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <ProjectReference Include="..\..\Cleipnir.ResilientFunctions\Stores\PostgreSQL\Cleipnir.ResilientFunctions.PostgreSQL\Cleipnir.ResilientFunctions.PostgreSQL.csproj" />
     <ProjectReference Include="..\..\SourceGeneration\Cleipnir.Flows.SourceGenerator\Cleipnir.Flows.SourceGenerator.csproj" OutputItemType="Analyzer" ReferenceOutputAssembly="false" />
     <ProjectReference Include="..\..\Cleipnir.Flows\Cleipnir.Flows.csproj" />
   </ItemGroup>

Samples/Cleipnir.Flows.Samples.Console/Program.cs

@@ -4,6 +4,6 @@ public static class Program
 {
     private static async Task Main(string[] args)
     {
-        await Middleware.Example.Do();
+        await Retry.Example.Do();
     }
 }

Samples/Cleipnir.Flows.Samples.Console/Retry/Example.cs

@@ -0,0 +1,31 @@
+using Cleipnir.ResilientFunctions.PostgreSQL;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Cleipnir.Flows.Sample.ConsoleApp.Retry;
+
+public static class Example
+{
+    public static async Task Do()
+    {
+        const string connStr = "Server=localhost;Database=retryflows;User Id=postgres;Password=Pa55word!; Include Error Detail=true;";
+        await DatabaseHelper.RecreateDatabase(connStr);
+        var store = new PostgreSqlFunctionStore(connStr);
+        await store.Initialize();
+        
+        var serviceCollection = new ServiceCollection();
+        serviceCollection.AddTransient<RetryFlows>();
+        serviceCollection.AddTransient<RetryFlow>();
+        
+        var flowsContainer = new FlowsContainer(
+            store,
+            serviceCollection.BuildServiceProvider(),
+            new Options(unhandledExceptionHandler: Console.WriteLine)
+        );
+
+        var flows = new RetryFlows(flowsContainer);
+        var flowId = "MK-54321";
+        await flows.Schedule(flowId);
+
+        Console.ReadLine();
+    }
+}

Samples/Cleipnir.Flows.Samples.Console/Retry/RetryFlow.cs

@@ -0,0 +1,24 @@
+using Cleipnir.ResilientFunctions.Domain;
+
+namespace Cleipnir.Flows.Sample.ConsoleApp.Retry;
+
+[GenerateFlows]
+public class RetryFlow : Flow
+{
+    private readonly RetryPolicy _retryPolicy = RetryPolicy.CreateConstantDelay(interval: TimeSpan.FromSeconds(1), suspendThreshold: TimeSpan.Zero);
+    
+    public override async Task Run()
+    {
+        await Capture(async () =>
+        {
+            var i = await Effect.CreateOrGet("i", 0);
+            while (true)
+            {
+                Console.WriteLine($"Retrying: {i}");
+                i++;
+                await Effect.Upsert("i", i);
+                throw new TimeoutException();
+            }
+        }, _retryPolicy);
+    }
+}