apply cancellation tokens and ConfigureAwait

j0nimost · Oct 27, 2023 · c1ed8fc · c1ed8fc
1 parent d0fcbc8
commit c1ed8fc
Showing 1 changed file with 4 additions and 4 deletions.
diff --git a/src/Kafa/Kafa.Stream.cs b/src/Kafa/Kafa.Stream.cs
@@ -98,26 +98,26 @@ public static ReadOnlySpan<byte> Write<T>(List<T> entities, KafaOptions options
             return pooledBufferWriter.WrittenAsSpan;
         }
 
-        public static async ValueTask<Stream> WriteToStreamAsync<T>(List<T> entities, KafaOptions options = null)
+        public static async ValueTask<Stream> WriteToStreamAsync<T>(List<T> entities, KafaOptions options = null, CancellationToken token = default)
         {
             ArgumentNullException.ThrowIfNull(entities, nameof(entities));
             var reflection = SetupOptions<T>(options);
             var memoryStream = new MemoryStream();
             using var bufferWriter = new KafaWriter(memoryStream, reflection.TypeInfo.KafaOptions);
             reflection.GetProperties<T>(bufferWriter, entities);
-            await bufferWriter.FlushAsync();
+            await bufferWriter.FlushAsync(token).ConfigureAwait(false);
             return memoryStream;
         }
 
-        public static async ValueTask WriteToFileAsync<T>(List<T> entities, string path, KafaOptions options = null)
+        public static async ValueTask WriteToFileAsync<T>(List<T> entities, string path, KafaOptions options = null, CancellationToken token=default)
         {
             ArgumentNullException.ThrowIfNull(entities, nameof(entities));
             ArgumentNullException.ThrowIfNull(path, nameof(path));
             var reflection = SetupOptions<T>(options);
             using var fs = new FileStream(path, FileMode.Create);
             using var bufferWriter = new KafaWriter(fs, reflection.TypeInfo.KafaOptions);
             reflection.GetProperties<T>(bufferWriter, entities);
-            await bufferWriter.FlushAsync();
+            await bufferWriter.FlushAsync(token).ConfigureAwait(false);
         }
 
         private static KafaReflection SetupOptions<T>(KafaOptions options)