Docs and Enhancing Tests

Author: Cooksauce

README.md

@@ -112,6 +112,21 @@ private async Task<ulong> WriteToDatabaseAsync(PostgreSQLCopyHelper<TestEntity>
 }
 ```
 
+Or asynchronously with asynchronous enumerables:
+
+```csharp
+private async Task<ulong> WriteToDatabaseAsync(PostgreSQLCopyHelper<TestEntity> copyHelper, IAsyncEnumerable<TestEntity> entities, CancellationToken cancellationToken = default)
+{
+    using (var connection = new NpgsqlConnection("Server=127.0.0.1;Port=5432;Database=sampledb;User Id=philipp;Password=test_pwd;"))
+    {
+        await connection.OpenAsync(cancellationToken);
+
+        // Returns count of rows written 
+        return await copyHelper.SaveAllAsync(connection, entities, cancellationToken);
+    }
+}
+```
+
 ## Case-Sensitive Identifiers ##
 
 By default the library does not apply quotes to identifiers, such as Table Names and Column Names. If you want PostgreSQL-conform quoting for identifiers, 

src/PostgreSQLCopyHelper.Test/Issues/Issue58_CancelRequest_Test.cs

@@ -1,9 +1,11 @@
-using System;
-using System.Collections.Generic;
+using System.Collections.Generic;
+using System.Linq;
+using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
 using Npgsql;
 using NUnit.Framework;
+using PostgreSQLCopyHelper.Test.Extensions;
 
 namespace PostgreSQLCopyHelper.Test.Issues
 {
@@ -26,40 +28,64 @@ protected override void OnSetupInTransaction()
         }
 
         [Test]
-        public async Task Test_CanceledBulkInsertThrowsWhenCanceled()
+        public Task Test_CanceledBulkInsertThrowsWhenCanceledBeforeStarting()
         {
             subject = new PostgreSQLCopyHelper<User>("sample", "TestUsers")
                      .MapInteger("Id", x => x.Id)
                      .MapText("Name", x => x.Name);
 
-            var cancellationTokenSource = new CancellationTokenSource();
-
-            // Try to work with the Bulk Inserter:
-            try
-            {
-                cancellationTokenSource.CancelAfter(15);
-                await subject.SaveAllAsync(connection, FetchUserData(), cancellationTokenSource.Token);
-                Assert.Fail("Should Never Reach Here!");
-            }
-            catch (TaskCanceledException)
+            using (var cts = new CancellationTokenSource(1))
             {
+                Assert.ThrowsAsync<TaskCanceledException>(async () =>
+                    await subject.SaveAllAsync(connection, FetchUserData(5), cts.Token), $"Should Throw Exception of Type {nameof(TaskCanceledException)}!");
             }
-            catch (Exception)
+
+            return Task.CompletedTask;
+        }
+
+        [Test]
+        public Task Test_CanceledBulkInsertThrowsWhenCanceled()
+        {
+            subject = new PostgreSQLCopyHelper<User>("sample", "TestUsers")
+                     .MapInteger("Id", x => x.Id)
+                     .MapText("Name", x => x.Name);
+
+            using (var cts = new CancellationTokenSource(15))
             {
-                Assert.Fail("Should Throw Exception of Type TaskCanceledException!");
+                Assert.ThrowsAsync<TaskCanceledException>(async () =>
+                    await subject.SaveAllAsync(connection, FetchUserData(10), cts.Token), $"Should Throw Exception of Type {nameof(TaskCanceledException)}!");
             }
-            finally
+
+            return Task.CompletedTask;
+        }
+
+        [Test]
+        public async Task Test_CanceledBulkInsertDoesNotThrowWhenCancelledAfterCompletion()
+        {
+            subject = new PostgreSQLCopyHelper<User>("sample", "TestUsers")
+                     .MapInteger("Id", x => x.Id)
+                     .MapText("Name", x => x.Name);
+
+            using (var cts = new CancellationTokenSource(50))
             {
-                cancellationTokenSource.Dispose();
+                var recordsSaved = await subject.SaveAllAsync(connection, FetchUserData(10), cts.Token);
+                var result = connection.GetAll("sample", "\"TestUsers\"")
+                    .Cast<User>()
+                    .OrderBy(x => x.Id)
+                    .ToList();
+
+                Assert.AreEqual(2, recordsSaved);
+                Assert.AreEqual(1, result.First().Id);
+                Assert.AreEqual(2, result.Last().Id);
             }
         }
 
-        private static async IAsyncEnumerable<User> FetchUserData()
+        private static async IAsyncEnumerable<User> FetchUserData(int delayMillis, [EnumeratorCancellation] CancellationToken cancellationToken = default)
         {
             for (var i = 1; i <= 2; i++)
             {
                 // Simulate waiting for data to come through.
-                await Task.Delay(10);
+                await Task.Delay(delayMillis, cancellationToken);
                 yield return new User
                 {
                     Id = i,

src/PostgreSQLCopyHelper/Model/ColumnDefinition.cs

@@ -1,6 +1,7 @@
 // Licensed under the MIT license. See LICENSE file in the project root for full license information.
 
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 using Npgsql;
 
@@ -10,11 +11,11 @@ internal class ColumnDefinition<TEntity>
     {
         public string ColumnName { get; set; }
 
-        public Func<NpgsqlBinaryImporter, TEntity, Task> Write { get; set; }
+        public Func<NpgsqlBinaryImporter, TEntity, CancellationToken, Task> WriteAsync { get; set; }
 
         public override string ToString()
         {
-            return $"ColumnDefinition (ColumnName = {ColumnName}, Serialize = {Write})";
+            return $"ColumnDefinition (ColumnName = {ColumnName}, Serialize = {WriteAsync})";
         }
     }
 }

src/PostgreSQLCopyHelper/PostgreSQLCopyHelper.cs

@@ -92,23 +92,23 @@ public PostgreSQLCopyHelper<TEntity> UsePostgresQuoting(bool enabled = true)
 
         public PostgreSQLCopyHelper<TEntity> Map<TProperty>(string columnName, Func<TEntity, TProperty> propertyGetter, NpgsqlDbType type)
         {
-            return AddColumn(columnName, (writer, entity) => writer.WriteAsync(propertyGetter(entity), type));
+            return AddColumn(columnName, (writer, entity, token) => writer.WriteAsync(propertyGetter(entity), type, cancellationToken: token));
         }
 
         public PostgreSQLCopyHelper<TEntity> MapNullable<TProperty>(string columnName, Func<TEntity, TProperty?> propertyGetter, NpgsqlDbType type)
             where TProperty : struct
         {
-            return AddColumn(columnName, async (writer, entity) =>
+            return AddColumn(columnName, async (writer, entity, token) =>
             {
                 var val = propertyGetter(entity);
 
                 if (!val.HasValue)
                 {
-                    await writer.WriteNullAsync();
+                    await writer.WriteNullAsync(cancellationToken: token);
                 }
                 else
                 {
-                    await writer.WriteAsync(val.Value, type);
+                    await writer.WriteAsync(val.Value, type, cancellationToken: token);
                 }
             });
         }
@@ -135,16 +135,16 @@ private async Task WriteToStreamAsync(NpgsqlBinaryImporter writer, TEntity entit
 
             foreach (var columnDefinition in _columns)
             {
-                await columnDefinition.Write(writer, entity);
+                await columnDefinition.WriteAsync(writer, entity, cancellationToken);
             }
         }
 
-        private PostgreSQLCopyHelper<TEntity> AddColumn(string columnName, Func<NpgsqlBinaryImporter, TEntity, Task> action)
+        private PostgreSQLCopyHelper<TEntity> AddColumn(string columnName, Func<NpgsqlBinaryImporter, TEntity, CancellationToken, Task> action)
         {
             _columns.Add(new ColumnDefinition<TEntity>
             {
                 ColumnName = columnName,
-                Write = action
+                WriteAsync = action
             });
 
             return this;