Add concurrency for parent packages (#2016)

Author: alerickson

src/code/ContainerRegistryServerAPICalls.cs

@@ -63,10 +63,9 @@ public override Task<FindResults> FindVersionAsync(string packageName, string ve
             });
             var filterBuilder = queryBuilder.FilterBuilder;
 
-            // We need to explicitly add 'Id eq <packageName>' whenever $filter is used, otherwise arbitrary results are returned.
-            filterBuilder.AddCriterion($"Id eq '{packageName}'");
-            filterBuilder.AddCriterion($"NormalizedVersion eq '{packageName}'");
-
+// We need to explicitly add 'Id eq <packageName>' whenever $filter is used, otherwise arbitrary results are returned.
+filterBuilder.AddCriterion($"Id eq '{packageName}'");
+filterBuilder.AddCriterion($"NormalizedVersion eq '{version}'");
             var requestUrl = $"{Repository.Uri}/FindPackagesById()?{queryBuilder.BuildQueryString()}";
             string response = HttpRequestCallAsync(requestUrl, debugMsgs, out ErrorRecord errRecord);
             FindResults findResponse = new FindResults(stringResponse: new string[] { response }, hashtableResponse: emptyHashResponses, responseType: FindResponseType);
@@ -87,13 +86,46 @@ public override Task<FindResults> FindVersionAsync(string packageName, string ve
         public override Task<FindResults> FindVersionGlobbingAsync(string packageName, VersionRange versionRange, bool includePrerelease, ResourceType type, bool getOnlyLatest, ConcurrentQueue<ErrorRecord> errorMsgs, ConcurrentQueue<string> warningMsgs, ConcurrentQueue<string> debugMsgs, ConcurrentQueue<string> verboseMsgs)
         {
             debugMsgs.Enqueue("In NuGetServerAPICalls::FindVersionGlobbingAsync()");
-            FindResults findResponse = FindVersionGlobbing(packageName, versionRange, includePrerelease, type, getOnlyLatest, out ErrorRecord errRecord);
+            List<string> responses = new List<string>();
+            int skip = 0;
+
+            var initialResponse = FindVersionGlobbingAsync(packageName, versionRange, includePrerelease, skip, getOnlyLatest, debugMsgs, out ErrorRecord errRecord);
             if (errRecord != null)
             {
                 errorMsgs.Enqueue(errRecord);
+                return Task.FromResult(new FindResults(stringResponse: responses.ToArray(), hashtableResponse: emptyHashResponses, responseType: FindResponseType));
             }
 
-            return Task.FromResult(findResponse);
+            responses.Add(initialResponse);
+
+            if (!getOnlyLatest)
+            {
+                int initialCount = GetCountFromResponse(initialResponse, out errRecord);
+                if (errRecord != null)
+                {
+                    errorMsgs.Enqueue(errRecord);
+                    return Task.FromResult(new FindResults(stringResponse: responses.ToArray(), hashtableResponse: emptyHashResponses, responseType: FindResponseType));
+                }
+
+                int count = (int)Math.Ceiling((double)initialCount / 100) - 1;
+
+                while (count > 0)
+                {
+                    // skip 100
+                    skip += 100;
+                    var tmpResponse = FindVersionGlobbingAsync(packageName, versionRange, includePrerelease, skip, getOnlyLatest, debugMsgs, out errRecord);
+                    if (errRecord != null)
+                    {
+                        errorMsgs.Enqueue(errRecord);
+                        return Task.FromResult(new FindResults(stringResponse: responses.ToArray(), hashtableResponse: emptyHashResponses, responseType: FindResponseType));
+                    }
+
+                    responses.Add(tmpResponse);
+                    count--;
+                }
+            }
+
+            return Task.FromResult(new FindResults(stringResponse: responses.ToArray(), hashtableResponse: emptyHashResponses, responseType: FindResponseType));
         }
         /// <summary>
         /// Find method which allows for searching for all packages from a repository and returns latest version for each.
@@ -122,7 +154,7 @@ public override FindResults FindAll(bool includePrerelease, ResourceType type, o
                 return new FindResults(stringResponse: responses.ToArray(), hashtableResponse: emptyHashResponses, responseType: FindResponseType);
             }
 
-            int count = initialCount / 6000;
+            int count = (int)Math.Ceiling((double)initialCount / 6000) - 1;
             // if more than 100 count, loop and add response to list
             while (count > 0)
             {
@@ -166,7 +198,7 @@ public override FindResults FindTags(string[] tags, bool includePrerelease, Reso
                 return new FindResults(stringResponse: responses.ToArray(), hashtableResponse: emptyHashResponses, responseType: FindResponseType);
             }
 
-            int count = initialCount / 100;
+            int count = (int)Math.Ceiling((double)initialCount / 100) - 1;
             // if more than 100 count, loop and add response to list
             while (count > 0)
             {
@@ -238,7 +270,19 @@ public override FindResults FindName(string packageName, bool includePrerelease,
         public override Task<FindResults> FindNameAsync(string packageName, bool includePrerelease, ResourceType type, ConcurrentQueue<ErrorRecord> errorMsgs, ConcurrentQueue<string> warningMsgs, ConcurrentQueue<string> debugMsgs, ConcurrentQueue<string> verboseMsgs)
         {
             debugMsgs.Enqueue("In NuGetServerAPICalls::FindNameAsync()");
-            FindResults findResponse = FindName(packageName, includePrerelease, type, out ErrorRecord errRecord);
+            var queryBuilder = new NuGetV2QueryBuilder(new Dictionary<string, string>{
+                { "id", $"'{packageName}'" },
+            });
+            var filterBuilder = queryBuilder.FilterBuilder;
+
+            filterBuilder.AddCriterion(includePrerelease ? "IsAbsoluteLatestVersion" : "IsLatestVersion");
+
+            // We need to explicitly add 'Id eq <packageName>' whenever $filter is used, otherwise arbitrary results are returned.
+            filterBuilder.AddCriterion($"Id eq '{packageName}'");
+
+            var requestUrl = $"{Repository.Uri}/FindPackagesById()?{queryBuilder.BuildQueryString()}";
+            string response = HttpRequestCallAsync(requestUrl, debugMsgs, out ErrorRecord errRecord);
+            FindResults findResponse = new FindResults(stringResponse: new string[] { response }, hashtableResponse: emptyHashResponses, responseType: FindResponseType);
             if (errRecord != null)
             {
                 errorMsgs.Enqueue(errRecord);
@@ -310,7 +354,7 @@ public override FindResults FindNameGlobbing(string packageName, bool includePre
                 return new FindResults(stringResponse: responses.ToArray(), hashtableResponse: emptyHashResponses, responseType: FindResponseType);
             }
 
-            int count = initialCount / 100;
+            int count = (int)Math.Ceiling((double)initialCount / 100) - 1;
             // if more than 100 count, loop and add response to list
             while (count > 0)
             {
@@ -355,7 +399,7 @@ public override FindResults FindNameGlobbingWithTag(string packageName, string[]
                 return new FindResults(stringResponse: responses.ToArray(), hashtableResponse: emptyHashResponses, responseType: FindResponseType);
             }
 
-            int count = initialCount / 100;
+            int count = (int)Math.Ceiling((double)initialCount / 100) - 1;
             // if more than 100 count, loop and add response to list
             while (count > 0)
             {
@@ -404,7 +448,7 @@ public override FindResults FindVersionGlobbing(string packageName, VersionRange
                     return new FindResults(stringResponse: responses.ToArray(), hashtableResponse: emptyHashResponses, responseType: FindResponseType);
                 }
 
-                int count = initialCount / 100;
+                int count = (int)Math.Ceiling((double)initialCount / 100) - 1;
 
                 while (count > 0)
                 {
@@ -521,7 +565,19 @@ public override Stream InstallPackage(string packageName, string packageVersion,
         public override Task<Stream> InstallPackageAsync(string packageName, string packageVersion, bool includePrerelease, ConcurrentQueue<ErrorRecord> errorMsgs, ConcurrentQueue<string> warningMsgs, ConcurrentQueue<string> debugMsgs, ConcurrentQueue<string> verboseMsgs)
         {
             debugMsgs.Enqueue("In NuGetServerAPICalls::InstallPackageAsync()");
-            Stream results = InstallPackage(packageName, packageVersion, includePrerelease, out ErrorRecord errRecord);
+            Stream results = new MemoryStream();
+            if (string.IsNullOrEmpty(packageVersion))
+            {
+                errorMsgs.Enqueue(new ErrorRecord(
+                    exception: new ArgumentNullException($"Package version could not be found for {packageName}"),
+                    "PackageVersionNullOrEmptyError",
+                    ErrorCategory.InvalidArgument,
+                    _cmdletPassedIn));
+
+                return Task.FromResult(results);
+            }
+
+            results = InstallVersionAsync(packageName, packageVersion, debugMsgs, out ErrorRecord errRecord);
             if (errRecord != null)
             {
                 errorMsgs.Enqueue(errRecord);
@@ -628,6 +684,55 @@ private HttpContent HttpRequestCallForContent(string requestUrl, out ErrorRecord
             return content;
         }
 
+        /// <summary>
+        /// Helper method that makes the HTTP request for install APIs on worker threads; enqueues diagnostics instead of writing to cmdlet streams.
+        /// </summary>
+        private HttpContent HttpRequestCallForContentAsync(string requestUrl, ConcurrentQueue<string> debugMsgs, out ErrorRecord errRecord)
+        {
+            debugMsgs.Enqueue("In NuGetServerAPICalls::HttpRequestCallForContentAsync()");
+            errRecord = null;
+            HttpContent content = null;
+
+            try
+            {
+                debugMsgs.Enqueue($"Request url is: '{requestUrl}'");
+                HttpRequestMessage request = new HttpRequestMessage(HttpMethod.Get, requestUrl);
+
+                content = SendRequestForContentAsync(request, _sessionClient).GetAwaiter().GetResult();
+            }
+            catch (HttpRequestException e)
+            {
+                errRecord = new ErrorRecord(
+                    exception: e,
+                    "HttpRequestFailure",
+                    ErrorCategory.ConnectionError ,
+                    this);
+            }
+            catch (ArgumentNullException e)
+            {
+                errRecord = new ErrorRecord(
+                    exception: e,
+                    "HttpRequestFailure",
+                    ErrorCategory.InvalidData,
+                    this);
+            }
+            catch (InvalidOperationException e)
+            {
+                errRecord = new ErrorRecord(
+                    exception: e,
+                    "HttpRequestFailure",
+                    ErrorCategory.InvalidOperation,
+                    this);
+            }
+
+            if (string.IsNullOrEmpty(content?.ToString()))
+            {
+                debugMsgs.Enqueue("Response is empty");
+            }
+
+            return content;
+        }
+
         /// <summary>
         /// Helper method that makes the HTTP request for the NuGet server protocol url passed in for async find APIs.
         /// This helper writes diagnostics to the provided debug queue and avoids cmdlet stream writes.
@@ -912,6 +1017,25 @@ private string FindNameGlobbingWithTag(string packageName, string[] tags, bool i
         private string FindVersionGlobbing(string packageName, VersionRange versionRange, bool includePrerelease, int skip, bool getOnlyLatest, out ErrorRecord errRecord)
         {
             _cmdletPassedIn.WriteDebug("In NuGetServerAPICalls::FindVersionGlobbing()");
+            var requestUrl = GetVersionGlobbingRequestUrl(packageName, versionRange, includePrerelease, skip, getOnlyLatest);
+            return HttpRequestCall(requestUrl, out errRecord);
+        }
+
+        /// <summary>
+        /// Worker-thread counterpart of FindVersionGlobbing(); enqueues diagnostics instead of writing to cmdlet streams.
+        /// </summary>
+        private string FindVersionGlobbingAsync(string packageName, VersionRange versionRange, bool includePrerelease, int skip, bool getOnlyLatest, ConcurrentQueue<string> debugMsgs, out ErrorRecord errRecord)
+        {
+            debugMsgs.Enqueue("In NuGetServerAPICalls::FindVersionGlobbingAsync()");
+            var requestUrl = GetVersionGlobbingRequestUrl(packageName, versionRange, includePrerelease, skip, getOnlyLatest);
+            return HttpRequestCallAsync(requestUrl, debugMsgs, out errRecord);
+        }
+
+        /// <summary>
+        /// Builds the FindPackagesById() request url for version-globbing searches.
+        /// </summary>
+        private string GetVersionGlobbingRequestUrl(string packageName, VersionRange versionRange, bool includePrerelease, int skip, bool getOnlyLatest)
+        {
             //https://www.powershellgallery.com/api/v2//FindPackagesById()?id='blah'&includePrerelease=false&$filter= NormalizedVersion gt '1.0.0' and NormalizedVersion lt '2.2.5' and substringof('PSModule', Tags) eq true
             //https://www.powershellgallery.com/api/v2//FindPackagesById()?id='PowerShellGet'&includePrerelease=false&$filter= NormalizedVersion gt '1.1.1' and NormalizedVersion lt '2.2.5'
             // NormalizedVersion doesn't include trailing zeroes
@@ -980,9 +1104,7 @@ private string FindVersionGlobbing(string packageName, VersionRange versionRange
             // We need to explicitly add 'Id eq <packageName>' whenever $filter is used, otherwise arbitrary results are returned.
             filterBuilder.AddCriterion($"Id eq '{packageName}'");
 
-            var requestUrl = $"{Repository.Uri}/FindPackagesById()?{queryBuilder.BuildQueryString()}";
-
-            return HttpRequestCall(requestUrl, out errRecord);
+            return $"{Repository.Uri}/FindPackagesById()?{queryBuilder.BuildQueryString()}";
         }
 
         /// <summary>
@@ -1040,6 +1162,29 @@ private Stream InstallVersion(string packageName, string version, out ErrorRecor
             return response.ReadAsStreamAsync().Result;
         }
 
+        /// <summary>
+        /// Worker-thread counterpart of InstallVersion(); enqueues diagnostics instead of writing to cmdlet streams.
+        /// </summary>
+        private Stream InstallVersionAsync(string packageName, string version, ConcurrentQueue<string> debugMsgs, out ErrorRecord errRecord)
+        {
+            debugMsgs.Enqueue("In NuGetServerAPICalls::InstallVersionAsync()");
+            var requestUrl = $"{Repository.Uri}/Packages(Id='{packageName}',Version='{version}')/Download";
+            var response = HttpRequestCallForContentAsync(requestUrl, debugMsgs, out errRecord);
+
+            if (response is null)
+            {
+                errRecord = new ErrorRecord(
+                    new Exception($"No content was returned by repository '{Repository.Name}'"),
+                    "InstallFailureContentNullNuGetServer",
+                    ErrorCategory.InvalidResult,
+                    this);
+
+                return null;
+            }
+
+            return response.ReadAsStreamAsync().Result;
+        }
+
         /// <summary>
         /// Helper method that makes gets 'count' property from http response string.
         /// The count property is used to determine the number of total results found (for pagination).

src/code/FindHelper.cs

@@ -1737,7 +1737,7 @@ public static void EnqueueIfNotNull<T>(ConcurrentQueue<T> queue, T value)
         }
 
 
-        public static void WriteOutConcurrentQueue(PSCmdlet cmdletPassedIn, ConcurrentQueue<ErrorRecord> errorMsgs, ConcurrentQueue<string> warningMsgs, ConcurrentQueue<string> debugMsgs, ConcurrentQueue<string> verboseMsgs)
+        public static void WriteOutConcurrentQueue(PSCmdlet cmdletPassedIn, ConcurrentQueue<ErrorRecord> errorMsgs, ConcurrentQueue<string> warningMsgs, ConcurrentQueue<string> debugMsgs, ConcurrentQueue<string> verboseMsgs, ConcurrentQueue<InformationRecord> informationMsgs = null)
         {
 
             while (errorMsgs.TryDequeue(out ErrorRecord error))
@@ -1756,6 +1756,10 @@ public static void WriteOutConcurrentQueue(PSCmdlet cmdletPassedIn, ConcurrentQu
             {
                 cmdletPassedIn.WriteVerbose(verboseMsg);
             }
+            while (informationMsgs?.TryDequeue(out InformationRecord informationRecord) == true)
+            {
+                cmdletPassedIn.WriteInformation(informationRecord);
+            }
         }
 
         #endregion

src/code/InstallHelper.cs

@@ -206,7 +206,7 @@ public override FindResults FindTags(string[] tags, bool includePrerelease, Reso
                 if (initialScriptCount != 0)
                 {
                     responses.Add(initialScriptResponse);
-                    int count = initialScriptCount / 100;
+                    int count = (int)Math.Ceiling((double)initialScriptCount / 100) - 1;
                     // if more than 100 count, loop and add response to list
                     while (count > 0)
                     {
@@ -242,7 +242,7 @@ public override FindResults FindTags(string[] tags, bool includePrerelease, Reso
                 if (initialModuleCount != 0)
                 {
                     responses.Add(initialModuleResponse);
-                    int count = initialModuleCount / 100;
+                    int count = (int)Math.Ceiling((double)initialModuleCount / 100) - 1;
                     // if more than 100 count, loop and add response to list
                     while (count > 0)
                     {
@@ -296,7 +296,7 @@ public override FindResults FindCommandOrDscResource(string[] tags, bool include
             if (initialCount != 0)
             {
                 responses.Add(initialResponse);
-                int count = (int)Math.Ceiling((double)(initialCount / 100));
+                int count = (int)Math.Ceiling((double)initialCount / 100) - 1;
 
                 while (count > 0)
                 {
@@ -596,7 +596,7 @@ public override FindResults FindNameGlobbing(string packageName, bool includePre
                 return new FindResults(stringResponse: Utils.EmptyStrArray, hashtableResponse: emptyHashResponses, responseType: v2FindResponseType);
             }
 
-            int count = (int)Math.Ceiling((double)(initialCount / 100));
+            int count = (int)Math.Ceiling((double)initialCount / 100) - 1;
             // if more than 100 count, loop and add response to list
             while (count > 0)
             {
@@ -648,7 +648,7 @@ public override FindResults FindNameGlobbingWithTag(string packageName, string[]
                 return new FindResults(stringResponse: Utils.EmptyStrArray, hashtableResponse: emptyHashResponses, responseType: v2FindResponseType);
             }
 
-            int count = (int)Math.Ceiling((double)(initialCount / 100));
+            int count = (int)Math.Ceiling((double)initialCount / 100) - 1;
             // if more than 100 count, loop and add response to list
             while (count > 0)
             {
@@ -704,7 +704,7 @@ public override FindResults FindVersionGlobbing(string packageName, VersionRange
 
             if (!getOnlyLatest)
             {
-                int count = (int)Math.Ceiling((double)(initialCount / 100));
+                int count = (int)Math.Ceiling((double)initialCount / 100) - 1;
 
                 while (count > 0)
                 {
@@ -1735,7 +1735,7 @@ public override async Task<FindResults> FindVersionGlobbingAsync(string packageN
 
             if (!getOnlyLatest)
             {
-                int count = (int)Math.Ceiling((double)(initialCount / 100));
+                int count = (int)Math.Ceiling((double)initialCount / 100) - 1;
 
                 while (count > 0)
                 {