Parallel.ForEach

GuillaumeVadeBe · GuillaumeVadeBe · commit efe21c3da0f3 · 2024-06-05T10:46:56.000+02:00
diff --git a/Extensions/AzureTableAPI/Cosmos.DataTransfer.AzureTableAPIExtension/AzureTableAPIDataSinkExtension.cs b/Extensions/AzureTableAPI/Cosmos.DataTransfer.AzureTableAPIExtension/AzureTableAPIDataSinkExtension.cs
@@ -23,10 +23,10 @@ public async Task WriteAsync(IAsyncEnumerable<IDataItem> dataItems, IConfigurati
 
             await tableClient.CreateIfNotExistsAsync(cancellationToken);
 
-            await foreach (var batch in GetBatches(dataItems, settings).WithCancellation(cancellationToken))
+            await Parallel.ForEachAsync(GetBatches(dataItems, settings), cancellationToken, async (batch, token) =>
             {
-                await InnerWriteAsync(batch, tableClient, logger, cancellationToken);
-            }
+                await InnerWriteAsync(batch, tableClient, logger, token);
+            });
         }
 
         private static async IAsyncEnumerable<List<TableEntity>> GetBatches(IAsyncEnumerable<IDataItem> dataItems, AzureTableAPIDataSinkSettings settings)

Original file line number	Diff line number	Diff line change
`@@ -23,10 +23,10 @@ public async Task WriteAsync(IAsyncEnumerable<IDataItem> dataItems, IConfigurati`
`23`	`23`
`24`	`24`	`await tableClient.CreateIfNotExistsAsync(cancellationToken);`
`25`	`25`
`26`		`- await foreach (var batch in GetBatches(dataItems, settings).WithCancellation(cancellationToken))`
	`26`	`+ await Parallel.ForEachAsync(GetBatches(dataItems, settings), cancellationToken, async (batch, token) =>`
`27`	`27`	`{`
`28`		`- await InnerWriteAsync(batch, tableClient, logger, cancellationToken);`
`29`		`- }`
	`28`	`+ await InnerWriteAsync(batch, tableClient, logger, token);`
	`29`	`+ });`
`30`	`30`	`}`
`31`	`31`
`32`	`32`	`private static async IAsyncEnumerable<List<TableEntity>> GetBatches(IAsyncEnumerable<IDataItem> dataItems, AzureTableAPIDataSinkSettings settings)`