[Storage][DataMovement] Containerize single item transfers to simplify internal code (Azure#49107)

Author: jalauzon-msft

sdk/storage/Azure.Storage.DataMovement.Blobs/tests/BlobContainerServiceToServiceJobTests.cs

@@ -115,7 +115,6 @@ await checkpointer.AddNewJobAsync(
                 sourceMock.Object,
                 destinationMock.Object,
                 ServiceToServiceJobPart.CreateJobPartAsync,
-                ServiceToServiceJobPart.CreateJobPartAsync,
                 new TransferOptions(),
                 checkpointer,
                 TransferErrorMode.StopOnAnyFailure,
@@ -171,7 +170,6 @@ await checkpointer.AddNewJobAsync(
                 sourceMock.Object,
                 destinationMock.Object,
                 ServiceToServiceJobPart.CreateJobPartAsync,
-                ServiceToServiceJobPart.CreateJobPartAsync,
                 new TransferOptions(),
                 checkpointer,
                 TransferErrorMode.StopOnAnyFailure,

sdk/storage/Azure.Storage.DataMovement/src/JobBuilder.cs

@@ -3,7 +3,6 @@
 
 using System.Threading.Tasks;
 using System.Threading;
-using System.IO;
 using System.Buffers;
 using Azure.Core.Pipeline;
 using System;
@@ -51,26 +50,20 @@ internal JobBuilder(
         bool resumeJob,
         CancellationToken cancellationToken)
     {
-        JobBuilder.ValidateTransferOptions(transferOptions);
+        ValidateTransferOptions(transferOptions);
 
         TransferOperation transferOperation = new(id: transferId);
         TransferJobInternal transferJobInternal;
 
-        // Single transfer
+        // For single item transfers, wrap in single item container
         if (sourceResource is StorageResourceItem sourceItem &&
             destinationResource is StorageResourceItem destationItem)
         {
-            transferJobInternal = await BuildSingleTransferJob(
-                sourceItem,
-                destationItem,
-                transferOptions,
-                checkpointer,
-                transferOperation,
-                resumeJob,
-                cancellationToken).ConfigureAwait(false);
+            sourceResource = new SingleItemStorageResourceContainer(sourceItem);
+            destinationResource = new SingleItemStorageResourceContainer(destationItem);
         }
-        // Container transfer
-        else if (sourceResource is StorageResourceContainer sourceContainer &&
+
+        if (sourceResource is StorageResourceContainer sourceContainer &&
             destinationResource is StorageResourceContainer destinationContainer)
         {
             transferJobInternal = await BuildContainerTransferJob(
@@ -91,69 +84,6 @@ internal JobBuilder(
         return (transferOperation, transferJobInternal);
     }
 
-    private async Task<TransferJobInternal> BuildSingleTransferJob(
-        StorageResourceItem sourceResource,
-        StorageResourceItem destinationResource,
-        TransferOptions transferOptions,
-        ITransferCheckpointer checkpointer,
-        TransferOperation transferOperation,
-        bool resumeJob,
-        CancellationToken cancellationToken)
-    {
-        TransferJobInternal.CreateJobPartSingleAsync single;
-        TransferJobInternal.CreateJobPartMultiAsync multi;
-        Func<TransferJobInternal, JobPartPlanHeader, StorageResourceItem, StorageResourceItem, JobPartInternal> rehydrate;
-        if (sourceResource.IsLocalResource() && !destinationResource.IsLocalResource())
-        {
-            single = StreamToUriJobPart.CreateJobPartAsync;
-            multi = StreamToUriJobPart.CreateJobPartAsync;
-            rehydrate = DataMovementExtensions.ToStreamToUriJobPartAsync;
-        }
-        else if (!sourceResource.IsLocalResource() && destinationResource.IsLocalResource())
-        {
-            single = UriToStreamJobPart.CreateJobPartAsync;
-            multi = UriToStreamJobPart.CreateJobPartAsync;
-            rehydrate = DataMovementExtensions.ToUriToStreamJobPartAsync;
-        }
-        else if (!sourceResource.IsLocalResource() && !destinationResource.IsLocalResource())
-        {
-            single = ServiceToServiceJobPart.CreateJobPartAsync;
-            multi = ServiceToServiceJobPart.CreateJobPartAsync;
-            rehydrate = DataMovementExtensions.ToServiceToServiceJobPartAsync;
-        }
-        else
-        {
-            throw Errors.InvalidSourceDestinationParams();
-        }
-
-        TransferJobInternal job = new(
-            transferOperation: transferOperation,
-            sourceResource: sourceResource,
-            destinationResource: destinationResource,
-            single,
-            multi,
-            transferOptions: transferOptions,
-            checkpointer: checkpointer,
-            errorHandling: _errorHandling,
-            arrayPool: _arrayPool,
-            clientDiagnostics: ClientDiagnostics);
-
-        int jobPartCount = await checkpointer.GetCurrentJobPartCountAsync(
-                transferId: transferOperation.Id,
-                cancellationToken: cancellationToken).ConfigureAwait(false);
-        if (resumeJob && jobPartCount > 0)
-        {
-            JobPartPlanHeader part = await checkpointer.GetJobPartAsync(transferOperation.Id, partNumber: 0).ConfigureAwait(false);
-            job.AppendJobPart(
-                rehydrate(
-                    job,
-                    part,
-                    sourceResource,
-                    destinationResource));
-        }
-        return job;
-    }
-
     private async Task<TransferJobInternal> BuildContainerTransferJob(
         StorageResourceContainer sourceResource,
         StorageResourceContainer destinationResource,
@@ -163,25 +93,21 @@ private async Task<TransferJobInternal> BuildContainerTransferJob(
         bool resumeJob,
         CancellationToken cancellationToken)
     {
-        TransferJobInternal.CreateJobPartSingleAsync single;
-        TransferJobInternal.CreateJobPartMultiAsync multi;
+        TransferJobInternal.CreateJobPartAsync createPart;
         Func<TransferJobInternal, JobPartPlanHeader, StorageResourceContainer, StorageResourceContainer, JobPartInternal> rehydrate;
         if (sourceResource.IsLocalResource() && !destinationResource.IsLocalResource())
         {
-            single = StreamToUriJobPart.CreateJobPartAsync;
-            multi = StreamToUriJobPart.CreateJobPartAsync;
+            createPart = StreamToUriJobPart.CreateJobPartAsync;
             rehydrate = DataMovementExtensions.ToStreamToUriJobPartAsync;
         }
         else if (!sourceResource.IsLocalResource() && destinationResource.IsLocalResource())
         {
-            single = UriToStreamJobPart.CreateJobPartAsync;
-            multi = UriToStreamJobPart.CreateJobPartAsync;
+            createPart = UriToStreamJobPart.CreateJobPartAsync;
             rehydrate = DataMovementExtensions.ToUriToStreamJobPartAsync;
         }
         else if (!sourceResource.IsLocalResource() && !destinationResource.IsLocalResource())
         {
-            single = ServiceToServiceJobPart.CreateJobPartAsync;
-            multi = ServiceToServiceJobPart.CreateJobPartAsync;
+            createPart = ServiceToServiceJobPart.CreateJobPartAsync;
             rehydrate = DataMovementExtensions.ToServiceToServiceJobPartAsync;
         }
         else
@@ -193,8 +119,7 @@ private async Task<TransferJobInternal> BuildContainerTransferJob(
             transferOperation: transferOperation,
             sourceResource: sourceResource,
             destinationResource: destinationResource,
-            single,
-            multi,
+            createPart,
             transferOptions: transferOptions,
             checkpointer: checkpointer,
             errorHandling: _errorHandling,

sdk/storage/Azure.Storage.DataMovement/src/ServiceToServiceJobPart.cs

@@ -22,31 +22,6 @@ internal class ServiceToServiceJobPart : JobPartInternal, IAsyncDisposable
         /// </summary>
         private CommitChunkHandler _commitBlockHandler;
 
-        /// <summary>
-        /// Creating job part based on a single transfer job
-        /// </summary>
-        private ServiceToServiceJobPart(TransferJobInternal job, int partNumber)
-            : base(transferOperation: job._transferOperation,
-                  partNumber: partNumber,
-                  sourceResource: job._sourceResource,
-                  destinationResource: job._destinationResource,
-                  transferChunkSize: job._maximumTransferChunkSize,
-                  initialTransferSize: job._initialTransferSize,
-                  errorHandling: job._errorMode,
-                  createMode: job._creationPreference,
-                  checkpointer: job._checkpointer,
-                  progressTracker: job._progressTracker,
-                  arrayPool: job.UploadArrayPool,
-                  jobPartEventHandler: job.GetJobPartStatusEventHandler(),
-                  statusEventHandler: job.TransferStatusEventHandler,
-                  failedEventHandler: job.TransferFailedEventHandler,
-                  skippedEventHandler: job.TransferSkippedEventHandler,
-                  singleTransferEventHandler: job.TransferItemCompletedEventHandler,
-                  clientDiagnostics: job.ClientDiagnostics,
-                  cancellationToken: job._cancellationToken)
-        {
-        }
-
         /// <summary>
         /// Creating transfer job based on a storage resource created from listing.
         /// </summary>
@@ -119,19 +94,6 @@ public async ValueTask DisposeAsync()
             await DisposeHandlersAsync().ConfigureAwait(false);
         }
 
-        /// <summary>
-        /// Called when creating a job part from a single transfer.
-        /// </summary>
-        public static async Task<JobPartInternal> CreateJobPartAsync(
-            TransferJobInternal job,
-            int partNumber)
-        {
-            // Create Job Part file as we're initializing the job part
-            ServiceToServiceJobPart part = new ServiceToServiceJobPart(job, partNumber);
-            await part.AddJobPartToCheckpointerAsync().ConfigureAwait(false);
-            return part;
-        }
-
         /// <summary>
         /// Called when creating a job part from a container transfer.
         /// </summary>

sdk/storage/Azure.Storage.DataMovement/src/Shared/DataMovementExtensions.cs

@@ -22,103 +22,22 @@ internal static StorageResourceItemProperties ToStorageResourceProperties(this F
             };
         }
 
-        public static StreamToUriJobPart ToStreamToUriJobPartAsync(
-            this TransferJobInternal baseJob,
-            JobPartPlanHeader header,
-            StorageResourceItem sourceResource,
-            StorageResourceItem destinationResource)
-        {
-            // Override header values if options were specified by user.
-            long initialTransferSize = baseJob._initialTransferSize ?? header.InitialTransferSize;
-            long transferChunkSize = baseJob._maximumTransferChunkSize ?? header.ChunkSize;
-            StorageResourceCreationMode createPreference =
-                baseJob._creationPreference != StorageResourceCreationMode.Default ?
-                baseJob._creationPreference : header.CreatePreference;
-
-            StreamToUriJobPart jobPart = StreamToUriJobPart.CreateJobPartFromCheckpoint(
-                job: baseJob,
-                partNumber: Convert.ToInt32(header.PartNumber),
-                sourceResource: sourceResource,
-                destinationResource: destinationResource,
-                jobPartStatus: header.JobPartStatus,
-                initialTransferSize: initialTransferSize,
-                transferChunkSize: transferChunkSize,
-                createPreference: createPreference);
-
-            jobPart.VerifyJobPartPlanHeader(header);
-
-            // TODO: When enabling resume chunked upload Add each transfer to the CommitChunkHandler
-            return jobPart;
-        }
-
-        public static ServiceToServiceJobPart ToServiceToServiceJobPartAsync(
-            this TransferJobInternal baseJob,
-            JobPartPlanHeader header,
-            StorageResourceItem sourceResource,
-            StorageResourceItem destinationResource)
-        {
-            // Override header values if options were specified by user.
-            long initialTransferSize = baseJob._initialTransferSize ?? header.InitialTransferSize;
-            long transferChunkSize = baseJob._maximumTransferChunkSize ?? header.ChunkSize;
-            StorageResourceCreationMode createPreference =
-                baseJob._creationPreference != StorageResourceCreationMode.Default ?
-                baseJob._creationPreference : header.CreatePreference;
-
-            ServiceToServiceJobPart jobPart = ServiceToServiceJobPart.CreateJobPartFromCheckpoint(
-                job: baseJob,
-                partNumber: Convert.ToInt32(header.PartNumber),
-                sourceResource: sourceResource,
-                destinationResource: destinationResource,
-                jobPartStatus: header.JobPartStatus,
-                initialTransferSize: initialTransferSize,
-                transferChunkSize: transferChunkSize,
-                createPreference: createPreference);
-
-            jobPart.VerifyJobPartPlanHeader(header);
-
-            // TODO: When enabling resume chunked upload Add each transfer to the CommitChunkHandler
-            return jobPart;
-        }
-
-        public static UriToStreamJobPart ToUriToStreamJobPartAsync(
-            this TransferJobInternal baseJob,
-            JobPartPlanHeader header,
-            StorageResourceItem sourceResource,
-            StorageResourceItem destinationResource)
-        {
-            // Override header values if options were specified by user.
-            long initialTransferSize = baseJob._initialTransferSize ?? header.InitialTransferSize;
-            long transferChunkSize = baseJob._maximumTransferChunkSize ?? header.ChunkSize;
-            StorageResourceCreationMode createPreference =
-                baseJob._creationPreference != StorageResourceCreationMode.Default ?
-                baseJob._creationPreference : header.CreatePreference;
-
-            UriToStreamJobPart jobPart = UriToStreamJobPart.CreateJobPartFromCheckpoint(
-                job: baseJob,
-                partNumber: Convert.ToInt32(header.PartNumber),
-                sourceResource: sourceResource,
-                destinationResource: destinationResource,
-                jobPartStatus: header.JobPartStatus,
-                initialTransferSize: initialTransferSize,
-                transferChunkSize: transferChunkSize,
-                createPreference: createPreference);
-
-            jobPart.VerifyJobPartPlanHeader(header);
-
-            // TODO: When enabling resume chunked upload Add each transfer to the CommitChunkHandler
-            return jobPart;
-        }
-
         public static StreamToUriJobPart ToStreamToUriJobPartAsync(
             this TransferJobInternal baseJob,
             JobPartPlanHeader header,
             StorageResourceContainer sourceResource,
             StorageResourceContainer destinationResource)
         {
+            // If saved path equals the cotnainer Uri, its a single item trasfer.
+            // Set the resource name to the full Uri.
             string childSourcePath = header.SourcePath;
-            string childSourceName = childSourcePath.Substring(sourceResource.Uri.AbsoluteUri.Length + 1);
+            string childSourceName = header.SourcePath == sourceResource.Uri.AbsoluteUri.ToString() ?
+                childSourcePath :
+                childSourcePath.Substring(sourceResource.Uri.AbsoluteUri.Length + 1);
             string childDestinationPath = header.DestinationPath;
-            string childDestinationName = childDestinationPath.Substring(destinationResource.Uri.AbsoluteUri.Length + 1);
+            string childDestinationName = header.DestinationPath == destinationResource.Uri.AbsoluteUri.ToString() ?
+                childDestinationPath :
+                childDestinationPath.Substring(destinationResource.Uri.AbsoluteUri.Length + 1);
             // Override header values if options were specified by user.
             long initialTransferSize = baseJob._initialTransferSize ?? header.InitialTransferSize;
             long transferChunkSize = baseJob._maximumTransferChunkSize ?? header.ChunkSize;
@@ -148,8 +67,16 @@ public static ServiceToServiceJobPart ToServiceToServiceJobPartAsync(
             StorageResourceContainer sourceResource,
             StorageResourceContainer destinationResource)
         {
+            // If saved path equals the cotnainer Uri, its a single item trasfer, so the resource name
+            // does not matter. Just set it to the path.
             string childSourcePath = header.SourcePath;
+            string childSourceName = header.SourcePath == sourceResource.Uri.AbsoluteUri.ToString() ?
+                childSourcePath :
+                childSourcePath.Substring(sourceResource.Uri.AbsoluteUri.Length + 1);
             string childDestinationPath = header.DestinationPath;
+            string childDestinationName = header.DestinationPath == destinationResource.Uri.AbsoluteUri.ToString() ?
+                childDestinationPath :
+                childDestinationPath.Substring(destinationResource.Uri.AbsoluteUri.Length + 1);
             // Override header values if options were specified by user.
             long initialTransferSize = baseJob._initialTransferSize ?? header.InitialTransferSize;
             long transferChunkSize = baseJob._maximumTransferChunkSize ?? header.ChunkSize;
@@ -160,8 +87,8 @@ public static ServiceToServiceJobPart ToServiceToServiceJobPartAsync(
             ServiceToServiceJobPart jobPart = ServiceToServiceJobPart.CreateJobPartFromCheckpoint(
                 job: baseJob,
                 partNumber: Convert.ToInt32(header.PartNumber),
-                sourceResource: sourceResource.GetStorageResourceReference(childSourcePath.Substring(sourceResource.Uri.AbsoluteUri.Length + 1), header.SourceTypeId),
-                destinationResource: destinationResource.GetStorageResourceReference(childDestinationPath.Substring(destinationResource.Uri.AbsoluteUri.Length + 1), header.DestinationTypeId),
+                sourceResource: sourceResource.GetStorageResourceReference(childSourceName, header.SourceTypeId),
+                destinationResource: destinationResource.GetStorageResourceReference(childDestinationName, header.DestinationTypeId),
                 jobPartStatus: header.JobPartStatus,
                 initialTransferSize: initialTransferSize,
                 transferChunkSize: transferChunkSize,
@@ -179,11 +106,16 @@ public static UriToStreamJobPart ToUriToStreamJobPartAsync(
             StorageResourceContainer sourceResource,
             StorageResourceContainer destinationResource)
         {
-            // Apply credentials to the saved transfer job path
+            // If saved path equals the cotnainer Uri, its a single item trasfer, so the resource name
+            // does not matter. Just set it to the path.
             string childSourcePath = header.SourcePath;
-            string childSourceName = childSourcePath.Substring(sourceResource.Uri.AbsoluteUri.Length + 1);
+            string childSourceName = header.SourcePath == sourceResource.Uri.AbsoluteUri.ToString() ?
+                childSourcePath :
+                childSourcePath.Substring(sourceResource.Uri.AbsoluteUri.Length + 1);
             string childDestinationPath = header.DestinationPath;
-            string childDestinationName = childDestinationPath.Substring(destinationResource.Uri.AbsoluteUri.Length + 1);
+            string childDestinationName = header.DestinationPath == destinationResource.Uri.AbsoluteUri.ToString() ?
+                childDestinationPath :
+                childDestinationPath.Substring(destinationResource.Uri.AbsoluteUri.Length + 1);
             // Override header values if options were specified by user.
             long initialTransferSize = baseJob._initialTransferSize ?? header.InitialTransferSize;
             long transferChunkSize = baseJob._maximumTransferChunkSize ?? header.ChunkSize;