Initial release

.github/workflows/deploy-to-nuget.yml

@@ -21,7 +21,7 @@ jobs:
     - name: Set variables
       id: setvars
       run: |
-        if [[ "${{github.base_ref}}" == "master" || "${{github.ref}}" == "refs/heads/master" ]]; then
+        if [[ "${{github.base_ref}}" == "main" || "${{github.ref}}" == "refs/heads/main" ]]; then
           echo "VERSION_SUFFIX=" >> "$GITHUB_ENV"
         fi
 

src/Root16.Sprout.Dataverse/DataSources/Dataverse/DataverseDataSource.cs

@@ -2,7 +2,9 @@
 using Microsoft.PowerPlatform.Dataverse.Client;
 using Microsoft.Xrm.Sdk;
 using Microsoft.Xrm.Sdk.Messages;
-using Root16.Sprout.Extensions;
+using System.Collections.Concurrent;
+using System.Net;
+using System.ServiceModel;
 
 namespace Root16.Sprout.DataSources.Dataverse;
 
@@ -11,10 +13,18 @@ public class DataverseDataSource : IDataSource<Entity>
     private readonly ILogger<DataverseDataSource> logger;
     public string? ImpersonateUsingAttribute { get; set; }
 
+    const int MaxRetries = 10;
+
     public DataverseDataSource(ServiceClient crmServiceClient, ILogger<DataverseDataSource> logger)
     {
         CrmServiceClient = crmServiceClient;
         ServiceClient.MaxConnectionTimeout = TimeSpan.FromMinutes(11);
+        CrmServiceClient.EnableAffinityCookie = false;
+        // TODO: Is there a better place to do this?
+        ThreadPool.SetMinThreads(100, 100);
+        ServicePointManager.DefaultConnectionLimit = 65000;
+        ServicePointManager.Expect100Continue = false;
+        ServicePointManager.UseNagleAlgorithm = false;
         this.logger = logger;
     }
 
@@ -102,7 +112,7 @@ public async Task<IReadOnlyList<DataOperationResult<Entity>>> ExecuteMultipleAsy
             {
                 if (!dryRun)
                 {
-                    var response = await CrmServiceClient.ExecuteAsync(requestCollection[0]);
+                    var response = await TryExecuteRequestAsync(requestCollection[0]);
                 }
                 results.Add(ResultFromRequestType(requestCollection[0], true));
             }
@@ -123,42 +133,30 @@ public async Task<IReadOnlyList<DataOperationResult<Entity>>> ExecuteMultipleAsy
             }
             else
             {
-                List<OrganizationRequestCollection> ListofRequestCollections = [];
-
-                foreach (var request in requestCollection.ChunkBy(1000))
-                {
-                    var orgRequestCollection = new OrganizationRequestCollection();
-                    orgRequestCollection.AddRange(request);
-                    ListofRequestCollections.Add(orgRequestCollection);
-                }
+                ConcurrentBag<DataOperationResult<Entity>> parallelResults = [];
 
-                List<Task<OrganizationResponse>> requestTasks = [];
+                ParallelOptions parallelOptions = new() { MaxDegreeOfParallelism = CrmServiceClient.RecommendedDegreesOfParallelism };
 
-                foreach (var requests in ListofRequestCollections)
+                await Parallel.ForEachAsync(requestCollection.Chunk(10), parallelOptions, async (batch, token) =>
                 {
-                    requestTasks.Add(CrmServiceClient.ExecuteAsync(new ExecuteMultipleRequest
+                    ExecuteMultipleRequest request = new()
                     {
                         Settings = new ExecuteMultipleSettings
                         {
                             ContinueOnError = true,
                         },
-                        Requests = requests
-                    }));
-                }
+                        Requests = []
+                    };
+                    request.Requests.AddRange(batch);
 
-                OrganizationResponse[] organizationResponses = await Task.WhenAll(requestTasks);
-                List<ExecuteMultipleResponse> executeMultipleResponses = organizationResponses.Select(x => (ExecuteMultipleResponse)x).ToList();
+                    ExecuteMultipleResponse batchResponse = await TryExecuteRequestAsync<ExecuteMultipleResponse>(request, token);
 
-                for (var i = 0; i < executeMultipleResponses.Count; i++)
-                {
-                    var responses = executeMultipleResponses[i].Responses;
-                    var matchingRequests = ListofRequestCollections[i];
-                    for (var k = 0; k < matchingRequests.Count; k++)
+                    for (var k = 0; k < batch.Length; k++)
                     {
-                        var response = responses.FirstOrDefault(r => r.RequestIndex == k);
+                        var response = batchResponse.Responses.FirstOrDefault(r => r.RequestIndex == k);
                         if (response?.Fault is not null)
                         {
-                            results.Add(ResultFromRequestType(matchingRequests[k], false));
+                            parallelResults.Add(ResultFromRequestType(batch[k], false));
                             if (response?.Fault?.InnerFault?.InnerFault?.Message is not null
                                 && response.Fault.InnerFault.InnerFault is OrganizationServiceFault innermostFault)
                             {
@@ -171,10 +169,12 @@ public async Task<IReadOnlyList<DataOperationResult<Entity>>> ExecuteMultipleAsy
                         }
                         else
                         {
-                            results.Add(ResultFromRequestType(matchingRequests[k], true));
+                            parallelResults.Add(ResultFromRequestType(batch[k], true));
                         }
                     }
-                }
+                });
+
+                results.AddRange(parallelResults);
             }
         }
         return results;
@@ -199,6 +199,11 @@ public IPagedQuery<Entity> CreateFetchXmlQuery(string fetchXml)
         return new DataverseFetchXmlPagedQuery(this, fetchXml);
     }
 
+    public IPagedQuery<Entity> CreateFetchXmlReducingQuery(string fetchXml, string? countByAttribute = null)
+    {
+        return new DataverseFetchXmlReducingQuery(this, fetchXml, countByAttribute);
+    }
+
     protected static OrganizationRequest? CreateOrganizationRequest(DataOperation<Entity> change, IEnumerable<string> dataOperationFlags)
     {
         OrganizationRequest request;
@@ -240,4 +245,32 @@ public IPagedQuery<Entity> CreateFetchXmlQuery(string fetchXml)
 
         return request;
     }
+
+    private Task<OrganizationResponse> TryExecuteRequestAsync(OrganizationRequest request, CancellationToken token = default)
+        => TryExecuteRequestAsync<OrganizationResponse>(request, token);
+
+    private async Task<T> TryExecuteRequestAsync<T>(OrganizationRequest request, CancellationToken token = default)
+        where T : OrganizationResponse
+    {
+        var retryCount = 0;
+        Exception? lastException = null;
+        do
+        {
+            try
+            {
+                return (T)await CrmServiceClient.ExecuteAsync(request, token);
+            }
+            catch (FaultException<OrganizationServiceFault>) { throw; }
+            catch (Exception ex)
+            {
+                if (lastException is null || !ex.Message.Equals(lastException.Message, StringComparison.OrdinalIgnoreCase))
+                {
+                    logger.LogError(ex, ex.Message);
+                }
+                lastException = ex;
+            }
+        } while (retryCount++ < MaxRetries);
+
+        throw lastException;
+    }
 }

src/Root16.Sprout.Dataverse/DataSources/Dataverse/DataverseDataSourceFactory.cs

@@ -2,18 +2,12 @@
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Microsoft.PowerPlatform.Dataverse.Client;
-using Root16.Sprout.DataSources.Dataverse;
 
-namespace Root16.Sprout.DependencyInjection;
+namespace Root16.Sprout.DataSources.Dataverse;
 
-public class DataverseDataSourceFactory : IDataverseDataSourceFactory
+public class DataverseDataSourceFactory(IServiceProvider serviceProvider) : IDataverseDataSourceFactory
 {
-    private readonly IServiceProvider serviceProvider;
-
-    public DataverseDataSourceFactory(IServiceProvider serviceProvider)
-    {
-        this.serviceProvider = serviceProvider;
-    }
+    private readonly IServiceProvider serviceProvider = serviceProvider;
 
     public DataverseDataSource CreateDataSource(string connectionStringName)
     {

src/Root16.Sprout.Dataverse/DataSources/Dataverse/DataverseFetchXmlPagedQuery.cs

@@ -6,16 +6,10 @@
 
 namespace Root16.Sprout.DataSources.Dataverse;
 
-public class DataverseFetchXmlPagedQuery : IPagedQuery<Entity>
+public class DataverseFetchXmlPagedQuery(DataverseDataSource dataSource, string fetchXml) : IPagedQuery<Entity>
 {
-    private readonly DataverseDataSource dataSource;
-    private readonly string fetchXml;
-
-    public DataverseFetchXmlPagedQuery(DataverseDataSource dataSource, string fetchXml)
-    {
-        this.dataSource = dataSource;
-        this.fetchXml = fetchXml;
-    }
+    private readonly DataverseDataSource dataSource = dataSource;
+    private readonly string fetchXml = fetchXml;
 
     private static string AddPaging(string fetchXml, int page, int pageSize, string? pagingCookie)
     {
@@ -36,7 +30,7 @@ private static void AddPaging(XDocument fetchDoc, int page, int pageSize, string
 
     public async Task<PagedQueryResult<Entity>> GetNextPageAsync(int pageNumber, int pageSize, object? bookmark)
     {
-        var results = await dataSource.CrmServiceClient.RetrieveMultipleAsync(new FetchExpression(AddPaging(fetchXml, ++pageNumber, pageSize, (string?)bookmark)));
+        var results = await dataSource.CrmServiceClient.RetrieveMultipleWithRetryAsync(new FetchExpression(AddPaging(fetchXml, ++pageNumber, pageSize, (string?)bookmark)));
 
         return new PagedQueryResult<Entity>
         (

src/Root16.Sprout.Dataverse/DataSources/Dataverse/DataverseFetchXmlReducingQuery.cs

@@ -0,0 +1,104 @@
+using Microsoft.PowerPlatform.Dataverse.Client.Extensions;
+using Microsoft.Xrm.Sdk;
+using Microsoft.Xrm.Sdk.Metadata;
+using Microsoft.Xrm.Sdk.Query;
+using System.Xml.Linq;
+
+namespace Root16.Sprout.DataSources.Dataverse;
+
+public class DataverseFetchXmlReducingQuery(DataverseDataSource dataSource, string fetchXml, string? countByAttribute = null) : IPagedQuery<Entity>
+{
+    private readonly DataverseDataSource dataSource = dataSource;
+    private readonly string fetchXml = fetchXml;
+    private readonly string? countByAttribute = countByAttribute;
+    private static string AddPaging(string fetchXml, int? page, int? pageSize, string? pagingCookie)
+    {
+        var fetchDoc = XDocument.Parse(fetchXml);
+        AddPaging(fetchDoc, page, pageSize, pagingCookie);
+        return fetchDoc.ToString(SaveOptions.DisableFormatting);
+    }
+
+    private static void AddPaging(XDocument fetchDoc, int? page, int? pageSize, string? pagingCookie)
+    {
+        if (page is not null)
+            fetchDoc.Root?.SetAttributeValue("page", page);
+        if (pageSize is not null)
+            fetchDoc.Root?.SetAttributeValue("count", pageSize);
+        if (pagingCookie is not null)
+        {
+            fetchDoc.Root?.SetAttributeValue("paging-cookie", pagingCookie);
+        }
+    }
+
+    // results should be 'different' everytime cause reducing
+    public async Task<PagedQueryResult<Entity>> GetNextPageAsync(int pageNumber, int pageSize, object? bookmark)
+    {
+        var results = await dataSource.CrmServiceClient.RetrieveMultipleWithRetryAsync(new FetchExpression(fetchXml)); 
+
+        return new PagedQueryResult<Entity>
+        (
+            [.. results.Entities.Take(pageSize)],
+            results.MoreRecords || results.Entities.Count > pageSize,
+            results.PagingCookie
+        );
+    }
+
+    public Task<int?> GetTotalRecordCountAsync()
+    {
+        var fetchDoc = XDocument.Parse(fetchXml);
+        if (fetchDoc.Root is null)
+        {
+            return Task.FromResult<int?>(null);
+        }
+
+        if ((bool?)fetchDoc.Root?.Attribute("aggregate") == true)
+        {
+            return Task.FromResult<int?>(null);
+        }
+
+        var entityElem = fetchDoc.Root?.Element("entity");
+        if (entityElem is null)
+        {
+            return Task.FromResult<int?>(null);
+        }
+
+        var attributeElements = fetchDoc.Root?.Descendants().Where(e => e.Name == "attribute").ToArray();
+        if (attributeElements is null)
+        {
+            return Task.FromResult<int?>(null);
+        }
+
+        foreach (var attr in attributeElements)
+        {
+            attr.Remove();
+        }
+
+        string? primaryAttribute = this.countByAttribute;
+        if (string.IsNullOrWhiteSpace(primaryAttribute))
+        {
+            var entityMetadata = dataSource.CrmServiceClient.GetEntityMetadata(entityElem.Attribute("name")?.Value, EntityFilters.Entity);
+            primaryAttribute = entityMetadata.PrimaryIdAttribute;
+        }
+        entityElem.Add(new XElement("attribute", new XAttribute("name", primaryAttribute)));
+
+        int page = 1;
+
+        bool moreRecords;
+        string? pagingCookie = null;
+        int pageSize = 5000;
+        int totalCount = 0;
+
+        do
+        {
+            AddPaging(fetchDoc, page, pageSize, pagingCookie);
+            var results = dataSource.CrmServiceClient.RetrieveMultiple(new FetchExpression { Query = fetchDoc.ToString(SaveOptions.DisableFormatting) });
+            totalCount += results.Entities.Count;
+            moreRecords = results.MoreRecords;
+            pagingCookie = results.PagingCookie;
+            page++;
+        }
+        while (moreRecords);
+
+        return Task.FromResult<int?>(totalCount);
+    }
+}