Pipeline and Filters Pattern using C#

Introduction

Pipeline and filters is a very useful and neat pattern in the scenario when a set of filtering (processing) needs to be performed on an object to transform it into a useful state, as described below in this picture.

The code used in this article is the complete implementation of Pipeline and Filter pattern in a generic fashion.

Background

Understanding of messaging patterns is critical to understand Enterprise Integration patterns:

• http://www.enterpriseintegrationpatterns.com/patterns/messaging/PipesAndFilters.html

Using the Code

The code here considers there is an object repository AgentStatusRepository, which has a static property Agents to return List<Agent>, this list of AgentStatus is the input to the pipeline on which various filtering needs to happen.

AgentStatus class (Step 1)

public class AgentStatus
   {
       public string AgentId { get; set; }
       public string PresenceStatus { get; set; }
       public int CurrentWorkload { get; set; }
       public int MaxWorkload { get; set; }
       public DateTime StatusUpdateDatetime { get; set; }
   }

Possible Values of PresenceStatus:

• Available
• Busy
• Research
• Offline

And below is the list of filtering that needs to be performed on the List<AgentStatus> object (Object1 in the picture above), retrieved from the AgentStatusRepository:

• AgentAvailabilityFilter (Filter1 in the picture above) - Filter the List<Agent> where Agent.PresenceStatus == 'Available'
• AgentWorkloadFilter (Filter2 in the picture above) - Filter the list filtered from Filter1 where Agent.CurrentWorkload < Agent.MaxWorkload
• AgentPresenceUpdateDatetimeFilter (Filter3 in the picture above) - Filter the list filtered from Filter2 where (DateTime.UtcNow - Agent.StatusUpdateDatetime).TotalMinutes < 3

AgentStatusRepository code:

public class AgentStatusRepository
   {
       public static List<Agent> Agents
       {
           get
           {
               return new List<Agent>
               {
                   new Agent {AgentId = "agent_id_1",
                   PresenceStatus = "Available", CurrentWorkload = 1,
                   MaxWorkload = 2, StatusUpdateDatetime = DateTime.UtcNow },
                   new Agent {AgentId = "agent_id_2",
                   PresenceStatus = "Busy", CurrentWorkload = 2,
                   MaxWorkload = 2, StatusUpdateDatetime = DateTime.UtcNow },
                   new Agent {AgentId = "agent_id_3",
                   PresenceStatus = "Available", CurrentWorkload = 2,
                   MaxWorkload = 3, StatusUpdateDatetime = DateTime.UtcNow.AddMinutes (-5) },
                   new Agent {AgentId = "agent_id_4",
                   PresenceStatus = "Research", CurrentWorkload = 2,
                   MaxWorkload = 3, StatusUpdateDatetime = DateTime.UtcNow},
                   new Agent {AgentId = "agent_id_5",
                   PresenceStatus = "Available", CurrentWorkload = 2,
                   MaxWorkload = 3, StatusUpdateDatetime = DateTime.UtcNow },
                   new Agent {AgentId = "agent_id_6",
                   PresenceStatus = "Available", CurrentWorkload = 5,
                   MaxWorkload = 5, StatusUpdateDatetime = DateTime.UtcNow}
               };
           }
       }
   }

Pipeline and Filters Class Diagram:

IFilter code:

/// <summary>
    /// A filter to be registered in the message processing pipeline
    /// </summary>
    /// <typeparam name="T"></typeparam>
    public interface IFilter<T>
    {
        /// <summary>
        /// Filter implementing this method would perform processing on the input type T
        /// </summary>
        /// <param name="input">The input to be executed by the filter</param>
        /// <returns></returns>
        T Execute(T input);       
    }

Pipeline code:

/// <summary>
   /// An abstract Pipeline with a list of filters and abstract Process method
   /// </summary>
   /// <typeparam name="T"></typeparam>
   public abstract class Pipeline<T>
   {
       /// <summary>
       /// List of filters in the pipeline
       /// </summary>
       protected readonly List<IFilter<T>> filters = new List<IFilter<T>>();

       /// <summary>
       /// To Register filter in the pipeline
       /// </summary>
       /// <param name="filter">A filter object implementing IFilter interface</param>
       /// <returns></returns>
       public Pipeline<T> Register(IFilter<T> filter)
       {
           filters.Add(filter);
           return this;
       }

       /// <summary>
       /// To start processing on the Pipeline
       /// </summary>
       /// <param name="input">
       /// The input object on which filter processing would execute</param>
       /// <returns></returns>
       public abstract T Process(T input);
   }

AgentSelectionPipeline (ConcretePipeline) code:

/// <summary>
/// Pipeline which to select final list of applicable agents
/// </summary>
public class AgentSelectionPipeline : Pipeline<IEnumerable<Agent>>
{
    /// <summary>
    /// Method which executes the filter on a given Input
    /// </summary>
    /// <param name="input">Input on which filtering
    /// needs to happen as implementing in individual filters</param>
    /// <returns></returns>
    public override IEnumerable<Agent> Process(IEnumerable<Agent> input)
    {
        foreach (var filter in filters)
        {
            input = filter.Execute(input);
        }

        return input;
    }
}

AgentAvailabilityFilter code (ConcreteFilter 1):

/// <summary>
  /// This output of this filter is the list of all available agents
  /// </summary>
  public class AgentAvailabilityFilter : IFilter<IEnumerable<Agent>>
  {
      public IEnumerable<Agent> Execute(IEnumerable<Agent> input)
      {
          if (input == null || input.Count() < 1)
          {
              return input;
          }
          return input.Where(agent => agent.PresenceStatus == "Available");
      }
  }

AgentWorkloadFilter code (ConcreteFilter 2):

/// <summary>
/// The output of this filter is the list of Agent for which CurrentWorkload is less than MaxWorklod
/// </summary>
public class AgentWorkloadFilter : IFilter<IEnumerable<Agent>>
{
    public IEnumerable<Agent> Execute(IEnumerable<Agent> input)
    {
        if (input == null || input.Count() < 1)
        {
            return input;
        }

        return input.Where(agent => agent.CurrentWorkload < agent.MaxWorkload);
    }
}

AgentPresenceUpdateDatetimeFilter code (ConcreteFilter 3):

/// <summary>
/// The output of this filter is the list of
/// all those agents who has StatusUpdateDatetime less than 3 minutes
/// </summary>
public class AgentPresenceUpdateDatetimeFilter : IFilter<IEnumerable<Agent>>
{
    public IEnumerable<Agent> Execute(IEnumerable<Agent> input)
    {
        if (input == null || input.Count() < 1)
        {
            return input;
        }

        return input.Where(agent =>
        (DateTime.UtcNow - agent.StatusUpdateDatetime).TotalMinutes < 3) ;
    }
}

Pipeline and Filter, how to use code:

public class Program
{
    static void Main(string[] args)
    {
        //Get the Agents from repository
        var agentsStatus = AgentStatusRepository.Agents;

        //Construct the Pipeline object
        AgentSelectionPipeline agentStatusPipeline = new AgentSelectionPipeline();

        //Register the filters to be executed
         agentStatusPipeline.Register(new AgentAvailabilityFilter())
            .Register(new AgentWorkloadFilter())
            .Register(new AgentPresenceUpdateDatetimeFilter());

        //Start pipeline processing
        var agentsStatus_1 = agentStatusPipeline.Process(agentsStatus);
    }
}