Refactor diagnostics and improve channel processing

- Changed severity levels for diagnostics in `.editorconfig` to `silent` for `IDE0301`, `IDE0305`, `IDE0306`, and `RCS1077`.
- Simplified methods in `ChannelDbExtensions` by adding `Contract.EndContractBlock()` and using lambda expressions for channel writing.
- Refactored `PipeResultsTo` and `PipeResultsToAsync` methods in `Transformer.cs` for better clarity and handling of transformations.
- Added new asynchronous methods in `DataReaderExtensions` for converting `DbDataReader` to `IAsyncEnumerable<object[]>`.
- Expanded test cases in `IDataReaderToChannelObjectArrayExtensionTests.cs` to cover additional scenarios and ensure proper null checks.
- Updated `ToChannelExtensionsContractTests` to include asynchronous tests for null parameter checks.

Author: electricessence

.editorconfig

@@ -186,3 +186,6 @@ dotnet_diagnostic.IDE0305.severity = silent
 
 # IDE0306: Simplify collection initialization
 dotnet_diagnostic.IDE0306.severity = silent
+
+# RCS1077: Optimize LINQ usage.
+dotnet_diagnostic.RCS1077.severity = silent

Source/Channel/ToChannel.Target.cs

@@ -16,6 +16,7 @@ private static async ValueTask<long> ToChannelCore<T>(
 		if (command is null) throw new ArgumentNullException(nameof(command));
 		if (writer is null) throw new ArgumentNullException(nameof(writer));
 		if (command.Connection is null) throw new InvalidOperationException("Command has no connection.");
+		Contract.EndContractBlock();
 
 		if (!command.Connection.State.HasFlag(ConnectionState.Open))
 		{
@@ -101,17 +102,11 @@ public static ValueTask<long> ToChannel(this IDataReader reader,
 		ChannelWriter<object[]> target,
 		bool complete,
 		CancellationToken cancellationToken = default)
-	{
-		if (reader is null) throw new ArgumentNullException(nameof(reader));
-		if (target is null) throw new ArgumentNullException(nameof(target));
-		Contract.EndContractBlock();
-
-		return target.WriteAll(
+		=> target.WriteAll(
 			reader.AsEnumerable(),
 			complete,
 			false,
 			cancellationToken);
-	}
 
 	/// <summary>
 	/// Iterates an <see cref="IDataReader"/> and writes each record as an array to the channel.
@@ -127,17 +122,11 @@ public static ValueTask<long> ToChannel(this IDataReader reader,
 		bool complete,
 		ArrayPool<object>? arrayPool,
 		CancellationToken cancellationToken = default)
-	{
-		if (reader is null) throw new ArgumentNullException(nameof(reader));
-		if (target is null) throw new ArgumentNullException(nameof(target));
-		Contract.EndContractBlock();
-
-		return target.WriteAll(
+		=> target.WriteAll(
 			reader.AsEnumerable(arrayPool),
 			complete,
 			false,
 			cancellationToken);
-	}
 
 	/// <summary>
 	/// Iterates an <see cref="IDataReader"/> through the transform function and writes each record to the channel.
@@ -154,18 +143,11 @@ public static ValueTask<long> ToChannel<T>(this IDataReader reader,
 		bool complete,
 		Func<IDataRecord, T> transform,
 		CancellationToken cancellationToken = default)
-	{
-		if (reader is null) throw new ArgumentNullException(nameof(reader));
-		if (target is null) throw new ArgumentNullException(nameof(target));
-		if (transform is null) throw new ArgumentNullException(nameof(transform));
-		Contract.EndContractBlock();
-
-		return target.WriteAll(
+		=> target.WriteAll(
 			reader.Select(transform, cancellationToken),
 			complete,
 			false,
 			cancellationToken);
-	}
 
 	/// <summary>
 	/// Iterates an <see cref="IDataReader"/> mapping the results to classes of type <typeparamref name="T"/> and writes each record to the channel.
@@ -521,17 +503,11 @@ public static ValueTask<long> ToChannelAsync(this DbDataReader reader,
 		ChannelWriter<object[]> target,
 		bool complete,
 		CancellationToken cancellationToken = default)
-	{
-		if (reader is null) throw new ArgumentNullException(nameof(reader));
-		if (target is null) throw new ArgumentNullException(nameof(target));
-		Contract.EndContractBlock();
-
-		return target.WriteAllAsync(
+		=> target.WriteAllAsync(
 			reader.AsAsyncEnumerable(cancellationToken),
 			complete,
 			false,
 			cancellationToken);
-	}
 
 	/// <summary>
 	/// Asynchronously iterates an DbDataReader and writes each record as an array to the channel.
@@ -546,17 +522,11 @@ public static ValueTask<long> ToChannelAsync(this DbDataReader reader,
 		bool complete,
 		ArrayPool<object>? arrayPool,
 		CancellationToken cancellationToken = default)
-	{
-		if (reader is null) throw new ArgumentNullException(nameof(reader));
-		if (target is null) throw new ArgumentNullException(nameof(target));
-		Contract.EndContractBlock();
-
-		return target.WriteAllAsync(
+		=> target.WriteAllAsync(
 			reader.AsAsyncEnumerable(arrayPool, cancellationToken),
 			complete,
 			false,
 			cancellationToken);
-	}
 
 	/// <summary>
 	/// Asynchronously iterates an DbDataReader through the transform function and writes each record to the channel.

Source/Channel/Transformer.cs

@@ -1,4 +1,6 @@
-namespace Open.Database.Extensions;
+using System.Diagnostics;
+
+namespace Open.Database.Extensions;
 
 /// <summary>Constructs a <see cref="Transformer{T}"/>.</summary>
 /// <param name="fieldMappingOverrides">An optional override map of field names to column names where the keys are the property names, and values are the column names.</param>
@@ -10,24 +12,27 @@ internal class Transformer<T>(IEnumerable<(string Field, string? Column)>? field
 	/// Static utility for creating a Transformer <typeparamref name="T"/>.
 	/// </summary>
 	/// <param name="fieldMappingOverrides">An optional override map of field names to column names where the keys are the property names, and values are the column names.</param>
+	[ExcludeFromCodeCoverage]
 	public static new Transformer<T> Create(IEnumerable<(string Field, string? Column)>? fieldMappingOverrides = null)
 		=> new(fieldMappingOverrides);
 
 	/// <summary>
-	/// Transforms the results from the reader by first buffering the results and if/when the buffer size is reached, the results are transformed to a channel for reading.
+	/// Transforms the results from the reader by first buffering the results and then the final results are transformed to the target channel for reading.
 	/// </summary>
-	/// <param name="reader">The reader to read from.</param>
-	/// <param name="target">The target channel to write to.</param>
-	/// <param name="complete">Will call complete when no more results are avaiable.</param>
-	/// <param name="cancellationToken">The cancellation token.</param>
-	/// <returns>The ChannelReader of the target.</returns>
-	internal async ValueTask<long> PipeResultsTo(IDataReader reader, ChannelWriter<T> target, bool complete, CancellationToken cancellationToken)
+	/// <remarks>
+	/// This is necessary to absorb as much data as possible first and defer the transformation till later.
+	/// </remarks>
+	private ChannelWriter<object[]> PipeResultsToPrep(
+		IDataReader reader,
+		ChannelWriter<T> target,
+		bool complete,
+		CancellationToken cancellationToken)
 	{
 		if (reader is null) throw new ArgumentNullException(nameof(reader));
+		if (target is null) throw new ArgumentNullException(nameof(target));
 		Contract.EndContractBlock();
 
 		(string Name, int Ordinal)[] columns = reader.GetMatchingOrdinals(ColumnNames, true);
-		var ordinals = columns.Select(m => m.Ordinal).ToImmutableArray();
 		var names = columns.Select(m => m.Name).ToImmutableArray();
 
 		var processor = new Processor(this, names);
@@ -36,7 +41,7 @@ internal async ValueTask<long> PipeResultsTo(IDataReader reader, ChannelWriter<T
 		Channel<object[]> channel = ChannelDbExtensions.CreateChannel<object[]>(MaxArrayBuffer, true);
 		ChannelWriter<object[]> writer = channel.Writer;
 
-		ValueTask<long> piped = channel
+		_ = channel
 			.Reader
 			.Transform(a =>
 			{
@@ -49,26 +54,29 @@ internal async ValueTask<long> PipeResultsTo(IDataReader reader, ChannelWriter<T
 					LocalPool.Return(a);
 				}
 			})
-			.PipeTo(target, complete, cancellationToken);
-
-		if (complete)
-		{
-			_ = piped
-				.AsTask()
-				.ContinueWith(t =>
-				{
-					if (t.IsFaulted) target.Complete(t.Exception);
-					else target.Complete();
-				},
-				CancellationToken.None,
-				TaskContinuationOptions.ExecuteSynchronously,
-				TaskScheduler.Current);
-		}
+			.PipeTo(target, complete, cancellationToken)
+			.AsTask();
 
-		return await reader
-			.ToChannel(writer, false, LocalPool, cancellationToken).ConfigureAwait(false);
+		return writer;
 	}
 
+	/// <summary>
+	/// Transforms the results from the reader by first buffering the results and if/when the buffer size is reached, the results are transformed to a channel for reading.
+	/// </summary>
+	/// <param name="reader">The reader to read from.</param>
+	/// <param name="target">The target channel to write to.</param>
+	/// <param name="complete">Will call complete when no more results are avaiable.</param>
+	/// <param name="cancellationToken">The cancellation token.</param>
+	/// <returns>The ChannelReader of the target.</returns>
+	internal ValueTask<long> PipeResultsTo(
+		IDataReader reader,
+		ChannelWriter<T> target,
+		bool complete,
+		CancellationToken cancellationToken)
+		=> reader.ToChannel(
+			PipeResultsToPrep(reader, target, complete, cancellationToken),
+			true, LocalPool, cancellationToken);
+
 #if NETSTANDARD2_0
 #else
 	/// <summary>
@@ -79,53 +87,15 @@ internal async ValueTask<long> PipeResultsTo(IDataReader reader, ChannelWriter<T
 	/// <param name="complete">Will call complete when no more results are avaiable.</param>
 	/// <param name="cancellationToken">The cancellation token.</param>
 	/// <returns>The ChannelReader of the target.</returns>
-	internal ValueTask<long> PipeResultsToAsync(DbDataReader reader, ChannelWriter<T> target, bool complete, CancellationToken cancellationToken)
-	{
-		if (reader is null) throw new ArgumentNullException(nameof(reader));
-		Contract.EndContractBlock();
-
-		(string Name, int Ordinal)[] columns = reader.GetMatchingOrdinals(ColumnNames, true);
-		var ordinals = columns.Select(m => m.Ordinal).ToImmutableArray();
-		var names = columns.Select(m => m.Name).ToImmutableArray();
-
-		var processor = new Processor(this, names);
-		Func<object?[], T> transform = processor.Transform;
-
-		Channel<object[]> channel = ChannelDbExtensions.CreateChannel<object[]>(MaxArrayBuffer, true);
-		ChannelWriter<object[]> writer = channel.Writer;
-
-		ValueTask<long> piped = channel
-			.Reader
-			.Transform(a =>
-			{
-				try
-				{
-					return transform(a);
-				}
-				finally
-				{
-					LocalPool.Return(a);
-				}
-			})
-			.PipeTo(target, complete, cancellationToken);
-
-		if (complete)
-		{
-			_ = piped
-				.AsTask()
-				.ContinueWith(t =>
-				{
-					if (t.IsFaulted) target.Complete(t.Exception);
-					else target.Complete();
-				},
-				CancellationToken.None,
-				TaskContinuationOptions.ExecuteSynchronously,
-				TaskScheduler.Current);
-		}
-
-		return reader
-			.ToChannelAsync(writer, false, LocalPool, cancellationToken);
-	}
+	[ExcludeFromCodeCoverage]
+	internal ValueTask<long> PipeResultsToAsync(
+		DbDataReader reader,
+		ChannelWriter<T> target,
+		bool complete,
+		CancellationToken cancellationToken)
+		=> reader.ToChannelAsync(
+			PipeResultsToPrep(reader, target, complete, cancellationToken),
+			true, LocalPool, cancellationToken);
 #endif
 
 }