NickDev.SimplePipeline 1.0.0

Basics

The pipeline model is composed of queues and steps. It starts with an "In" queue and finishes with an "Out" queue, in between is what you decide to do.

The simplest example would be :

• In - step - Out

Queues

Defining a queue is the easiest thing:

PipelineQueue("in")

without the queue size (defaults to 1000 items) or

PipelineQueue("in", 10)

with queue size

Steps

Two types of steps exist

• ActionStep : has an input queue and an output queue, this step does something to the item provided
• RouterStep : has an input queue and several output queues, this step routes the item to a queue based on condition

Defining a step depends on it's type:

PipelineAction(Func<IPipelineItem, IPipelineItem> function, string inputQName, string outputQName, [int parallelismDegree = 1])

The function of the action step returns the modified item which is placed automatically on the output queue

PipelineRouter(Func<IPipelineItem, string> function, string inputQName, string[] outputQNames, [int parallelismDegree ) 1])

The function of the router step returns the name of the output queue upon which to place the item

Pipeline Item

In order to process anything and everything, a special item interface is used. This interface must be implemented to allow items to be placed in the pipeline. A simple implementation would be :

public class item : IPipelineItem
{
    public ItemStatus Status { get; set; } = ItemStatus.New;
    public ItemTreatment Treatment { get; set; } = new ItemTreatment();
    public object Content { get; set; }
    public Type ContentType { get { return typeof(int); } set => throw new NotImplementedException(); }
    public TaskCompletionSource<IPipelineItem> TaskCompletionSource { get; set; } = new TaskCompletionSource<IPipelineItem>();
}

The content property holds the data you need to process

Pipeline

The pipeline itself is generated using a factory:

IAwaitablePipeline<IPipelineItem> PipelineFactory.CreatePipeline(string inQueue, string outQueue, IEnumerable<IPipelineStep> steps, IEnumerable<PipelineQueue> queues, Func<IPipelineItem, bool> logger)

The work starts when the first item is placed in the in queue

Task<IPipelineItem> IAwaitablePipeline<IPipelineItem>.Execute(IPipelineItem input)

Output

The output queue must be read in order for the pipeline to work, filling the queue will block the entire system. Example

Task.Run(() => {
    foreach (var o in outP.GetConsumingEnumerable()) 
    { 
        Console.WriteLine(string.Concat(o.Treatment.Message));
    } 
});