Performance Improvements

.claude/settings.local.json

@@ -0,0 +1,19 @@
+{
+  "permissions": {
+    "allow": [
+      "Bash(git restore:*)",
+      "Bash(rm:*)",
+      "Bash(dotnet script:*)",
+      "Bash(dotnet restore:*)",
+      "Bash(csc:*)",
+      "Bash(verify_changes.exe)",
+      "Bash(del verify_changes.cs)",
+      "Bash(dotnet run:*)",
+      "Bash(timeout 30 dotnet run:*)",
+      "Bash(.EnumerableAsyncProcessor.Example.exe)",
+      "Bash(EnumerableAsyncProcessor.Example.exe)",
+      "Bash(del \"EnumerableAsyncProcessor.UnitTests\\ValidationTests.cs\")"
+    ],
+    "deny": []
+  }
+}

EnumerableAsyncProcessor.Example/ChannelProcessingExamples.cs

@@ -0,0 +1,237 @@
+#if NET6_0_OR_GREATER
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using EnumerableAsyncProcessor.Extensions;
+
+namespace EnumerableAsyncProcessor.Example;
+
+/// <summary>
+/// Examples demonstrating channel-based processing capabilities
+/// </summary>
+public static class ChannelProcessingExamples
+{
+    /// <summary>
+    /// Demonstrates basic channel-based processing with multiple consumers
+    /// </summary>
+    public static async Task BasicChannelProcessingExample()
+    {
+        Console.WriteLine("=== Basic Channel Processing Example ===");
+
+        var urls = new[]
+        {
+            "https://api.example.com/data/1",
+            "https://api.example.com/data/2", 
+            "https://api.example.com/data/3",
+            "https://api.example.com/data/4",
+            "https://api.example.com/data/5",
+            "https://api.example.com/data/6"
+        };
+
+        // Create channel options with multiple consumers
+        var options = ChannelProcessorOptions.CreateUnbounded(consumerCount: 3);
+
+        var processedCount = 0;
+
+        // Process URLs with multiple consumers working in parallel
+        var processor = urls.ForEachWithChannelAsync(async url =>
+        {
+            // Simulate HTTP request
+            await Task.Delay(Random.Shared.Next(100, 500));
+
+            var count = Interlocked.Increment(ref processedCount);
+            Console.WriteLine($"[Consumer {Environment.CurrentManagedThreadId}] Processed {url} ({count}/{urls.Length})");
+
+        }, options);
+
+        var stopwatch = Stopwatch.StartNew();
+        await processor;
+        stopwatch.Stop();
+
+        Console.WriteLine($"Processed {urls.Length} URLs in {stopwatch.ElapsedMilliseconds}ms using multiple consumers");
+    }
+
+    /// <summary>
+    /// Demonstrates bounded channel with backpressure handling
+    /// </summary>
+    public static async Task BoundedChannelWithBackpressureExample()
+    {
+        Console.WriteLine("\n=== Bounded Channel with Backpressure Example ===");
+
+        var documents = Enumerable.Range(1, 20).Select(i => $"Document_{i:D3}.pdf").ToArray();
+
+        // Create bounded channel with limited capacity
+        var options = ChannelProcessorOptions.CreateBounded(
+            capacity: 3,           // Small buffer to demonstrate backpressure
+            consumerCount: 1       // Single slow consumer
+        );
+
+        var processedDocs = new List<string>();
+        var lockObj = new object();
+
+        var processor = documents.ForEachWithChannelAsync(async doc =>
+        {
+            // Simulate slow document processing
+            Console.WriteLine($"Processing {doc}...");
+            await Task.Delay(200);
+
+            lock (lockObj)
+            {
+                processedDocs.Add(doc);
+            }
+
+            Console.WriteLine($"✓ Completed {doc} ({processedDocs.Count}/{documents.Length})");
+
+        }, options);
+
+        var stopwatch = Stopwatch.StartNew();
+        await processor;
+        stopwatch.Stop();
+
+        Console.WriteLine($"Processed {documents.Length} documents in {stopwatch.ElapsedMilliseconds}ms");
+        Console.WriteLine("Note: Producer was throttled by bounded channel capacity");
+    }
+
+    /// <summary>
+    /// Demonstrates result processing with streaming output
+    /// </summary>
+    public static async Task ResultProcessingWithStreamingExample()
+    {
+        Console.WriteLine("\n=== Result Processing with Streaming Example ===");
+
+        var numbers = Enumerable.Range(1, 10).ToArray();
+
+        // Process numbers and return results using multiple consumers
+        var options = ChannelProcessorOptions.CreateUnbounded(consumerCount: 4);
+
+        var processor = numbers.SelectWithChannelAsync(async number =>
+        {
+            // Simulate CPU-intensive calculation
+            await Task.Delay(Random.Shared.Next(50, 200));
+
+            var result = Math.Pow(number, 3); // Calculate cube
+            Console.WriteLine($"[Consumer {Environment.CurrentManagedThreadId}] {number}³ = {result}");
+
+            return new { Number = number, Cube = result };
+
+        }, options);
+
+        Console.WriteLine("Streaming results as they become available:");
+
+        var results = new List<object>();
+        await foreach (var result in processor.GetResultsAsyncEnumerable())
+        {
+            results.Add(result);
+            Console.WriteLine($"Got result: {result.Number}³ = {result.Cube}");
+        }
+
+        Console.WriteLine($"Total results received: {results.Count}");
+    }
+
+    /// <summary>
+    /// Demonstrates channel options configuration
+    /// </summary>
+    public static async Task ChannelOptionsConfigurationExample()
+    {
+        Console.WriteLine("\n=== Channel Options Configuration Example ===");
+
+        var tasks = Enumerable.Range(1, 15);
+
+        // Configure channel with specific options
+        var options = new ChannelProcessorOptions
+        {
+            Capacity = 5,                                    // Bounded channel
+            FullMode = BoundedChannelFullMode.Wait,         // Block producer when full
+            ConsumerCount = 3,                              // Multiple consumers
+            SingleWriter = true,                            // Optimize for single producer
+            SingleReader = false,                           // Multiple readers
+            AllowSynchronousContinuations = false          // Use thread pool for continuations
+        };
+
+        var processedTasks = new List<int>();
+        var lockObj = new object();
+
+        var processor = tasks.ForEachWithChannelAsync(async task =>
+        {
+            Console.WriteLine($"[Consumer {Environment.CurrentManagedThreadId}] Starting task {task}");
+            await Task.Delay(100);
+
+            lock (lockObj)
+            {
+                processedTasks.Add(task);
+            }
+
+            Console.WriteLine($"[Consumer {Environment.CurrentManagedThreadId}] Completed task {task}");
+
+        }, options);
+
+        await processor;
+
+        Console.WriteLine($"Processed {processedTasks.Count} tasks with custom channel configuration");
+        Console.WriteLine($"Tasks processed: [{string.Join(", ", processedTasks.OrderBy(x => x))}]");
+    }
+
+    /// <summary>
+    /// Performance comparison between channel-based and batch processing
+    /// </summary>
+    public static async Task PerformanceComparisonExample()
+    {
+        Console.WriteLine("\n=== Performance Comparison Example ===");
+
+        var workItems = Enumerable.Range(1, 200).ToArray();
+
+        // Test channel-based processing
+        Console.WriteLine("Testing Channel-based processing...");
+        var channelOptions = ChannelProcessorOptions.CreateUnbounded(consumerCount: 4);
+        var channelStopwatch = Stopwatch.StartNew();
+
+        var channelProcessor = workItems.ForEachWithChannelAsync(async item =>
+        {
+            await Task.Delay(5); // Simulate work
+        }, channelOptions);
+
+        await channelProcessor;
+        channelStopwatch.Stop();
+
+        // Test traditional batch processing
+        Console.WriteLine("Testing Batch processing...");
+        var batchStopwatch = Stopwatch.StartNew();
+
+        var batchProcessor = workItems.ForEachAsync(async item =>
+        {
+            await Task.Delay(5); // Simulate work
+        }).ProcessInBatches(4);
+
+        await batchProcessor;
+        batchStopwatch.Stop();
+
+        // Compare results
+        Console.WriteLine($"Channel-based: {channelStopwatch.ElapsedMilliseconds}ms");
+        Console.WriteLine($"Batch-based:   {batchStopwatch.ElapsedMilliseconds}ms");
+        Console.WriteLine($"Performance ratio: {(double)channelStopwatch.ElapsedMilliseconds / batchStopwatch.ElapsedMilliseconds:F2}x");
+
+        if (channelStopwatch.ElapsedMilliseconds < batchStopwatch.ElapsedMilliseconds * 1.2)
+        {
+            Console.WriteLine("✓ Channel-based processing is competitive with batch processing");
+        }
+    }
+
+    /// <summary>
+    /// Run all channel processing examples
+    /// </summary>
+    public static async Task RunAllExamples()
+    {
+        await BasicChannelProcessingExample();
+        await BoundedChannelWithBackpressureExample();
+        await ResultProcessingWithStreamingExample();
+        await ChannelOptionsConfigurationExample();
+        await PerformanceComparisonExample();
+
+        Console.WriteLine("\n=== All Channel Processing Examples Completed ===");
+    }
+}
+#endif

EnumerableAsyncProcessor.Example/Program.cs

@@ -1,5 +1,6 @@
 using EnumerableAsyncProcessor.Builders;
 using EnumerableAsyncProcessor.Extensions;
+using EnumerableAsyncProcessor.Example;
 
 async Task ItemAsyncProcessor()
 {
@@ -63,4 +64,9 @@ Task<HttpResponseMessage> PingAsync()
     {
         return httpClient.GetAsync("https://localhost:8080/ping");
     }
-}
+}
+
+#if NET6_0_OR_GREATER
+// Run channel processing examples
+await ChannelProcessingExamples.RunAllExamples();
+#endif