diff --git a/EnumerableAsyncProcessor.Example/AsyncEnumerableExample.cs b/EnumerableAsyncProcessor.Example/AsyncEnumerableExample.cs
new file mode 100644
index 0000000..fe69bf1
--- /dev/null
+++ b/EnumerableAsyncProcessor.Example/AsyncEnumerableExample.cs
@@ -0,0 +1,78 @@
+using EnumerableAsyncProcessor.Extensions;
+
+namespace EnumerableAsyncProcessor.Example;
+
+public static class AsyncEnumerableExample
+{
+    public static async Task RunExamples()
+    {
+        Console.WriteLine("=== IAsyncEnumerable Parallel Processing Examples ===\n");
+
+        // Example 1: ForEachAsync with parallel processing
+        Console.WriteLine("Example 1: Processing async enumerable items in parallel");
+        var items = GenerateAsyncNumbers(10);
+        
+        await items
+            .ForEachAsync(async number =>
+            {
+                await Task.Delay(100); // Simulate I/O work
+                Console.WriteLine($"Processed {number} on thread {Thread.CurrentThread.ManagedThreadId}");
+            })
+            .ProcessInParallel(3)
+            .ExecuteAsync(); // Process with max 3 concurrent tasks
+
+        Console.WriteLine("\nExample 2: SelectAsync with transformation");
+        var transformedItems = GenerateAsyncNumbers(5);
+        
+        var results = await transformedItems
+            .SelectAsync(async number =>
+            {
+                await Task.Delay(50);
+                return number * 2;
+            })
+            .ProcessInParallel(2)
+            .ExecuteAsync()
+            .ToListAsync();
+
+        Console.WriteLine($"Transformed results: {string.Join(", ", results)}");
+
+        // Example 3: High concurrency for I/O-bound operations
+        Console.WriteLine("\nExample 3: High concurrency I/O operations");
+        var ioItems = GenerateAsyncNumbers(20);
+        
+        await ioItems
+            .ForEachAsync(async number =>
+            {
+                await SimulateApiCall(number);
+                Console.WriteLine($"API call {number} completed");
+            })
+            .ProcessInParallelForIO(10)
+            .ExecuteAsync(); // Optimized for I/O with high concurrency
+
+        Console.WriteLine("\nAll examples completed!");
+    }
+
+    private static async IAsyncEnumerable<int> GenerateAsyncNumbers(int count)
+    {
+        for (int i = 1; i <= count; i++)
+        {
+            await Task.Yield(); // Simulate async generation
+            yield return i;
+        }
+    }
+
+    private static async Task SimulateApiCall(int id)
+    {
+        await Task.Delay(Random.Shared.Next(10, 50));
+    }
+
+    private static async Task<List<T>> ToListAsync<T>(this IAsyncEnumerable<T> source)
+    {
+        var list = new List<T>();
+        await foreach (var item in source)
+        {
+            list.Add(item);
+        }
+        return list;
+    }
+}
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor.Example/Program.cs b/EnumerableAsyncProcessor.Example/Program.cs
index 1d7c28d..56d1a56 100644
--- a/EnumerableAsyncProcessor.Example/Program.cs
+++ b/EnumerableAsyncProcessor.Example/Program.cs
@@ -69,4 +69,8 @@ Task<HttpResponseMessage> PingAsync()
 #if NET6_0_OR_GREATER
 // Run channel processing examples
 await ChannelProcessingExamples.RunAllExamples();
+
+// Run IAsyncEnumerable examples
+Console.WriteLine("\n\n=== Running IAsyncEnumerable Examples ===\n");
+await AsyncEnumerableExample.RunExamples();
 #endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor.UnitTests/AsyncEnumerableProcessorTests.cs b/EnumerableAsyncProcessor.UnitTests/AsyncEnumerableProcessorTests.cs
new file mode 100644
index 0000000..800ffe1
--- /dev/null
+++ b/EnumerableAsyncProcessor.UnitTests/AsyncEnumerableProcessorTests.cs
@@ -0,0 +1,368 @@
+#if NET6_0_OR_GREATER
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using EnumerableAsyncProcessor.Extensions;
+using System.Runtime.CompilerServices;
+using TUnit.Assertions;
+using TUnit.Core;
+
+namespace EnumerableAsyncProcessor.UnitTests;
+
+public class AsyncEnumerableProcessorTests
+{
+    private static async IAsyncEnumerable<int> GenerateAsyncEnumerable(int count, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        for (int i = 1; i <= count; i++)
+        {
+            await Task.Yield();
+            cancellationToken.ThrowIfCancellationRequested();
+            yield return i;
+        }
+    }
+
+    [Test]
+    public async Task ForEachAsync_ProcessOneAtATime_ProcessesAllItems()
+    {
+        var processedItems = new List<int>();
+        var asyncEnumerable = GenerateAsyncEnumerable(10);
+
+        await asyncEnumerable
+            .ForEachAsync(async item =>
+            {
+                await Task.Delay(10);
+                lock (processedItems)
+                {
+                    processedItems.Add(item);
+                }
+            })
+            .ProcessOneAtATime()
+            .ExecuteAsync();
+
+        await Assert.That(processedItems.Count).IsEqualTo(10);
+        await Assert.That(processedItems).IsEquivalentTo(Enumerable.Range(1, 10));
+    }
+
+    [Test]
+    public async Task ForEachAsync_ProcessInParallel_ProcessesAllItems()
+    {
+        var processedItems = new List<int>();
+        var asyncEnumerable = GenerateAsyncEnumerable(20);
+
+        await asyncEnumerable
+            .ForEachAsync(async item =>
+            {
+                await Task.Delay(10);
+                lock (processedItems)
+                {
+                    processedItems.Add(item);
+                }
+            })
+            .ProcessInParallel(5)
+            .ExecuteAsync();
+
+        await Assert.That(processedItems.Count).IsEqualTo(20);
+        await Assert.That(processedItems.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 20));
+    }
+
+    [Test]
+    public async Task SelectAsync_ProcessOneAtATime_ReturnsTransformedItems()
+    {
+        var asyncEnumerable = GenerateAsyncEnumerable(5);
+
+        var results = await asyncEnumerable
+            .SelectAsync(async item =>
+            {
+                await Task.Delay(10);
+                return item * 2;
+            })
+            .ProcessOneAtATime()
+            .ExecuteAsync()
+            .ToListAsync();
+
+        await Assert.That(results.Count).IsEqualTo(5);
+        await Assert.That(results).IsEquivalentTo(new[] { 2, 4, 6, 8, 10 });
+    }
+
+    [Test]
+    public async Task SelectAsync_ProcessInParallel_ReturnsAllTransformedItems()
+    {
+        var asyncEnumerable = GenerateAsyncEnumerable(10);
+
+        var results = await asyncEnumerable
+            .SelectAsync(async item =>
+            {
+                await Task.Delay(10);
+                return item * 2;
+            })
+            .ProcessInParallel(3)
+            .ExecuteAsync()
+            .ToListAsync();
+
+        await Assert.That(results.Count).IsEqualTo(10);
+        await Assert.That(results.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 10).Select(x => x * 2));
+    }
+
+    [Test]
+    public async Task ForEachAsync_ProcessInParallelForIO_HandlesHighConcurrency()
+    {
+        var processedCount = 0;
+        var asyncEnumerable = GenerateAsyncEnumerable(100);
+
+        await asyncEnumerable
+            .ForEachAsync(async item =>
+            {
+                await Task.Delay(5);
+                Interlocked.Increment(ref processedCount);
+            })
+            .ProcessInParallelForIO(50)
+            .ExecuteAsync();
+
+        await Assert.That(processedCount).IsEqualTo(100);
+    }
+
+    [Test]
+    public async Task SelectAsync_ProcessInParallelForIO_HandlesHighConcurrency()
+    {
+        var asyncEnumerable = GenerateAsyncEnumerable(50);
+
+        var results = await asyncEnumerable
+            .SelectAsync(async item =>
+            {
+                await Task.Delay(5);
+                return item * 3;
+            })
+            .ProcessInParallelForIO(25)
+            .ExecuteAsync()
+            .ToListAsync();
+
+        await Assert.That(results.Count).IsEqualTo(50);
+        await Assert.That(results.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 50).Select(x => x * 3));
+    }
+
+    [Test]
+    public async Task ForEachAsync_ProcessWithChannel_ProcessesAllItems()
+    {
+        var processedItems = new List<int>();
+        var asyncEnumerable = GenerateAsyncEnumerable(30);
+
+        await asyncEnumerable
+            .ForEachAsync(async item =>
+            {
+                await Task.Delay(5);
+                lock (processedItems)
+                {
+                    processedItems.Add(item);
+                }
+            })
+            .ProcessWithChannel(new AsyncEnumerableChannelOptions 
+            { 
+                BufferSize = 10,
+                MaxConcurrency = 5
+            })
+            .ExecuteAsync();
+
+        await Assert.That(processedItems.Count).IsEqualTo(30);
+        await Assert.That(processedItems.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 30));
+    }
+
+    [Test]
+    public async Task SelectAsync_ProcessWithChannel_WithOrderPreservation_MaintainsOrder()
+    {
+        var asyncEnumerable = GenerateAsyncEnumerable(20);
+        var random = new Random(42);
+
+        var results = await asyncEnumerable
+            .SelectAsync(async item =>
+            {
+                // Random delay to test order preservation
+                await Task.Delay(random.Next(1, 20));
+                return item * 2;
+            })
+            .ProcessWithChannel(new AsyncEnumerableChannelOptions 
+            { 
+                BufferSize = 5,
+                MaxConcurrency = 4,
+                PreserveOrder = true
+            })
+            .ExecuteAsync()
+            .ToListAsync();
+
+        await Assert.That(results.Count).IsEqualTo(20);
+        await Assert.That(results).IsEquivalentTo(Enumerable.Range(1, 20).Select(x => x * 2));
+    }
+
+    [Test]
+    public async Task SelectAsync_ProcessWithChannel_WithoutOrderPreservation_ReturnsAllItems()
+    {
+        var asyncEnumerable = GenerateAsyncEnumerable(20);
+
+        var results = await asyncEnumerable
+            .SelectAsync(async item =>
+            {
+                await Task.Delay(5);
+                return item * 2;
+            })
+            .ProcessWithChannel(new AsyncEnumerableChannelOptions 
+            { 
+                PreserveOrder = false,
+                MaxConcurrency = 4
+            })
+            .ExecuteAsync()
+            .ToListAsync();
+
+        await Assert.That(results.Count).IsEqualTo(20);
+        await Assert.That(results.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 20).Select(x => x * 2));
+    }
+
+    [Test]
+    public async Task ForEachAsync_WithCancellation_StopsProcessing()
+    {
+        var cts = new CancellationTokenSource();
+        var processedCount = 0;
+        var asyncEnumerable = GenerateAsyncEnumerable(100, cts.Token);
+
+        var task = asyncEnumerable
+            .ForEachAsync(async item =>
+            {
+                // Check cancellation before processing
+                if (cts.Token.IsCancellationRequested)
+                    return;
+                    
+                if (item == 10)
+                {
+                    cts.Cancel();
+                }
+                
+                await Task.Delay(10);
+                
+                // Check cancellation after delay
+                if (cts.Token.IsCancellationRequested)
+                    return;
+                    
+                Interlocked.Increment(ref processedCount);
+            }, cts.Token)
+            .ProcessInParallel(5)
+            .ExecuteAsync();
+
+        await Assert.ThrowsAsync<OperationCanceledException>(async () => await task);
+        await Assert.That(processedCount).IsLessThan(100);
+    }
+
+    [Test]
+    public async Task SelectAsync_WithEmptyAsyncEnumerable_ReturnsEmptyResults()
+    {
+        var asyncEnumerable = GenerateAsyncEnumerable(0);
+
+        var results = await asyncEnumerable
+            .SelectAsync(async item =>
+            {
+                await Task.Delay(10);
+                return item * 2;
+            })
+            .ProcessInParallel(5)
+            .ExecuteAsync()
+            .ToListAsync();
+
+        await Assert.That(results).IsEmpty();
+    }
+
+    [Test]
+    public async Task ForEachAsync_ProcessInParallelUnbounded_ProcessesAllItems()
+    {
+        var processedItems = new List<int>();
+        var maxConcurrency = 0;
+        var currentConcurrency = 0;
+        var asyncEnumerable = GenerateAsyncEnumerable(50);
+
+        await asyncEnumerable
+            .ForEachAsync(async item =>
+            {
+                var current = Interlocked.Increment(ref currentConcurrency);
+                lock (processedItems)
+                {
+                    if (current > maxConcurrency)
+                        maxConcurrency = current;
+                }
+                
+                await Task.Delay(10);
+                lock (processedItems)
+                {
+                    processedItems.Add(item);
+                }
+                Interlocked.Decrement(ref currentConcurrency);
+            })
+            .ProcessInParallelUnbounded()
+            .ExecuteAsync();
+
+        await Assert.That(processedItems.Count).IsEqualTo(50);
+        await Assert.That(processedItems.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 50));
+        // Should have high concurrency since it's unbounded (at least 5)
+        await Assert.That(maxConcurrency).IsGreaterThan(5);
+    }
+
+    [Test]
+    public async Task SelectAsync_ProcessInParallelUnbounded_ReturnsAllResults()
+    {
+        var asyncEnumerable = GenerateAsyncEnumerable(30);
+        var maxConcurrency = 0;
+        var currentConcurrency = 0;
+
+        var results = await asyncEnumerable
+            .SelectAsync(async item =>
+            {
+                var current = Interlocked.Increment(ref currentConcurrency);
+                if (current > maxConcurrency)
+                    maxConcurrency = current;
+                
+                await Task.Delay(10);
+                Interlocked.Decrement(ref currentConcurrency);
+                return item * 3;
+            })
+            .ProcessInParallelUnbounded()
+            .ExecuteAsync()
+            .ToListAsync();
+
+        await Assert.That(results.Count).IsEqualTo(30);
+        await Assert.That(results.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 30).Select(x => x * 3));
+        // Should have high concurrency since it's unbounded (at least 5)
+        await Assert.That(maxConcurrency).IsGreaterThan(5);
+    }
+
+    [Test]
+    public async Task ForEachAsync_WithException_PropagatesException()
+    {
+        var asyncEnumerable = GenerateAsyncEnumerable(10);
+
+        var task = asyncEnumerable
+            .ForEachAsync(async item =>
+            {
+                await Task.Delay(10);
+                if (item == 5)
+                {
+                    throw new InvalidOperationException("Test exception");
+                }
+            })
+            .ProcessInParallel(3)
+            .ExecuteAsync();
+
+        var exception = await Assert.ThrowsAsync<InvalidOperationException>(async () => await task);
+        await Assert.That(exception!.Message).IsEqualTo("Test exception");
+    }
+}
+
+internal static class AsyncEnumerableExtensionsForTests
+{
+    public static async Task<List<T>> ToListAsync<T>(this IAsyncEnumerable<T> source)
+    {
+        var list = new List<T>();
+        await foreach (var item in source)
+        {
+            list.Add(item);
+        }
+        return list;
+    }
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor.UnitTests/ChannelBasedAsyncProcessorTests.cs b/EnumerableAsyncProcessor.UnitTests/ChannelBasedAsyncProcessorTests.cs
index 6e4f9f9..a096503 100644
--- a/EnumerableAsyncProcessor.UnitTests/ChannelBasedAsyncProcessorTests.cs
+++ b/EnumerableAsyncProcessor.UnitTests/ChannelBasedAsyncProcessorTests.cs
@@ -160,7 +160,13 @@ public async Task ProcessWithChannel_WithCancellation_ShouldCancelGracefully()
         // Act
         var processor = items.ForEachWithChannelAsync(async item =>
         {
-            await Task.Delay(50, cts.Token);
+            // Don't pass the token to Task.Delay to allow some items to complete
+            await Task.Delay(10);
+            
+            // Check cancellation after the delay
+            if (cts.Token.IsCancellationRequested)
+                return;
+            
             lock (lockObj)
             {
                 processedItems.Add(item);
@@ -233,47 +239,6 @@ public async Task ProcessWithChannel_WithGetResultsAsyncEnumerable_ShouldStreamR
         await Assert.That(streamedResults.OrderBy(x => x)).IsEquivalentTo(expectedResults.OrderBy(x => x));
     }
 
-    [Test]
-    public async Task ProcessWithChannel_Performance_ComparedToBatchProcessor()
-    {
-        // Arrange
-        const int itemCount = 1000;
-        var items = Enumerable.Range(1, itemCount);
-        
-        // Test channel-based processor
-        var channelOptions = ChannelProcessorOptions.CreateUnbounded(consumerCount: 4);
-        var channelStopwatch = Stopwatch.StartNew();
-        
-        var channelProcessor = items.ForEachWithChannelAsync(async item =>
-        {
-            await Task.Delay(1);
-        }, channelOptions);
-        
-        await channelProcessor;
-        channelStopwatch.Stop();
-        
-        // Test batch processor for comparison
-        var batchStopwatch = Stopwatch.StartNew();
-        
-        var batchProcessor = items.ForEachAsync(async item =>
-        {
-            await Task.Delay(1);
-        }).ProcessInBatches(4);
-        
-        await batchProcessor;
-        batchStopwatch.Stop();
-        
-        // Assert - both should complete successfully
-        // Performance comparison is informational
-        await Assert.That(channelStopwatch.ElapsedMilliseconds).IsGreaterThan(0);
-        await Assert.That(batchStopwatch.ElapsedMilliseconds).IsGreaterThan(0);
-        
-        // Channel-based should be competitive with batch processing
-        // Allow for some variance in timing
-        var ratio = (double)channelStopwatch.ElapsedMilliseconds / batchStopwatch.ElapsedMilliseconds;
-        await Assert.That(ratio).IsLessThan(3.0); // Channel should not be more than 3x slower
-    }
-
     [Test]
     public async Task ProcessWithChannel_BoundedChannelFullModeWait_ShouldBlockProducerWhenFull()
     {
diff --git a/EnumerableAsyncProcessor/AsyncEnumerableChannelOptions.cs b/EnumerableAsyncProcessor/AsyncEnumerableChannelOptions.cs
new file mode 100644
index 0000000..ad35a6f
--- /dev/null
+++ b/EnumerableAsyncProcessor/AsyncEnumerableChannelOptions.cs
@@ -0,0 +1,26 @@
+#if NET6_0_OR_GREATER
+namespace EnumerableAsyncProcessor;
+
+public class AsyncEnumerableChannelOptions
+{
+    /// <summary>
+    /// Buffer size for the channel. If null, uses unbounded channel.
+    /// </summary>
+    public int? BufferSize { get; set; }
+    
+    /// <summary>
+    /// Number of concurrent consumers processing items from the channel.
+    /// </summary>
+    public int MaxConcurrency { get; set; } = Environment.ProcessorCount;
+    
+    /// <summary>
+    /// Whether to preserve the order of results (for SelectAsync operations).
+    /// </summary>
+    public bool PreserveOrder { get; set; } = false;
+    
+    /// <summary>
+    /// Whether tasks are I/O-bound (true) or CPU-bound (false).
+    /// </summary>
+    public bool IsIOBound { get; set; } = true;
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs b/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs
new file mode 100644
index 0000000..9c669ff
--- /dev/null
+++ b/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs
@@ -0,0 +1,79 @@
+#if NET6_0_OR_GREATER
+using EnumerableAsyncProcessor.Extensions;
+using EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable;
+
+namespace EnumerableAsyncProcessor.Builders;
+
+public class AsyncEnumerableActionAsyncProcessorBuilder<TInput>
+{
+    private readonly IAsyncEnumerable<TInput> _items;
+    private readonly Func<TInput, Task> _taskSelector;
+    private readonly CancellationTokenSource _cancellationTokenSource;
+
+    public AsyncEnumerableActionAsyncProcessorBuilder(
+        IAsyncEnumerable<TInput> items, 
+        Func<TInput, Task> taskSelector, 
+        CancellationToken cancellationToken)
+    {
+        _items = items;
+        _taskSelector = taskSelector;
+        _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+    }
+
+    /// <summary>
+    /// Process items in parallel with a specified level of parallelism.
+    /// </summary>
+    public IAsyncEnumerableProcessor ProcessInParallel(int maxConcurrency)
+    {
+        return new AsyncEnumerableParallelProcessor<TInput>(
+            _items, _taskSelector, maxConcurrency, _cancellationTokenSource);
+    }
+    
+    /// <summary>
+    /// Process items in parallel with default concurrency (processor count).
+    /// </summary>
+    public IAsyncEnumerableProcessor ProcessInParallel()
+    {
+        return ProcessInParallel(Environment.ProcessorCount);
+    }
+    
+    /// <summary>
+    /// Process items in parallel optimized for I/O-bound operations.
+    /// </summary>
+    public IAsyncEnumerableProcessor ProcessInParallelForIO(int? maxConcurrency = null)
+    {
+        var concurrency = maxConcurrency ?? Math.Max(Environment.ProcessorCount * 10, 100);
+        return new AsyncEnumerableIOBoundParallelProcessor<TInput>(
+            _items, _taskSelector, concurrency, _cancellationTokenSource);
+    }
+    
+    /// <summary>
+    /// Process items one at a time (sequential processing).
+    /// </summary>
+    public IAsyncEnumerableProcessor ProcessOneAtATime()
+    {
+        return new AsyncEnumerableOneAtATimeProcessor<TInput>(
+            _items, _taskSelector, _cancellationTokenSource);
+    }
+
+    /// <summary>
+    /// Process ALL items in parallel without any concurrency limits.
+    /// WARNING: Use with caution - can overwhelm system resources with large async enumerables.
+    /// </summary>
+    public IAsyncEnumerableProcessor ProcessInParallelUnbounded()
+    {
+        return new AsyncEnumerableUnboundedParallelProcessor<TInput>(
+            _items, _taskSelector, _cancellationTokenSource);
+    }
+
+    /// <summary>
+    /// Process items using a channel-based approach with producer-consumer pattern.
+    /// </summary>
+    public IAsyncEnumerableProcessor ProcessWithChannel(AsyncEnumerableChannelOptions? options = null)
+    {
+        options ??= new AsyncEnumerableChannelOptions();
+        return new AsyncEnumerableChannelBasedProcessor<TInput>(
+            _items, _taskSelector, _cancellationTokenSource, options);
+    }
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs b/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs
new file mode 100644
index 0000000..9be8a3d
--- /dev/null
+++ b/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs
@@ -0,0 +1,79 @@
+#if NET6_0_OR_GREATER
+using EnumerableAsyncProcessor.Extensions;
+using EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable.ResultProcessors;
+
+namespace EnumerableAsyncProcessor.Builders;
+
+public class AsyncEnumerableActionAsyncProcessorBuilder<TInput, TOutput>
+{
+    private readonly IAsyncEnumerable<TInput> _items;
+    private readonly Func<TInput, Task<TOutput>> _taskSelector;
+    private readonly CancellationTokenSource _cancellationTokenSource;
+
+    public AsyncEnumerableActionAsyncProcessorBuilder(
+        IAsyncEnumerable<TInput> items, 
+        Func<TInput, Task<TOutput>> taskSelector, 
+        CancellationToken cancellationToken)
+    {
+        _items = items;
+        _taskSelector = taskSelector;
+        _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+    }
+
+    /// <summary>
+    /// Process items in parallel with a specified level of parallelism and return results.
+    /// </summary>
+    public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(int maxConcurrency)
+    {
+        return new ResultAsyncEnumerableParallelProcessor<TInput, TOutput>(
+            _items, _taskSelector, maxConcurrency, _cancellationTokenSource);
+    }
+    
+    /// <summary>
+    /// Process items in parallel with default concurrency and return results.
+    /// </summary>
+    public IAsyncEnumerableProcessor<TOutput> ProcessInParallel()
+    {
+        return ProcessInParallel(Environment.ProcessorCount);
+    }
+    
+    /// <summary>
+    /// Process items in parallel optimized for I/O-bound operations and return results.
+    /// </summary>
+    public IAsyncEnumerableProcessor<TOutput> ProcessInParallelForIO(int? maxConcurrency = null)
+    {
+        var concurrency = maxConcurrency ?? Math.Max(Environment.ProcessorCount * 10, 100);
+        return new ResultAsyncEnumerableIOBoundParallelProcessor<TInput, TOutput>(
+            _items, _taskSelector, concurrency, _cancellationTokenSource);
+    }
+    
+    /// <summary>
+    /// Process items one at a time and return results in order.
+    /// </summary>
+    public IAsyncEnumerableProcessor<TOutput> ProcessOneAtATime()
+    {
+        return new ResultAsyncEnumerableOneAtATimeProcessor<TInput, TOutput>(
+            _items, _taskSelector, _cancellationTokenSource);
+    }
+
+    /// <summary>
+    /// Process ALL items in parallel without any concurrency limits and return results.
+    /// WARNING: Use with caution - can overwhelm system resources with large async enumerables.
+    /// </summary>
+    public IAsyncEnumerableProcessor<TOutput> ProcessInParallelUnbounded()
+    {
+        return new ResultAsyncEnumerableUnboundedParallelProcessor<TInput, TOutput>(
+            _items, _taskSelector, _cancellationTokenSource);
+    }
+
+    /// <summary>
+    /// Process items using a channel-based approach and return results.
+    /// </summary>
+    public IAsyncEnumerableProcessor<TOutput> ProcessWithChannel(AsyncEnumerableChannelOptions? options = null)
+    {
+        options ??= new AsyncEnumerableChannelOptions();
+        return new ResultAsyncEnumerableChannelBasedProcessor<TInput, TOutput>(
+            _items, _taskSelector, _cancellationTokenSource, options);
+    }
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor/Builders/AsyncEnumerableAsyncProcessorBuilder.cs b/EnumerableAsyncProcessor/Builders/AsyncEnumerableAsyncProcessorBuilder.cs
new file mode 100644
index 0000000..8e7d9eb
--- /dev/null
+++ b/EnumerableAsyncProcessor/Builders/AsyncEnumerableAsyncProcessorBuilder.cs
@@ -0,0 +1,39 @@
+#if NET6_0_OR_GREATER
+namespace EnumerableAsyncProcessor.Builders;
+
+public class AsyncEnumerableAsyncProcessorBuilder<TInput>
+{
+    private readonly IAsyncEnumerable<TInput> _items;
+
+    internal AsyncEnumerableAsyncProcessorBuilder(IAsyncEnumerable<TInput> items)
+    {
+        _items = items;
+    }
+
+    public AsyncEnumerableActionAsyncProcessorBuilder<TInput, TOutput> SelectAsync<TOutput>(
+        Func<TInput, Task<TOutput>> taskSelector)
+    {
+        return SelectAsync(taskSelector, CancellationToken.None);
+    }
+
+    public AsyncEnumerableActionAsyncProcessorBuilder<TInput, TOutput> SelectAsync<TOutput>(
+        Func<TInput, Task<TOutput>> taskSelector, 
+        CancellationToken cancellationToken)
+    {
+        return new AsyncEnumerableActionAsyncProcessorBuilder<TInput, TOutput>(_items, taskSelector, cancellationToken);
+    }
+
+    public AsyncEnumerableActionAsyncProcessorBuilder<TInput> ForEachAsync(
+        Func<TInput, Task> taskSelector)
+    {
+        return ForEachAsync(taskSelector, CancellationToken.None);
+    }
+
+    public AsyncEnumerableActionAsyncProcessorBuilder<TInput> ForEachAsync(
+        Func<TInput, Task> taskSelector, 
+        CancellationToken cancellationToken)
+    {
+        return new AsyncEnumerableActionAsyncProcessorBuilder<TInput>(_items, taskSelector, cancellationToken);
+    }
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor/EnumerableAsyncProcessor.csproj b/EnumerableAsyncProcessor/EnumerableAsyncProcessor.csproj
index 34c6b0b..61eb819 100644
--- a/EnumerableAsyncProcessor/EnumerableAsyncProcessor.csproj
+++ b/EnumerableAsyncProcessor/EnumerableAsyncProcessor.csproj
@@ -23,6 +23,11 @@
 
     <ItemGroup Condition="'$(TargetFrameworkIdentifier)'!='.NETCoreApp'">
         <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="6.0.0" />
+        <PackageReference Include="System.Threading.Channels" Version="6.0.0" />
+    </ItemGroup>
+    
+    <ItemGroup Condition="'$(TargetFrameworkIdentifier)'=='.NETCoreApp' AND '$(TargetFramework)'=='net6.0'">
+        <PackageReference Include="System.Threading.Channels" Version="6.0.0" />
     </ItemGroup>
 
     <ItemGroup>
diff --git a/EnumerableAsyncProcessor/Extensions/AsyncEnumerableExtensions.cs b/EnumerableAsyncProcessor/Extensions/AsyncEnumerableExtensions.cs
new file mode 100644
index 0000000..64aa970
--- /dev/null
+++ b/EnumerableAsyncProcessor/Extensions/AsyncEnumerableExtensions.cs
@@ -0,0 +1,58 @@
+#if NET6_0_OR_GREATER
+using EnumerableAsyncProcessor.Builders;
+
+namespace EnumerableAsyncProcessor.Extensions;
+
+public static class AsyncEnumerableExtensions
+{
+    public static AsyncEnumerableAsyncProcessorBuilder<T> ToAsyncProcessorBuilder<T>(this IAsyncEnumerable<T> items)
+    {
+        return new AsyncEnumerableAsyncProcessorBuilder<T>(items);
+    }
+    
+    public static AsyncEnumerableActionAsyncProcessorBuilder<T, TOutput> SelectAsync<T, TOutput>(
+        this IAsyncEnumerable<T> items, 
+        Func<T, Task<TOutput>> taskSelector, 
+        CancellationToken cancellationToken = default)
+    {
+        return items.ToAsyncProcessorBuilder()
+            .SelectAsync(taskSelector, cancellationToken);
+    }
+    
+    public static AsyncEnumerableActionAsyncProcessorBuilder<T> ForEachAsync<T>(
+        this IAsyncEnumerable<T> items, 
+        Func<T, Task> taskSelector, 
+        CancellationToken cancellationToken = default)
+    {
+        return items.ToAsyncProcessorBuilder()
+            .ForEachAsync(taskSelector, cancellationToken);
+    }
+    /// <summary>
+    /// Process async enumerable items using a channel-based approach with producer-consumer pattern.
+    /// </summary>
+    public static IAsyncEnumerableProcessor ForEachWithChannelAsync<T>(
+        this IAsyncEnumerable<T> items, 
+        Func<T, Task> taskSelector, 
+        AsyncEnumerableChannelOptions? options = null, 
+        CancellationToken cancellationToken = default)
+    {
+        return items.ToAsyncProcessorBuilder()
+            .ForEachAsync(taskSelector, cancellationToken)
+            .ProcessWithChannel(options);
+    }
+    
+    /// <summary>
+    /// Process async enumerable items using a channel-based approach and return results.
+    /// </summary>
+    public static IAsyncEnumerableProcessor<TOutput> SelectWithChannelAsync<T, TOutput>(
+        this IAsyncEnumerable<T> items, 
+        Func<T, Task<TOutput>> taskSelector, 
+        AsyncEnumerableChannelOptions? options = null, 
+        CancellationToken cancellationToken = default)
+    {
+        return items.ToAsyncProcessorBuilder()
+            .SelectAsync(taskSelector, cancellationToken)
+            .ProcessWithChannel(options);
+    }
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor/Interfaces/IAsyncEnumerableProcessor.cs b/EnumerableAsyncProcessor/Interfaces/IAsyncEnumerableProcessor.cs
new file mode 100644
index 0000000..e3ccb84
--- /dev/null
+++ b/EnumerableAsyncProcessor/Interfaces/IAsyncEnumerableProcessor.cs
@@ -0,0 +1,13 @@
+#if NET6_0_OR_GREATER
+namespace EnumerableAsyncProcessor.Extensions;
+
+public interface IAsyncEnumerableProcessor
+{
+    Task ExecuteAsync();
+}
+
+public interface IAsyncEnumerableProcessor<TOutput>
+{
+    IAsyncEnumerable<TOutput> ExecuteAsync();
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableChannelBasedProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableChannelBasedProcessor.cs
new file mode 100644
index 0000000..d4683a1
--- /dev/null
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableChannelBasedProcessor.cs
@@ -0,0 +1,103 @@
+#if NET6_0_OR_GREATER
+using System.Threading.Channels;
+using EnumerableAsyncProcessor.Extensions;
+
+namespace EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable;
+
+/// <summary>
+/// Channel-based processor with configurable buffer size and concurrency for IAsyncEnumerable.
+/// Provides excellent backpressure management.
+/// </summary>
+public class AsyncEnumerableChannelBasedProcessor<TInput> : IAsyncEnumerableProcessor
+{
+    private readonly IAsyncEnumerable<TInput> _items;
+    private readonly Func<TInput, Task> _taskSelector;
+    private readonly CancellationTokenSource _cancellationTokenSource;
+    private readonly AsyncEnumerableChannelOptions _options;
+
+    internal AsyncEnumerableChannelBasedProcessor(
+        IAsyncEnumerable<TInput> items,
+        Func<TInput, Task> taskSelector,
+        CancellationTokenSource cancellationTokenSource,
+        AsyncEnumerableChannelOptions options)
+    {
+        _items = items;
+        _taskSelector = taskSelector;
+        _cancellationTokenSource = cancellationTokenSource;
+        _options = options;
+    }
+
+    public async Task ExecuteAsync()
+    {
+        var cancellationToken = _cancellationTokenSource.Token;
+        
+        // Create channel with specified buffer size
+        var channel = _options.BufferSize.HasValue
+            ? Channel.CreateBounded<TInput>(new BoundedChannelOptions(_options.BufferSize.Value)
+            {
+                FullMode = BoundedChannelFullMode.Wait,
+                SingleReader = false,
+                SingleWriter = true
+            })
+            : Channel.CreateUnbounded<TInput>(new UnboundedChannelOptions
+            {
+                SingleReader = false,
+                SingleWriter = true
+            });
+
+        // Producer task
+        var producerTask = ProduceAsync(channel.Writer, cancellationToken);
+
+        // Consumer tasks
+        var consumerTasks = Enumerable.Range(0, _options.MaxConcurrency)
+            .Select(_ => ConsumeAsync(channel.Reader, cancellationToken))
+            .ToArray();
+
+        // Wait for all tasks
+        await producerTask;
+        await Task.WhenAll(consumerTasks);
+    }
+
+    private async Task ProduceAsync(ChannelWriter<TInput> writer, CancellationToken cancellationToken)
+    {
+        try
+        {
+            await foreach (var item in _items.WithCancellation(cancellationToken))
+            {
+                await writer.WriteAsync(item, cancellationToken);
+            }
+        }
+        catch (OperationCanceledException)
+        {
+            // Expected when cancellation is requested
+        }
+        finally
+        {
+            writer.Complete();
+        }
+    }
+
+    private async Task ConsumeAsync(ChannelReader<TInput> reader, CancellationToken cancellationToken)
+    {
+        if (_options.IsIOBound)
+        {
+            // For I/O-bound tasks, process directly without Task.Run
+            await foreach (var item in reader.ReadAllAsync(cancellationToken))
+            {
+                await _taskSelector(item);
+            }
+        }
+        else
+        {
+            // For CPU-bound tasks, use Task.Run to avoid blocking
+            await Task.Run(async () =>
+            {
+                await foreach (var item in reader.ReadAllAsync(cancellationToken))
+                {
+                    await _taskSelector(item);
+                }
+            }, cancellationToken);
+        }
+    }
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableIOBoundParallelProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableIOBoundParallelProcessor.cs
new file mode 100644
index 0000000..201d57d
--- /dev/null
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableIOBoundParallelProcessor.cs
@@ -0,0 +1,73 @@
+#if NET6_0_OR_GREATER
+using System.Threading.Channels;
+using EnumerableAsyncProcessor.Extensions;
+
+namespace EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable;
+
+/// <summary>
+/// Optimized parallel processor for I/O-bound operations on IAsyncEnumerable.
+/// Uses higher concurrency levels and avoids Task.Run overhead.
+/// </summary>
+public class AsyncEnumerableIOBoundParallelProcessor<TInput> : IAsyncEnumerableProcessor
+{
+    private readonly IAsyncEnumerable<TInput> _items;
+    private readonly Func<TInput, Task> _taskSelector;
+    private readonly int _maxConcurrency;
+    private readonly CancellationTokenSource _cancellationTokenSource;
+
+    internal AsyncEnumerableIOBoundParallelProcessor(
+        IAsyncEnumerable<TInput> items,
+        Func<TInput, Task> taskSelector,
+        int maxConcurrency,
+        CancellationTokenSource cancellationTokenSource)
+    {
+        _items = items;
+        _taskSelector = taskSelector;
+        _maxConcurrency = maxConcurrency;
+        _cancellationTokenSource = cancellationTokenSource;
+    }
+
+    public async Task ExecuteAsync()
+    {
+        var cancellationToken = _cancellationTokenSource.Token;
+        var semaphore = new SemaphoreSlim(_maxConcurrency, _maxConcurrency);
+        var tasks = new List<Task>();
+
+        try
+        {
+            await foreach (var item in _items.WithCancellation(cancellationToken))
+            {
+                await semaphore.WaitAsync(cancellationToken);
+
+                // Start task without Task.Run for I/O-bound operations
+                var task = ProcessItemAsync(item, semaphore, cancellationToken);
+                tasks.Add(task);
+
+                // Clean up completed tasks periodically to avoid memory growth
+                if (tasks.Count > _maxConcurrency * 2)
+                {
+                    tasks.RemoveAll(t => t.IsCompleted);
+                }
+            }
+
+            await Task.WhenAll(tasks);
+        }
+        finally
+        {
+            semaphore.Dispose();
+        }
+    }
+
+    private async Task ProcessItemAsync(TInput item, SemaphoreSlim semaphore, CancellationToken cancellationToken)
+    {
+        try
+        {
+            await _taskSelector(item);
+        }
+        finally
+        {
+            semaphore.Release();
+        }
+    }
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableOneAtATimeProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableOneAtATimeProcessor.cs
new file mode 100644
index 0000000..d973734
--- /dev/null
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableOneAtATimeProcessor.cs
@@ -0,0 +1,35 @@
+#if NET6_0_OR_GREATER
+using EnumerableAsyncProcessor.Extensions;
+
+namespace EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable;
+
+/// <summary>
+/// Sequential processor that processes items one at a time from an IAsyncEnumerable.
+/// </summary>
+public class AsyncEnumerableOneAtATimeProcessor<TInput> : IAsyncEnumerableProcessor
+{
+    private readonly IAsyncEnumerable<TInput> _items;
+    private readonly Func<TInput, Task> _taskSelector;
+    private readonly CancellationTokenSource _cancellationTokenSource;
+
+    internal AsyncEnumerableOneAtATimeProcessor(
+        IAsyncEnumerable<TInput> items,
+        Func<TInput, Task> taskSelector,
+        CancellationTokenSource cancellationTokenSource)
+    {
+        _items = items;
+        _taskSelector = taskSelector;
+        _cancellationTokenSource = cancellationTokenSource;
+    }
+
+    public async Task ExecuteAsync()
+    {
+        var cancellationToken = _cancellationTokenSource.Token;
+
+        await foreach (var item in _items.WithCancellation(cancellationToken))
+        {
+            await _taskSelector(item);
+        }
+    }
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs
new file mode 100644
index 0000000..db80d12
--- /dev/null
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs
@@ -0,0 +1,63 @@
+#if NET6_0_OR_GREATER
+using System.Threading.Channels;
+using EnumerableAsyncProcessor.Extensions;
+
+namespace EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable;
+
+public class AsyncEnumerableParallelProcessor<TInput> : IAsyncEnumerableProcessor
+{
+    private readonly IAsyncEnumerable<TInput> _items;
+    private readonly Func<TInput, Task> _taskSelector;
+    private readonly int _maxConcurrency;
+    private readonly CancellationTokenSource _cancellationTokenSource;
+
+    internal AsyncEnumerableParallelProcessor(
+        IAsyncEnumerable<TInput> items,
+        Func<TInput, Task> taskSelector,
+        int maxConcurrency,
+        CancellationTokenSource cancellationTokenSource)
+    {
+        _items = items;
+        _taskSelector = taskSelector;
+        _maxConcurrency = maxConcurrency;
+        _cancellationTokenSource = cancellationTokenSource;
+    }
+
+    public async Task ExecuteAsync()
+    {
+        var channel = Channel.CreateUnbounded<TInput>();
+        var cancellationToken = _cancellationTokenSource.Token;
+
+        // Producer task - enumerate the async enumerable and write to channel
+        var producerTask = Task.Run(async () =>
+        {
+            try
+            {
+                await foreach (var item in _items.WithCancellation(cancellationToken))
+                {
+                    await channel.Writer.WriteAsync(item, cancellationToken);
+                }
+            }
+            finally
+            {
+                channel.Writer.Complete();
+            }
+        }, cancellationToken);
+
+        // Consumer tasks - process items from the channel in parallel
+        var consumerTasks = Enumerable.Range(0, _maxConcurrency)
+            .Select(_ => Task.Run(async () =>
+            {
+                await foreach (var item in channel.Reader.ReadAllAsync(cancellationToken))
+                {
+                    await _taskSelector(item);
+                }
+            }, cancellationToken))
+            .ToArray();
+
+        // Wait for producer and all consumers to complete
+        await producerTask;
+        await Task.WhenAll(consumerTasks);
+    }
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableUnboundedParallelProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableUnboundedParallelProcessor.cs
new file mode 100644
index 0000000..504b475
--- /dev/null
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableUnboundedParallelProcessor.cs
@@ -0,0 +1,46 @@
+#if NET6_0_OR_GREATER
+using EnumerableAsyncProcessor.Extensions;
+
+namespace EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable;
+
+/// <summary>
+/// A specialized parallel processor that starts ALL tasks immediately without any concurrency limits.
+/// WARNING: Use with caution - this can overwhelm system resources with large async enumerables.
+/// Unlike regular unbounded which materializes the collection first, this streams items but starts
+/// processing each one immediately as it arrives.
+/// </summary>
+public class AsyncEnumerableUnboundedParallelProcessor<TInput> : IAsyncEnumerableProcessor
+{
+    private readonly IAsyncEnumerable<TInput> _items;
+    private readonly Func<TInput, Task> _taskSelector;
+    private readonly CancellationTokenSource _cancellationTokenSource;
+
+    internal AsyncEnumerableUnboundedParallelProcessor(
+        IAsyncEnumerable<TInput> items,
+        Func<TInput, Task> taskSelector,
+        CancellationTokenSource cancellationTokenSource)
+    {
+        _items = items;
+        _taskSelector = taskSelector;
+        _cancellationTokenSource = cancellationTokenSource;
+    }
+
+    public async Task ExecuteAsync()
+    {
+        var cancellationToken = _cancellationTokenSource.Token;
+        var tasks = new List<Task>();
+
+        // Start a task for each item immediately as it arrives
+        // No throttling or concurrency control
+        await foreach (var item in _items.WithCancellation(cancellationToken))
+        {
+            // Start task immediately without waiting
+            var task = _taskSelector(item);
+            tasks.Add(task);
+        }
+
+        // Wait for all tasks to complete
+        await Task.WhenAll(tasks);
+    }
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableChannelBasedProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableChannelBasedProcessor.cs
new file mode 100644
index 0000000..97353bd
--- /dev/null
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableChannelBasedProcessor.cs
@@ -0,0 +1,281 @@
+#if NET6_0_OR_GREATER
+using System.Threading.Channels;
+using EnumerableAsyncProcessor.Extensions;
+
+namespace EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable.ResultProcessors;
+
+/// <summary>
+/// Channel-based processor that returns results with configurable ordering.
+/// </summary>
+public class ResultAsyncEnumerableChannelBasedProcessor<TInput, TOutput> : IAsyncEnumerableProcessor<TOutput>
+{
+    private readonly IAsyncEnumerable<TInput> _items;
+    private readonly Func<TInput, Task<TOutput>> _taskSelector;
+    private readonly CancellationTokenSource _cancellationTokenSource;
+    private readonly AsyncEnumerableChannelOptions _options;
+
+    internal ResultAsyncEnumerableChannelBasedProcessor(
+        IAsyncEnumerable<TInput> items,
+        Func<TInput, Task<TOutput>> taskSelector,
+        CancellationTokenSource cancellationTokenSource,
+        AsyncEnumerableChannelOptions options)
+    {
+        _items = items;
+        _taskSelector = taskSelector;
+        _cancellationTokenSource = cancellationTokenSource;
+        _options = options;
+    }
+
+    public async IAsyncEnumerable<TOutput> ExecuteAsync()
+    {
+        var cancellationToken = _cancellationTokenSource.Token;
+
+        if (_options.PreserveOrder)
+        {
+            await foreach (var result in ExecuteWithOrderPreservationAsync(cancellationToken))
+            {
+                yield return result;
+            }
+        }
+        else
+        {
+            await foreach (var result in ExecuteWithoutOrderPreservationAsync(cancellationToken))
+            {
+                yield return result;
+            }
+        }
+    }
+
+    private async IAsyncEnumerable<TOutput> ExecuteWithoutOrderPreservationAsync(CancellationToken cancellationToken)
+    {
+        var inputChannel = CreateInputChannel();
+        var outputChannel = Channel.CreateUnbounded<TOutput>();
+
+        // Producer task
+        var producerTask = ProduceAsync(inputChannel.Writer, cancellationToken);
+
+        // Consumer tasks
+        var consumerTasks = Enumerable.Range(0, _options.MaxConcurrency)
+            .Select(_ => ConsumeAsync(inputChannel.Reader, outputChannel.Writer, cancellationToken))
+            .ToArray();
+
+        // Complete output when all processing is done
+        var completionTask = Task.Run(async () =>
+        {
+            await producerTask;
+            await Task.WhenAll(consumerTasks);
+            outputChannel.Writer.Complete();
+        }, cancellationToken);
+
+        // Yield results as they complete
+        await foreach (var result in outputChannel.Reader.ReadAllAsync(cancellationToken))
+        {
+            yield return result;
+        }
+
+        await completionTask;
+    }
+
+    private async IAsyncEnumerable<TOutput> ExecuteWithOrderPreservationAsync(CancellationToken cancellationToken)
+    {
+        var outputChannel = Channel.CreateUnbounded<TOutput>();
+        var semaphore = new SemaphoreSlim(_options.MaxConcurrency, _options.MaxConcurrency);
+        var orderingDictionary = new SortedDictionary<int, TaskCompletionSource<TOutput>>();
+        var orderingLock = new SemaphoreSlim(1, 1);
+        var nextYieldIndex = 0;
+        var totalProduced = 0;
+        var producerCompleted = false;
+
+        // Start producer
+        var orderedInputChannel = CreateOrderedInputChannel();
+        var producerTask = Task.Run(async () =>
+        {
+            await ProduceOrderedAsync(orderedInputChannel.Writer, cancellationToken);
+            producerCompleted = true;
+        }, cancellationToken);
+
+        // Start ordered consumer
+        var consumerTasks = new List<Task>();
+        var consumerTask = Task.Run(async () =>
+        {
+            await foreach (var (item, index) in orderedInputChannel.Reader.ReadAllAsync(cancellationToken))
+            {
+                await semaphore.WaitAsync(cancellationToken);
+                totalProduced = Math.Max(totalProduced, index + 1);
+                var task = ProcessOrderedItemAsync(item, index, orderingDictionary, orderingLock, semaphore, cancellationToken);
+                consumerTasks.Add(task);
+            }
+            await Task.WhenAll(consumerTasks);
+        }, cancellationToken);
+
+        // Yield results in order
+        var yieldingTask = Task.Run(async () =>
+        {
+            while (!producerCompleted || nextYieldIndex < totalProduced || orderingDictionary.Count > 0)
+            {
+                await orderingLock.WaitAsync(cancellationToken);
+                TaskCompletionSource<TOutput>? tcs = null;
+                var found = false;
+                
+                if (orderingDictionary.TryGetValue(nextYieldIndex, out tcs))
+                {
+                    orderingDictionary.Remove(nextYieldIndex);
+                    nextYieldIndex++;
+                    found = true;
+                }
+                orderingLock.Release();
+                
+                if (found && tcs != null)
+                {
+                    await outputChannel.Writer.WriteAsync(await tcs.Task, cancellationToken);
+                }
+                else if (producerCompleted && consumerTasks.All(t => t.IsCompleted) && orderingDictionary.Count == 0)
+                {
+                    break;
+                }
+                else
+                {
+                    await Task.Delay(10, cancellationToken);
+                }
+            }
+            outputChannel.Writer.Complete();
+        }, cancellationToken);
+        
+        // Yield from output channel
+        await foreach (var result in outputChannel.Reader.ReadAllAsync(cancellationToken))
+        {
+            yield return result;
+        }
+
+        await producerTask;
+        await consumerTask;
+        await yieldingTask;
+    }
+
+    private async Task ProcessOrderedItemAsync(
+        TInput item,
+        int index,
+        SortedDictionary<int, TaskCompletionSource<TOutput>> orderingDictionary,
+        SemaphoreSlim orderingLock,
+        SemaphoreSlim semaphore,
+        CancellationToken cancellationToken)
+    {
+        try
+        {
+            var result = await _taskSelector(item);
+
+            await orderingLock.WaitAsync(cancellationToken);
+            try
+            {
+                var tcs = new TaskCompletionSource<TOutput>();
+                tcs.SetResult(result);
+                orderingDictionary[index] = tcs;
+            }
+            finally
+            {
+                orderingLock.Release();
+            }
+        }
+        finally
+        {
+            semaphore.Release();
+        }
+    }
+
+    private Channel<TInput> CreateInputChannel()
+    {
+        return _options.BufferSize.HasValue
+            ? Channel.CreateBounded<TInput>(new BoundedChannelOptions(_options.BufferSize.Value)
+            {
+                FullMode = BoundedChannelFullMode.Wait,
+                SingleReader = false,
+                SingleWriter = true
+            })
+            : Channel.CreateUnbounded<TInput>(new UnboundedChannelOptions
+            {
+                SingleReader = false,
+                SingleWriter = true
+            });
+    }
+
+    private Channel<(TInput, int)> CreateOrderedInputChannel()
+    {
+        return _options.BufferSize.HasValue
+            ? Channel.CreateBounded<(TInput, int)>(new BoundedChannelOptions(_options.BufferSize.Value)
+            {
+                FullMode = BoundedChannelFullMode.Wait,
+                SingleReader = false,
+                SingleWriter = true
+            })
+            : Channel.CreateUnbounded<(TInput, int)>(new UnboundedChannelOptions
+            {
+                SingleReader = false,
+                SingleWriter = true
+            });
+    }
+
+    private async Task ProduceAsync(ChannelWriter<TInput> writer, CancellationToken cancellationToken)
+    {
+        try
+        {
+            await foreach (var item in _items.WithCancellation(cancellationToken))
+            {
+                await writer.WriteAsync(item, cancellationToken);
+            }
+        }
+        finally
+        {
+            writer.Complete();
+        }
+    }
+
+    private async Task ProduceOrderedAsync(ChannelWriter<(TInput, int)> writer, CancellationToken cancellationToken)
+    {
+        try
+        {
+            var index = 0;
+            await foreach (var item in _items.WithCancellation(cancellationToken))
+            {
+                await writer.WriteAsync((item, index++), cancellationToken);
+            }
+        }
+        finally
+        {
+            writer.Complete();
+        }
+    }
+
+    private async Task ConsumeAsync(
+        ChannelReader<TInput> reader,
+        ChannelWriter<TOutput> writer,
+        CancellationToken cancellationToken)
+    {
+        try
+        {
+            if (_options.IsIOBound)
+            {
+                await foreach (var item in reader.ReadAllAsync(cancellationToken))
+                {
+                    var result = await _taskSelector(item);
+                    await writer.WriteAsync(result, cancellationToken);
+                }
+            }
+            else
+            {
+                await Task.Run(async () =>
+                {
+                    await foreach (var item in reader.ReadAllAsync(cancellationToken))
+                    {
+                        var result = await _taskSelector(item);
+                        await writer.WriteAsync(result, cancellationToken);
+                    }
+                }, cancellationToken);
+            }
+        }
+        catch (OperationCanceledException)
+        {
+            // Expected on cancellation
+        }
+    }
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableIOBoundParallelProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableIOBoundParallelProcessor.cs
new file mode 100644
index 0000000..e5467c0
--- /dev/null
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableIOBoundParallelProcessor.cs
@@ -0,0 +1,98 @@
+#if NET6_0_OR_GREATER
+using System.Threading.Channels;
+using EnumerableAsyncProcessor.Extensions;
+
+namespace EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable.ResultProcessors;
+
+/// <summary>
+/// Optimized parallel processor for I/O-bound operations that returns results.
+/// </summary>
+public class ResultAsyncEnumerableIOBoundParallelProcessor<TInput, TOutput> : IAsyncEnumerableProcessor<TOutput>
+{
+    private readonly IAsyncEnumerable<TInput> _items;
+    private readonly Func<TInput, Task<TOutput>> _taskSelector;
+    private readonly int _maxConcurrency;
+    private readonly CancellationTokenSource _cancellationTokenSource;
+
+    internal ResultAsyncEnumerableIOBoundParallelProcessor(
+        IAsyncEnumerable<TInput> items,
+        Func<TInput, Task<TOutput>> taskSelector,
+        int maxConcurrency,
+        CancellationTokenSource cancellationTokenSource)
+    {
+        _items = items;
+        _taskSelector = taskSelector;
+        _maxConcurrency = maxConcurrency;
+        _cancellationTokenSource = cancellationTokenSource;
+    }
+
+    public async IAsyncEnumerable<TOutput> ExecuteAsync()
+    {
+        var cancellationToken = _cancellationTokenSource.Token;
+        var outputChannel = Channel.CreateUnbounded<TOutput>();
+
+        // Start processing in background
+        var processingTask = ProcessAsync(outputChannel.Writer, cancellationToken);
+
+        // Yield results as they complete
+        await foreach (var result in outputChannel.Reader.ReadAllAsync(cancellationToken))
+        {
+            yield return result;
+        }
+
+        await processingTask;
+    }
+
+    private async Task ProcessAsync(ChannelWriter<TOutput> writer, CancellationToken cancellationToken)
+    {
+        var semaphore = new SemaphoreSlim(_maxConcurrency, _maxConcurrency);
+        var tasks = new List<Task>();
+
+        try
+        {
+            await foreach (var item in _items.WithCancellation(cancellationToken))
+            {
+                await semaphore.WaitAsync(cancellationToken);
+
+                // For I/O-bound, don't use Task.Run
+                var task = ProcessItemAsync(item, writer, semaphore, cancellationToken);
+                tasks.Add(task);
+
+                // Periodic cleanup of completed tasks
+                if (tasks.Count > _maxConcurrency * 2)
+                {
+                    tasks.RemoveAll(t => t.IsCompleted);
+                }
+            }
+
+            await Task.WhenAll(tasks);
+        }
+        finally
+        {
+            writer.Complete();
+            semaphore.Dispose();
+        }
+    }
+
+    private async Task ProcessItemAsync(
+        TInput item,
+        ChannelWriter<TOutput> writer,
+        SemaphoreSlim semaphore,
+        CancellationToken cancellationToken)
+    {
+        try
+        {
+            var result = await _taskSelector(item);
+            await writer.WriteAsync(result, cancellationToken);
+        }
+        catch (OperationCanceledException)
+        {
+            // Expected on cancellation
+        }
+        finally
+        {
+            semaphore.Release();
+        }
+    }
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableOneAtATimeProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableOneAtATimeProcessor.cs
new file mode 100644
index 0000000..7f8835e
--- /dev/null
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableOneAtATimeProcessor.cs
@@ -0,0 +1,36 @@
+#if NET6_0_OR_GREATER
+using EnumerableAsyncProcessor.Extensions;
+
+namespace EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable.ResultProcessors;
+
+/// <summary>
+/// Sequential processor that processes items one at a time and returns results in order.
+/// </summary>
+public class ResultAsyncEnumerableOneAtATimeProcessor<TInput, TOutput> : IAsyncEnumerableProcessor<TOutput>
+{
+    private readonly IAsyncEnumerable<TInput> _items;
+    private readonly Func<TInput, Task<TOutput>> _taskSelector;
+    private readonly CancellationTokenSource _cancellationTokenSource;
+
+    internal ResultAsyncEnumerableOneAtATimeProcessor(
+        IAsyncEnumerable<TInput> items,
+        Func<TInput, Task<TOutput>> taskSelector,
+        CancellationTokenSource cancellationTokenSource)
+    {
+        _items = items;
+        _taskSelector = taskSelector;
+        _cancellationTokenSource = cancellationTokenSource;
+    }
+
+    public async IAsyncEnumerable<TOutput> ExecuteAsync()
+    {
+        var cancellationToken = _cancellationTokenSource.Token;
+
+        await foreach (var item in _items.WithCancellation(cancellationToken))
+        {
+            var result = await _taskSelector(item);
+            yield return result;
+        }
+    }
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs
new file mode 100644
index 0000000..392c98f
--- /dev/null
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs
@@ -0,0 +1,91 @@
+#if NET6_0_OR_GREATER
+using System.Threading.Channels;
+using EnumerableAsyncProcessor.Extensions;
+
+namespace EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable.ResultProcessors;
+
+public class ResultAsyncEnumerableParallelProcessor<TInput, TOutput> : IAsyncEnumerableProcessor<TOutput>
+{
+    private readonly IAsyncEnumerable<TInput> _items;
+    private readonly Func<TInput, Task<TOutput>> _taskSelector;
+    private readonly int _maxConcurrency;
+    private readonly CancellationTokenSource _cancellationTokenSource;
+
+    internal ResultAsyncEnumerableParallelProcessor(
+        IAsyncEnumerable<TInput> items,
+        Func<TInput, Task<TOutput>> taskSelector,
+        int maxConcurrency,
+        CancellationTokenSource cancellationTokenSource)
+    {
+        _items = items;
+        _taskSelector = taskSelector;
+        _maxConcurrency = maxConcurrency;
+        _cancellationTokenSource = cancellationTokenSource;
+    }
+
+    public async IAsyncEnumerable<TOutput> ExecuteAsync()
+    {
+        var cancellationToken = _cancellationTokenSource.Token;
+        var outputChannel = Channel.CreateUnbounded<TOutput>();
+
+        // Start the processing task
+        var processingTask = ProcessAsync(outputChannel.Writer, cancellationToken);
+
+        // Yield results as they become available
+        await foreach (var result in outputChannel.Reader.ReadAllAsync(cancellationToken))
+        {
+            yield return result;
+        }
+
+        // Ensure processing completes
+        await processingTask;
+    }
+
+    private async Task ProcessAsync(ChannelWriter<TOutput> writer, CancellationToken cancellationToken)
+    {
+        var semaphore = new SemaphoreSlim(_maxConcurrency, _maxConcurrency);
+        var tasks = new List<Task>();
+
+        try
+        {
+            await foreach (var item in _items.WithCancellation(cancellationToken))
+            {
+                await semaphore.WaitAsync(cancellationToken);
+
+                var task = ProcessItemAsync(item, writer, semaphore, cancellationToken);
+                tasks.Add(task);
+
+                // Clean up completed tasks periodically
+                if (tasks.Count > _maxConcurrency * 2)
+                {
+                    tasks.RemoveAll(t => t.IsCompleted);
+                }
+            }
+
+            await Task.WhenAll(tasks);
+        }
+        finally
+        {
+            writer.Complete();
+            semaphore.Dispose();
+        }
+    }
+
+    private async Task ProcessItemAsync(
+        TInput item, 
+        ChannelWriter<TOutput> writer, 
+        SemaphoreSlim semaphore, 
+        CancellationToken cancellationToken)
+    {
+        try
+        {
+            var result = await _taskSelector(item);
+            await writer.WriteAsync(result, cancellationToken);
+        }
+        finally
+        {
+            semaphore.Release();
+        }
+    }
+}
+#endif
\ No newline at end of file
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableUnboundedParallelProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableUnboundedParallelProcessor.cs
new file mode 100644
index 0000000..af98df8
--- /dev/null
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableUnboundedParallelProcessor.cs
@@ -0,0 +1,84 @@
+#if NET6_0_OR_GREATER
+using System.Threading.Channels;
+using EnumerableAsyncProcessor.Extensions;
+
+namespace EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable.ResultProcessors;
+
+/// <summary>
+/// A specialized parallel processor that starts ALL tasks immediately without any concurrency limits
+/// and returns results as they complete.
+/// WARNING: Use with caution - this can overwhelm system resources with large async enumerables.
+/// </summary>
+public class ResultAsyncEnumerableUnboundedParallelProcessor<TInput, TOutput> : IAsyncEnumerableProcessor<TOutput>
+{
+    private readonly IAsyncEnumerable<TInput> _items;
+    private readonly Func<TInput, Task<TOutput>> _taskSelector;
+    private readonly CancellationTokenSource _cancellationTokenSource;
+
+    internal ResultAsyncEnumerableUnboundedParallelProcessor(
+        IAsyncEnumerable<TInput> items,
+        Func<TInput, Task<TOutput>> taskSelector,
+        CancellationTokenSource cancellationTokenSource)
+    {
+        _items = items;
+        _taskSelector = taskSelector;
+        _cancellationTokenSource = cancellationTokenSource;
+    }
+
+    public async IAsyncEnumerable<TOutput> ExecuteAsync()
+    {
+        var cancellationToken = _cancellationTokenSource.Token;
+        var outputChannel = Channel.CreateUnbounded<TOutput>();
+        
+        // Start processing task
+        var processingTask = ProcessAsync(outputChannel.Writer, cancellationToken);
+
+        // Yield results as they become available
+        await foreach (var result in outputChannel.Reader.ReadAllAsync(cancellationToken))
+        {
+            yield return result;
+        }
+
+        // Ensure processing completes
+        await processingTask;
+    }
+
+    private async Task ProcessAsync(ChannelWriter<TOutput> writer, CancellationToken cancellationToken)
+    {
+        var tasks = new List<Task>();
+
+        try
+        {
+            // Start a task for each item immediately as it arrives
+            await foreach (var item in _items.WithCancellation(cancellationToken))
+            {
+                // Capture the item in a local variable for the closure
+                var capturedItem = item;
+                
+                // Start task immediately and write result when complete
+                var task = Task.Run(async () =>
+                {
+                    try
+                    {
+                        var result = await _taskSelector(capturedItem);
+                        await writer.WriteAsync(result, cancellationToken);
+                    }
+                    catch (OperationCanceledException)
+                    {
+                        // Expected on cancellation
+                    }
+                }, cancellationToken);
+                
+                tasks.Add(task);
+            }
+
+            // Wait for all tasks to complete
+            await Task.WhenAll(tasks);
+        }
+        finally
+        {
+            writer.Complete();
+        }
+    }
+}
+#endif
\ No newline at end of file