Pipeline.cs

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
 
namespace MPewsey.Common.Pipelines
{
    public class Pipeline
    {
        public List<IPipelineStep> Steps { get; private set; } = new List<IPipelineStep>();
 
        public Pipeline(params IPipelineStep[] steps)
        {
            Steps = new List<IPipelineStep>(steps);
        }
 
        public PipelineResults Run(Dictionary<string, object> inputs, Action<string> logger = null, CancellationToken cancellationToken = default)
        {
            logger?.Invoke("[Pipeline] Running pipeline...");
            var results = new PipelineResults(inputs);
 
            foreach (var step in Steps)
            {
                if (cancellationToken.IsCancellationRequested)
                {
                    logger?.Invoke("[Pipeline] Cancelled process.");
                    return results;
                }
 
                if (!step.ApplyStep(results, logger, cancellationToken))
                {
                    logger?.Invoke("[Pipeline] Pipeline failed.");
                    return results;
                }
            }
 
            results.Complete();
            logger?.Invoke("[Pipeline] Pipeline complete.");
            return results;
        }
 
        public Task<PipelineResults> RunAsync(Dictionary<string, object> inputs, Action<string> logger = null, CancellationToken cancellationToken = default)
        {
            return Task.Run(() => Run(inputs, logger, cancellationToken));
        }
    }
}