feat(copy): introduces CopyGraphOptions with events support

src/OrasProject.Oras/CopyGraphOptions.cs

@@ -0,0 +1,41 @@
+// Copyright The ORAS Authors.
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using OrasProject.Oras.Oci;
+
+namespace OrasProject.Oras;
+
+public struct CopyGraphOptions
+{
+    public event Action<Descriptor> PreCopy;
+
+    public event Action<Descriptor> PostCopy;
+
+    public event Action<Descriptor> CopySkipped;
+
+    internal void OnPreCopy(Descriptor descriptor)
+    {
+        PreCopy?.Invoke(descriptor);
+    }
+
+    internal void OnPostCopy(Descriptor descriptor)
+    {
+        PostCopy?.Invoke(descriptor);
+    }
+
+    internal void OnCopySkipped(Descriptor descriptor)
+    {
+        CopySkipped?.Invoke(descriptor);
+    }
+}

src/OrasProject.Oras/Extensions.cs

@@ -11,16 +11,35 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-using OrasProject.Oras.Oci;
 using System;
 using System.Threading;
 using System.Threading.Tasks;
+using OrasProject.Oras.Oci;
 using static OrasProject.Oras.Content.Extensions;
 
 namespace OrasProject.Oras;
 
 public static class Extensions
 {
+    /// <summary>
+    /// Copy copies a rooted directed acyclic graph (DAG) with the tagged root node
+    /// in the source Target to the destination Target.
+    /// The destination reference will be the same as the source reference if the
+    /// destination reference is left blank.
+    /// Returns the descriptor of the root node on successful copy.
+    /// </summary>
+    /// <param name="src"></param>
+    /// <param name="srcRef"></param>
+    /// <param name="dst"></param>
+    /// <param name="dstRef"></param>
+    /// <param name="cancellationToken"></param>
+    /// <returns></returns>
+    /// <exception cref="Exception"></exception>
+    public static Task<Descriptor> CopyAsync(this ITarget src, string srcRef, ITarget dst, string dstRef,
+        CancellationToken cancellationToken = default)
+    {
+        return src.CopyAsync(srcRef, dst, dstRef, new CopyGraphOptions(), cancellationToken);
+    }
 
     /// <summary>
     /// Copy copies a rooted directed acyclic graph (DAG) with the tagged root node
@@ -33,41 +52,52 @@ public static class Extensions
     /// <param name="srcRef"></param>
     /// <param name="dst"></param>
     /// <param name="dstRef"></param>
+    /// <param name="copyOptions"></param>
     /// <param name="cancellationToken"></param>
     /// <returns></returns>
     /// <exception cref="Exception"></exception>
-    public static async Task<Descriptor> CopyAsync(this ITarget src, string srcRef, ITarget dst, string dstRef, CancellationToken cancellationToken = default)
+    public static async Task<Descriptor> CopyAsync(this ITarget src, string srcRef, ITarget dst, string dstRef, CopyGraphOptions? copyOptions = default, CancellationToken cancellationToken = default)
     {
         if (string.IsNullOrEmpty(dstRef))
         {
             dstRef = srcRef;
         }
         var root = await src.ResolveAsync(srcRef, cancellationToken).ConfigureAwait(false);
-        await src.CopyGraphAsync(dst, root, cancellationToken).ConfigureAwait(false);
+        await src.CopyGraphAsync(dst, root, copyOptions, cancellationToken).ConfigureAwait(false);
         await dst.TagAsync(root, dstRef, cancellationToken).ConfigureAwait(false);
         return root;
     }
 
-    public static async Task CopyGraphAsync(this ITarget src, ITarget dst, Descriptor node, CancellationToken cancellationToken)
+    public static Task CopyGraphAsync(this ITarget src, ITarget dst, Descriptor node, CancellationToken cancellationToken = default)
+    {
+        return src.CopyGraphAsync(dst, node, new CopyGraphOptions(), cancellationToken);
+    }
+
+    public static async Task CopyGraphAsync(this ITarget src, ITarget dst, Descriptor node, CopyGraphOptions? copyOptions = default, CancellationToken cancellationToken = default)
     {
         // check if node exists in target
         if (await dst.ExistsAsync(node, cancellationToken).ConfigureAwait(false))
         {
+            copyOptions?.OnCopySkipped(node);
             return;
         }
 
         // retrieve successors
         var successors = await src.GetSuccessorsAsync(node, cancellationToken).ConfigureAwait(false);
-        // obtain data stream
-        var dataStream = await src.FetchAsync(node, cancellationToken).ConfigureAwait(false);
+
         // check if the node has successors
-        if (successors != null)
+        foreach (var childNode in successors)
         {
-            foreach (var childNode in successors)
-            {
-                await src.CopyGraphAsync(dst, childNode, cancellationToken).ConfigureAwait(false);
-            }
+            await src.CopyGraphAsync(dst, childNode, copyOptions, cancellationToken).ConfigureAwait(false);
         }
+
+        // perform the copy
+        copyOptions?.OnPreCopy(node);
+        var dataStream = await src.FetchAsync(node, cancellationToken).ConfigureAwait(false);
         await dst.PushAsync(node, dataStream, cancellationToken).ConfigureAwait(false);
+
+        // we copied it
+        copyOptions?.OnPostCopy(node);
     }
 }
+

tests/OrasProject.Oras.Tests/CopyTest.cs

@@ -16,6 +16,7 @@
 using System.Text;
 using System.Text.Json;
 using Xunit;
+using Index = OrasProject.Oras.Oci.Index;
 
 namespace OrasProject.Oras.Tests;
 
@@ -142,4 +143,146 @@ public async Task CanCopyBetweenMemoryTargets()
 
         }
     }
+
+    [Fact]
+    public async Task TestCopyGraph_FullCopy()
+    {
+        var src = new MemoryStore();
+        var dst = new MemoryStore();
+
+        // Generate test content
+        var blobs = new List<byte[]>();
+        var descs = new List<Descriptor>();
+
+        void AppendBlob(string mediaType, byte[] blob)
+        {
+            blobs.Add(blob);
+            var desc = new Descriptor
+            {
+                MediaType = mediaType,
+                Digest = Digest.ComputeSHA256(blob),
+                Size = blob.Length
+            };
+            descs.Add(desc);
+        }
+
+        void GenerateManifest(Descriptor config, params Descriptor[] layers)
+        {
+            var manifest = new Manifest
+            {
+                Config = config,
+                Layers = layers.ToList()
+            };
+            var manifestBytes = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(manifest));
+            AppendBlob(MediaType.ImageManifest, manifestBytes);
+        }
+
+        void GenerateIndex(params Descriptor[] manifests)
+        {
+            var index = new Index
+            {
+                Manifests = manifests.ToList()
+            };
+            var indexBytes = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(index));
+            AppendBlob(MediaType.ImageIndex, indexBytes);
+        }
+
+        // Append blobs and generate manifests and indices
+        var getBytes = (string data) => Encoding.UTF8.GetBytes(data);
+        AppendBlob(MediaType.ImageConfig, getBytes("config")); // Blob 0
+        AppendBlob(MediaType.ImageLayer, getBytes("foo"));     // Blob 1
+        AppendBlob(MediaType.ImageLayer, getBytes("bar"));     // Blob 2
+        AppendBlob(MediaType.ImageLayer, getBytes("hello"));   // Blob 3
+        GenerateManifest(descs[0], descs[1], descs[2]);                 // Blob 4
+        GenerateManifest(descs[0], descs[3]);                                      // Blob 5
+        GenerateManifest(descs[0], descs[1], descs[2], descs[3]);       // Blob 6
+        GenerateIndex(descs[4], descs[5]);                           // Blob 7
+        GenerateIndex(descs[6]);                                                   // Blob 8
+        GenerateIndex();                                                           // Blob 9
+        GenerateIndex(descs[7], descs[8], descs[9]);                 // Blob 10
+
+        var root = descs[^1]; // The last descriptor as the root
+
+        // Push blobs to the source memory store
+        for (int i = 0; i < blobs.Count; i++)
+        {
+            await src.PushAsync(descs[i], new MemoryStream(blobs[i]), CancellationToken.None);
+        }
+
+        // Set up tracking storage wrappers for verification
+        var srcTracker = new StorageTracker(src);
+        var dstTracker = new StorageTracker(dst);
+
+        // Perform the copy graph operation
+        var copyOptions = new CopyGraphOptions();
+        await srcTracker.CopyGraphAsync(dstTracker, root, copyOptions, CancellationToken.None);
+
+        // Verify contents in the destination
+        foreach (var (desc, blob) in descs.Zip(blobs, Tuple.Create))
+        {
+            Assert.True(await dst.ExistsAsync(desc, CancellationToken.None), $"Blob {desc.Digest} should exist in destination.");
+            var fetchedContent = await dst.FetchAsync(desc, CancellationToken.None);
+            using var memoryStream = new MemoryStream();
+            await fetchedContent.CopyToAsync(memoryStream);
+            Assert.Equal(blob, memoryStream.ToArray());
+        }
+
+        // Verify API counts
+        // REMARKS: FetchCount should equal to blobs.Count
+        // but since there's no caching implemented, it is not
+        Assert.Equal(18, srcTracker.FetchCount);
+        Assert.Equal(0, srcTracker.PushCount);
+        Assert.Equal(0, srcTracker.ExistsCount);
+        Assert.Equal(0, dstTracker.FetchCount);
+        Assert.Equal(blobs.Count, dstTracker.PushCount);
+
+        // REMARKS: ExistsCount should equal to blobs.Count
+        // but since there's no caching implemented, it is not
+        Assert.Equal(16, dstTracker.ExistsCount);
+    }
+
+    private class StorageTracker : ITarget
+    {
+        private readonly ITarget _storage;
+
+        public int FetchCount { get; private set; }
+        public int PushCount { get; private set; }
+        public int ExistsCount { get; private set; }
+
+        public IList<string> Fetched { get; } = [];
+
+        public StorageTracker(ITarget storage)
+        {
+            _storage = storage;
+        }
+
+        public async Task<bool> ExistsAsync(Descriptor desc, CancellationToken cancellationToken)
+        {
+            ExistsCount++;
+            return await _storage.ExistsAsync(desc, cancellationToken);
+        }
+
+        public async Task<Stream> FetchAsync(Descriptor desc, CancellationToken cancellationToken)
+        {
+            FetchCount++;
+            Fetched.Add(desc.Digest);
+            return await _storage.FetchAsync(desc, cancellationToken);
+        }
+
+        public async Task PushAsync(Descriptor desc, Stream content, CancellationToken cancellationToken)
+        {
+            PushCount++;
+            await _storage.PushAsync(desc, content, cancellationToken);
+        }
+
+        public async Task TagAsync(Descriptor desc, string reference, CancellationToken cancellationToken)
+        {
+            await _storage.TagAsync(desc, reference, cancellationToken);
+        }
+
+        public Task<Descriptor> ResolveAsync(string reference, CancellationToken cancellationToken)
+        {
+            return _storage.ResolveAsync(reference, cancellationToken);
+        }
+    }
 }