read improvements

Author: GuillaumeVadeBe

Extensions/AzureBlob/Cosmos.DataTransfer.AzureBlobStorage/AzureBlobDataSink.cs

@@ -45,8 +45,8 @@ public async Task WriteToTargetAsync(Func<Stream, Task> writeToStream, IConfigur
             var logInterval = TimeSpan.FromMinutes(1);
             long totalBytes = 0;
 
-            var sw = new Stopwatch();
-            sw.Start();
+            var swOpenWrite = new Stopwatch();
+            swOpenWrite.Start();
             await using var blobStream = await blob.OpenWriteAsync(true, new BlockBlobOpenWriteOptions
             {
                 BufferSize = settings.MaxBlockSizeinKB * 1024L,
@@ -62,16 +62,19 @@ public async Task WriteToTargetAsync(Func<Stream, Task> writeToStream, IConfigur
                 })
 
             }, cancellationToken);
+            swOpenWrite.Stop();
 
+            var swWrite = new Stopwatch();
+            swWrite.Start();
             await writeToStream(blobStream);
-            
-            sw.Stop();
+            swWrite.Stop();
 
+            logger.LogInformation("{BlobName}: open write for blob took {TotalTime} seconds.", settings.BlobName, swOpenWrite.Elapsed.Seconds);
             if (totalBytes != 0)
             {
                 var totalMib = (double) totalBytes / 1024 / 1024;
 
-                logger.LogInformation("{BlobName}: transferred {TotalMiB} Mib to Azure Blob in {TotalTime} seconds.", settings.BlobName, totalMib, (double) sw.ElapsedMilliseconds / 1000);
+                logger.LogInformation("{BlobName}: transferred {TotalMiB} Mib to Azure Blob in {TotalTime} seconds.", settings.BlobName, totalMib, swWrite.Elapsed.Seconds);
             }
         }
 

Extensions/AzureTableAPI/Cosmos.DataTransfer.AzureTableAPIExtension/AzureTableAPIDataSourceExtension.cs

@@ -23,23 +23,14 @@ public async IAsyncEnumerable<IDataItem> ReadAsync(IConfiguration config, ILogge
             var serviceClient = new TableServiceClient(settings.ConnectionString);
             var tableClient = serviceClient.GetTableClient(settings.Table);
 
-            //Pageable<TableEntity> queryResultsFilter = tableClient.Query<TableEntity>(filter: $"PartitionKey eq '{partitionKey}'");
-            AsyncPageable<TableEntity> queryResults;
-            if (!string.IsNullOrWhiteSpace(settings.QueryFilter)) {
-                queryResults = tableClient.QueryAsync<TableEntity>(filter: settings.QueryFilter);
-            } else {
-                queryResults = tableClient.QueryAsync<TableEntity>();
-            }
+            var queryResults = !string.IsNullOrWhiteSpace(settings.QueryFilter)
+                ? tableClient.QueryAsync<TableEntity>(filter: settings.QueryFilter, cancellationToken: cancellationToken)
+                : tableClient.QueryAsync<TableEntity>(cancellationToken: cancellationToken);
 
-            var enumerator = queryResults.GetAsyncEnumerator();
-            while (await enumerator.MoveNextAsync())
+            await foreach (var entity in queryResults.WithCancellation(cancellationToken))
             {
-                yield return new AzureTableAPIDataItem(enumerator.Current, settings.PartitionKeyFieldName, settings.RowKeyFieldName);
+                yield return new AzureTableAPIDataItem(entity, settings.PartitionKeyFieldName, settings.RowKeyFieldName);
             }
-            //do
-            //{
-            //    yield return new AzureTableAPIDataItem(enumerator.Current, settings.PartitionKeyFieldName, settings.RowKeyFieldName);
-            //} while (await enumerator.MoveNextAsync());
         }
 
         public IEnumerable<IDataExtensionSettings> GetSettings()