RtFlow.Pipelines.Extensions 1.0.0

RtFlow.Pipelines.Extensions

RtFlow.Pipelines.Extensions provides advanced extensions and utilities for RtFlow.Pipelines.Core, adding specialized operators and enhanced functionality for complex data processing scenarios.

Installation

# Install core package first
dotnet add package RtFlow.Pipelines.Core

# Then add extensions
dotnet add package RtFlow.Pipelines.Extensions

Key Features

• 🔄 Advanced Transforms - Specialized transformation operators for complex scenarios
• 📊 Aggregation Operations - Built-in aggregators and collectors
• 🔀 Flow Control - Advanced routing, splitting, and merging capabilities
• ⏱️ Time-based Operations - Windowing, throttling, and time-based batching
• 🧩 Custom Operators - Extended pipeline operations for specific use cases

Advanced Operators

Windowing Operations

using RtFlow.Pipelines.Extensions;

var pipeline = PipelineBuilder
    .Create<SensorReading>()
    .Window(TimeSpan.FromMinutes(5)) // 5-minute windows
    .Transform(window => CalculateAverage(window))
    .Build();

Throttling

var pipeline = PipelineBuilder
    .Create<ApiRequest>()
    .Throttle(maxPerSecond: 100) // Rate limiting
    .Transform(async req => await ProcessApiRequest(req))
    .Build();

Advanced Batching

var pipeline = PipelineBuilder
    .Create<LogEntry>()
    .BatchBy(entry => entry.Level) // Group by log level
    .Transform(batch => ProcessLogBatch(batch))
    .Build();

Conditional Routing

var pipeline = PipelineBuilder
    .Create<Order>()
    .Route(
        when: order => order.Priority == Priority.High,
        then: order => ProcessHighPriority(order),
        otherwise: order => ProcessNormalPriority(order)
    )
    .Build();

Aggregation and Collection

Built-in Aggregators

var pipeline = PipelineBuilder
    .Create<int>()
    .Aggregate(
        seed: 0,
        accumulator: (sum, value) => sum + value,
        windowSize: 100
    )
    .ForEach(sum => Console.WriteLine($"Running sum: {sum}"));

Collection Operations

var pipeline = PipelineBuilder
    .Create<DataPoint>()
    .Collect(capacity: 1000) // Collect up to 1000 items
    .Transform(collection => AnalyzeCollection(collection))
    .Build();

Time-based Processing

Sliding Windows

var pipeline = PipelineBuilder
    .Create<MetricValue>()
    .SlidingWindow(
        windowSize: TimeSpan.FromMinutes(10),
        slideInterval: TimeSpan.FromMinutes(1)
    )
    .Transform(window => CalculateMetrics(window))
    .Build();

Delayed Processing

var pipeline = PipelineBuilder
    .Create<Task>()
    .Delay(TimeSpan.FromSeconds(30)) // Process after 30 seconds
    .Transform(task => ExecuteTask(task))
    .Build();

Custom Extensions

You can create your own pipeline extensions:

public static class MyPipelineExtensions
{
    public static IPipelineBuilder<TOut> MyCustomTransform<TIn, TOut>(
        this IPipelineBuilder<TIn> builder,
        Func<TIn, TOut> transform)
    {
        return builder.Transform(transform);
    }
}

Requirements

• .NET 8.0 or later
• RtFlow.Pipelines.Core package

Documentation

For complete documentation and examples, visit the main project repository.

License

This project is licensed under the MIT License - see the LICENSE file for details.