Executing Chain of Responsibility variation

Question

I have got a pipeline of tasks which basically is a variation of chain of responsibility pattern.

A task in my pipeline looks as below

internal interface IPTask<T>
{
    bool CanExecute(T instance);
    T Process(T instance);
}

..and my processor looks like

internal interface IProcessor<T>
{
    T Execute(T instance);
}

and my concrete implementation looks like this:

public class Processor<T> : IProcessor<T>
{
    private readonly ITasks<T> tasks;

    public Processor(ITasks<T> tasks)
    {
        this.tasks= tasks;
    }

    public T Execute(T instance)
    {
         var taskstoExecute = tasks.GetTasks()
                                   .Where(task => task.CanExecute(instance));

         taskstoExecute.ToList().ForEach(task=>task.Process(instance));

         return T;
    }
}

..and my tasks look like this:

internal interface ITasks<T>
{
    IEnumerable<IPTask<T>> GetTasks();
}

T could be different instances, but bound by a generic contract. One of the tasks are about mapping the incoming object into a completely different object and forwarding that instance from there onwards.

Now as you would see, I'm executing all the tasks in the pipeline, I would like to modify this to the following:

• Input for the Execute method for the next task should be from the previously executed task.
• If the CanExecute fails for a task then the pipeline should stop processing the tasks.

Can you please help me in getting this. Also would you expect the code to be structured differently for this purpose?

Answer 1

How about this:

public T Execute(T instance)
{
     T result = instance;
     foreach(var individual in tasks.GetTasks())
     {
         if(!individual.CanExecute()) break;

         result = individual.Process(result);
     }

     return result;
}

As you have it currently, it is much more like a composite pattern than a chain of responsibility. This change makes it a little bit more CoR-ish. But it is more important to note whether it meets your needs than to use the correct design pattern jargon. :)

Answer 2

With this implementation, the CanProcess is actually used to trigger an exception that will stop the process.

I added a second implementation, called ExecuteWithPartial, that handles the exception, in case that's the behavior you are expecting. It processes but if there is an error, it returns the partial result up to that point.

public class Processor<T> : IProcessor<T>
{
    //... rest of your code

    public T Execute(T instance)
    {
        return this._tasks.GetTasks().Aggregate(instance, (current, task) => InternalExecute(task, current));
    }

    public T ExecuteWithPartial(T instance)
    {
        var target = instance;
        try
        {
            foreach (var task in this._tasks.GetTasks())
            {
                target = InternalExecute(task, target);
            }
            return target;
        }
        catch (CantExecuteException)
        {
            return target;
        }
    }


    private static T InternalExecute(IPTask<T> task, T instance)
    {
        if (!task.CanExecute(instance))
            throw new CantExecuteException();
        return task.Process(instance);
    }
}

And the new Exception class is:

public class CantExecuteException : Exception
{
}