Async support for When builder predicates (#6)

src/Flo/Builder.cs

@@ -59,6 +59,29 @@ public TBuilder When(
             });
         }
 
+        public TBuilder When(
+            Func<TIn, Task<bool>> predicate,
+            Func<TIn, Func<TIn, Task<TOut>>, Func<TIn, Task<TOut>>, Task<TOut>> handler,
+            Action<TBuilder> configurePipeline)
+        {
+            if (predicate == null) throw new ArgumentNullException(nameof(predicate));
+            if (handler == null) throw new ArgumentNullException(nameof(handler));
+            if (configurePipeline == null) throw new ArgumentNullException(nameof(configurePipeline));
+
+            return Add(async (input, next) =>
+            {
+                var doInvoke = await predicate.Invoke(input);
+                if (doInvoke)
+                {
+                    var builder = CreateBuilder();
+                    configurePipeline(builder);
+                    return await handler.Invoke(input, builder.Build(), next);
+                }
+
+                return await next.Invoke(input);
+            });
+        }
+
         protected abstract TBuilder CreateBuilder();
 
         public TBuilder Final(Func<TIn, Task<TOut>> handler)

src/Flo/OutputPipelineBuilder.cs

@@ -31,6 +31,24 @@ public OutputPipelineBuilder<TIn, TOut> When(
             configurePipeline);
         }
 
+        public OutputPipelineBuilder<TIn, TOut> When(
+            Func<TIn, Task<bool>> predicate,
+            Action<OutputPipelineBuilder<TIn, TOut>> configurePipeline,
+            Func<TOut, bool> continueIf = null)
+        {
+            return When(predicate, async (input, innerPipeline, next) =>
+            {
+                var result = await innerPipeline.Invoke(input);
+
+                // Continue on to the parent pipeline if the continueIf output predicate matches
+                if ((continueIf ?? DefaultContinuation).Invoke(result))
+                    return await next.Invoke(input);
+
+                return result;
+            },
+            configurePipeline);
+        }
+
         protected override OutputPipelineBuilder<TIn, TOut> CreateBuilder() => new OutputPipelineBuilder<TIn, TOut>(ServiceProvider);
     }
 }

src/Flo/PipelineBuilder.cs

@@ -24,6 +24,18 @@ public PipelineBuilder<T> When(
             configurePipeline);
         }
 
+        public PipelineBuilder<T> When(
+            Func<T, Task<bool>> predicate,
+            Action<PipelineBuilder<T>> configurePipeline)
+        {
+            return When(predicate, async (input, innerPipeline, next) =>
+            {
+                input = await innerPipeline.Invoke(input);
+                return await next.Invoke(input);
+            },
+            configurePipeline);
+        }
+
         protected override PipelineBuilder<T> CreateBuilder() => new PipelineBuilder<T>(ServiceProvider);
     }
 }

test/Flo.Tests/OutputPipelineBuilderWhenTests.cs

@@ -22,6 +22,21 @@ async Task it_ignores_the_handler_if_the_predicate_returns_false()
             result.ShouldBe(0);
         }
 
+        async Task it_ignores_the_handler_if_the_async_predicate_returns_false()
+        {
+            var pipeline = Pipeline.Build<string, int>(cfg =>
+                cfg.When(input => Task.FromResult(input == "hello world"),
+                    builder => builder.Add((input, next) =>
+                    {
+                        return Task.FromResult(input.Length);
+                    })
+                )
+            );
+
+            var result = await pipeline.Invoke("hello");
+            result.ShouldBe(0);
+        }
+
         async Task it_executes_the_handler_if_the_predicate_returns_true()
         {
             var pipeline = Pipeline.Build<string, int>(cfg =>
@@ -36,7 +51,22 @@ async Task it_executes_the_handler_if_the_predicate_returns_true()
             var result = await pipeline.Invoke("hello world");
             result.ShouldBe(11);
         }
-
+
+        async Task it_executes_the_handler_if_the_async_predicate_returns_true()
+        {
+            var pipeline = Pipeline.Build<string, int>(cfg =>
+                cfg.When(input => Task.FromResult(input == "hello world"),
+                    builder => builder.Add((input, next) =>
+                    {
+                        return Task.FromResult(input.Length);
+                    })
+                )
+            );
+
+            var result = await pipeline.Invoke("hello world");
+            result.ShouldBe(11);
+        }
+
         async Task it_continues_to_parent_pipeline_if_child_pipeline_returns_default_value()
         {
             var pipeline = Pipeline.Build<TestContext, TestContext>(cfg =>
@@ -56,6 +86,25 @@ async Task it_continues_to_parent_pipeline_if_child_pipeline_returns_default_val
             result.ShouldContainKey("Test");
         }
 
+        async Task it_continues_to_parent_pipeline_if_child_pipeline_returns_default_value_with_async_predicate()
+        {
+            var pipeline = Pipeline.Build<TestContext, TestContext>(cfg =>
+                cfg.When(input => Task.FromResult(true),
+                    builder => builder.Add((ctx, next) =>
+                    {
+                        return Task.FromResult(default(TestContext));
+                    })
+                )
+                .Final(ctx => {
+                    ctx.Add("Test", "TestValue");
+                    return Task.FromResult(ctx);
+                })
+            );
+
+            var result = await pipeline.Invoke(new TestContext());
+            result.ShouldContainKey("Test");
+        }
+
         async Task it_does_not_continue_to_parent_pipeline_if_child_pipeline_returns_value()
         {
             var pipeline = Pipeline.Build<TestContext, TestContext>(cfg =>
@@ -74,5 +123,24 @@ async Task it_does_not_continue_to_parent_pipeline_if_child_pipeline_returns_val
             var result = await pipeline.Invoke(new TestContext());
             result.ShouldNotContainKey("Test");
         }
+
+        async Task it_does_not_continue_to_parent_pipeline_if_child_pipeline_returns_value_with_async_predicate()
+        {
+            var pipeline = Pipeline.Build<TestContext, TestContext>(cfg =>
+                cfg.When(input => Task.FromResult(true),
+                    builder => builder.Add((ctx, next) =>
+                    {
+                        return Task.FromResult(ctx);
+                    })
+                )
+                .Final(ctx => {
+                    ctx.Add("Test", "TestValue");
+                    return Task.FromResult(ctx);
+                })
+            );
+
+            var result = await pipeline.Invoke(new TestContext());
+            result.ShouldNotContainKey("Test");
+        }
     }
 }