V3: extract parallelized commit processing into common utility class

src/Catalog/BatchProcessingException.cs

@@ -8,7 +8,7 @@ namespace NuGet.Services.Metadata.Catalog
     public sealed class BatchProcessingException : Exception
     {
         public BatchProcessingException(Exception inner)
-            : base(Strings.BatchProcessingFailure)
+            : base(Strings.BatchProcessingFailure, inner ?? throw new ArgumentNullException(nameof(inner)))
         {
         }
     }

src/Catalog/CatalogCommit.cs

@@ -0,0 +1,41 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+using System;
+using Newtonsoft.Json.Linq;
+
+namespace NuGet.Services.Metadata.Catalog
+{
+    /// <summary>
+    /// Represents a single catalog commit.
+    /// </summary>
+    public sealed class CatalogCommit : IComparable
+    {
+        private CatalogCommit(DateTime commitTimeStamp, Uri uri)
+        {
+            CommitTimeStamp = commitTimeStamp;
+            Uri = uri;
+        }
+
+        public DateTime CommitTimeStamp { get; }
+        public Uri Uri { get; }
+
+        public int CompareTo(object obj)
+        {
+            return CommitTimeStamp.CompareTo(((CatalogCommit)obj).CommitTimeStamp);
+        }
+
+        public static CatalogCommit Create(JObject commit)
+        {
+            if (commit == null)
+            {
+                throw new ArgumentNullException(nameof(commit));
+            }
+
+            var commitTimeStamp = Utils.Deserialize<DateTime>(commit, "commitTimeStamp");
+            var uri = Utils.Deserialize<Uri>(commit, "@id");
+
+            return new CatalogCommit(commitTimeStamp, uri);
+        }
+    }
+}

src/Catalog/CatalogCommitBatchTask.cs

@@ -0,0 +1,54 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+using System;
+using System.Threading.Tasks;
+
+namespace NuGet.Services.Metadata.Catalog
+{
+    /// <summary>
+    /// Represents an asynchrononous task associated with catalog changes for a specific commit item key
+    /// and potentially spanning multiple commits.
+    /// </summary>
+    public sealed class CatalogCommitBatchTask
+    {
+        /// <summary>
+        /// Initializes a <see cref="CatalogCommitItemBatch" /> instance.
+        /// </summary>
+        /// <param name="minCommitTimeStamp">The minimum commit timestamp for commit items with <paramref name="key" />.
+        /// <param name="key">A unique key.</param>
+        /// <exception cref="ArgumentException">Thrown if <paramref name="key" /> is <c>null</c>, empty,
+        /// or whitespace.</exception>
+        public CatalogCommitBatchTask(DateTime minCommitTimeStamp, string key)
+        {
+            if (string.IsNullOrWhiteSpace(key))
+            {
+                throw new ArgumentException(Strings.ArgumentMustNotBeNullEmptyOrWhitespace, nameof(key));
+            }
+
+            MinCommitTimeStamp = minCommitTimeStamp;
+            Key = key;
+        }
+
+        public DateTime MinCommitTimeStamp { get; }
+        public string Key { get; }
+        public Task Task { get; set; }
+
+        public override int GetHashCode()
+        {
+            return Key.GetHashCode();
+        }
+
+        public override bool Equals(object obj)
+        {
+            var other = obj as CatalogCommitBatchTask;
+
+            if (ReferenceEquals(other, null))
+            {
+                return false;
+            }
+
+            return GetHashCode() == other.GetHashCode();
+        }
+    }
+}

src/Catalog/CatalogCommitBatchTasks.cs

@@ -0,0 +1,23 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+using System;
+using System.Collections.Generic;
+
+namespace NuGet.Services.Metadata.Catalog
+{
+    /// <summary>
+    /// Represents a set of <see cref="CatalogCommitBatchTask" /> that share the same minimum commit timestamp.
+    /// </summary>
+    public sealed class CatalogCommitBatchTasks
+    {
+        public CatalogCommitBatchTasks(DateTime commitTimeStamp)
+        {
+            BatchTasks = new HashSet<CatalogCommitBatchTask>();
+            CommitTimeStamp = commitTimeStamp;
+        }
+
+        public HashSet<CatalogCommitBatchTask> BatchTasks { get; }
+        public DateTime CommitTimeStamp { get; }
+    }
+}

src/Catalog/CatalogCommitItem.cs

@@ -0,0 +1,99 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Linq;
+using Newtonsoft.Json.Linq;
+using NuGet.Packaging.Core;
+using NuGet.Versioning;
+
+namespace NuGet.Services.Metadata.Catalog
+{
+    /// <summary>
+    /// Represents a single item in a catalog commit.
+    /// </summary>
+    public sealed class CatalogCommitItem : IComparable
+    {
+        private const string _typeKeyword = "@type";
+
+        private CatalogCommitItem(
+            Uri uri,
+            string commitId,
+            DateTime commitTimeStamp,
+            IReadOnlyList<string> types,
+            IReadOnlyList<Uri> typeUris,
+            PackageIdentity packageIdentity)
+        {
+            Uri = uri;
+            CommitId = commitId;
+            CommitTimeStamp = commitTimeStamp;
+            PackageIdentity = packageIdentity;
+            Types = types;
+            TypeUris = typeUris;
+        }
+
+        public Uri Uri { get; }
+        public DateTime CommitTimeStamp { get; }
+        public string CommitId { get; }
+        public PackageIdentity PackageIdentity { get; }
+        public IReadOnlyList<string> Types { get; }
+        public IReadOnlyList<Uri> TypeUris { get; }
+
+        public int CompareTo(object obj)
+        {
+            return CommitTimeStamp.CompareTo(((CatalogCommitItem)obj).CommitTimeStamp);
+        }
+
+        public static CatalogCommitItem Create(JObject context, JObject commitItem)
+        {
+            if (context == null)
+            {
+                throw new ArgumentNullException(nameof(context));
+            }
+
+            if (commitItem == null)
+            {
+                throw new ArgumentNullException(nameof(commitItem));
+            }
+
+            var commitTimeStamp = Utils.Deserialize<DateTime>(commitItem, "commitTimeStamp");
+            var commitId = Utils.Deserialize<string>(commitItem, "commitId");
+            var idUri = Utils.Deserialize<Uri>(commitItem, "@id");
+            var packageId = Utils.Deserialize<string>(commitItem, "nuget:id");
+            var packageVersion = Utils.Deserialize<string>(commitItem, "nuget:version");
+            var packageIdentity = new PackageIdentity(packageId, new NuGetVersion(packageVersion));
+            var types = GetTypes(commitItem).ToArray();
+
+            if (!types.Any())
+            {
+                throw new ArgumentException(
+                    string.Format(CultureInfo.InvariantCulture, Strings.NonEmptyPropertyValueRequired, _typeKeyword),
+                    nameof(commitItem));
+            }
+
+            var typeUris = types.Select(type => Utils.Expand(context, type)).ToArray();
+
+            return new CatalogCommitItem(idUri, commitId, commitTimeStamp, types, typeUris, packageIdentity);
+        }
+
+        private static IEnumerable<string> GetTypes(JObject commitItem)
+        {
+            if (commitItem.TryGetValue(_typeKeyword, out var value))
+            {
+                if (value is JArray)
+                {
+                    foreach (JToken typeToken in ((JArray)value).Values())
+                    {
+                        yield return typeToken.ToString();
+                    }
+                }
+                else
+                {
+                    yield return value.ToString();
+                }
+            }
+        }
+    }
+}

src/Catalog/CatalogCommitItemBatch.cs

@@ -0,0 +1,47 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+
+namespace NuGet.Services.Metadata.Catalog
+{
+    /// <summary>
+    /// Represents a group of <see cref="CatalogCommitItem" />.
+    /// Items may span multiple commits but are grouped on common criteria (e.g.:  lower-cased package ID).
+    /// </summary>
+    public sealed class CatalogCommitItemBatch
+    {
+        /// <summary>
+        /// Initializes a <see cref="CatalogCommitItemBatch" /> instance.
+        /// </summary>
+        /// <param name="commitTimeStamp">A commit timestamp relevant to <paramref name="items" />.
+        /// For example, the minimum or maximum commit timestamp amongst all <paramref name="items" />,
+        /// depending on the <see cref="CommitCollector" />.</param>
+        /// <param name="key">A unique key for all items in a batch.  This is used for parallelization and may be
+        /// <c>null</c> if parallelization is not used.</param>
+        /// <param name="items">An enumerable of <see cref="CatalogCommitItem" />.  Items may span multiple commits.</param>
+        /// <exception cref="ArgumentException">Thrown if <paramref name="items" /> is either <c>null</c> or empty.</exception>
+        public CatalogCommitItemBatch(DateTime commitTimeStamp, string key, IEnumerable<CatalogCommitItem> items)
+        {
+            if (items == null || !items.Any())
+            {
+                throw new ArgumentException(Strings.ArgumentMustNotBeNullOrEmpty, nameof(items));
+            }
+
+            CommitTimeStamp = commitTimeStamp;
+            Key = key;
+
+            var list = items.ToList();
+
+            list.Sort();
+
+            Items = list;
+        }
+
+        public DateTime CommitTimeStamp { get; }
+        public IReadOnlyList<CatalogCommitItem> Items { get; }
+        public string Key { get; }
+    }
+}