diff --git a/Module/PSMinIO/bin/PSMinIO.dll b/Module/PSMinIO/bin/PSMinIO.dll index 657382b..a9768fa 100644 Binary files a/Module/PSMinIO/bin/PSMinIO.dll and b/Module/PSMinIO/bin/PSMinIO.dll differ diff --git a/src/Core/S3/MultipartUploadManager.cs b/src/Core/S3/MultipartUploadManager.cs index 538f504..a2bea21 100644 --- a/src/Core/S3/MultipartUploadManager.cs +++ b/src/Core/S3/MultipartUploadManager.cs @@ -48,11 +48,20 @@ namespace PSMinIO.Core.S3 var totalSize = fileInfo.Length; var totalParts = (int)Math.Ceiling((double)totalSize / effectiveChunkSize); - _progressCollector.QueueVerboseMessage("Starting multipart upload: {0} -> {1}/{2}", + _progressCollector.QueueVerboseMessage("Starting multipart upload: {0} -> {1}/{2}", fileInfo.Name, bucketName, objectName); - _progressCollector.QueueVerboseMessage("File size: {0}, Chunk size: {1}, Total parts: {2}", + _progressCollector.QueueVerboseMessage("File size: {0}, Chunk size: {1}, Total parts: {2}", SizeFormatter.FormatBytes(totalSize), SizeFormatter.FormatBytes(effectiveChunkSize), totalParts); + // Initialize 3-layer progress tracking + // Layer 1: Collection Progress (if multiple files) + _progressCollector.QueueProgressUpdate(1, "Multipart Upload Collection", + $"Processing {fileInfo.Name}", 0); + + // Layer 2: File Progress + _progressCollector.QueueProgressUpdate(2, "File Upload", + $"Uploading {fileInfo.Name} ({SizeFormatter.FormatBytes(totalSize)})", 0, 1); + var uploadId = resumeUploadId; var parts = new ConcurrentDictionary(); var operationStartTime = DateTime.UtcNow; @@ -105,15 +114,16 @@ namespace PSMinIO.Core.S3 // Update progress var currentUploaded = Interlocked.Add(ref uploadedBytes, partSize); - var progress = (double)currentUploaded / totalSize * 100; + var fileProgress = (double)currentUploaded / totalSize * 100; var elapsed = DateTime.UtcNow - operationStartTime; var speed = elapsed.TotalSeconds > 0 ? currentUploaded / elapsed.TotalSeconds : 0; - _progressCollector.QueueProgressUpdate(1, "Multipart Upload", - $"Part {partNum}/{totalParts} - {SizeFormatter.FormatBytes(currentUploaded)}/{SizeFormatter.FormatBytes(totalSize)} at {SizeFormatter.FormatSpeed(speed)}", - (int)progress); + // Update file progress (Layer 2) + _progressCollector.QueueProgressUpdate(2, "File Upload", + $"Part {partNum}/{totalParts} - {SizeFormatter.FormatBytes(currentUploaded)}/{SizeFormatter.FormatBytes(totalSize)} at {SizeFormatter.FormatSpeed(speed)}", + (int)fileProgress, 1); - _progressCollector.QueueVerboseMessage("Completed part {0}/{1} ({2})", + _progressCollector.QueueVerboseMessage("Completed part {0}/{1} ({2})", partNum, totalParts, SizeFormatter.FormatBytes(partSize)); } finally @@ -136,10 +146,12 @@ namespace PSMinIO.Core.S3 var averageSpeed = totalDuration.TotalSeconds > 0 ? totalSize / totalDuration.TotalSeconds : 0; _progressCollector.QueueVerboseMessage("Multipart upload completed successfully"); - _progressCollector.QueueVerboseMessage("Total time: {0}, Average speed: {1}", + _progressCollector.QueueVerboseMessage("Total time: {0}, Average speed: {1}", SizeFormatter.FormatDuration(totalDuration), SizeFormatter.FormatSpeed(averageSpeed)); - _progressCollector.QueueProgressCompletion(1, "Multipart Upload"); + // Complete all progress layers + _progressCollector.QueueProgressCompletion(2, "File Upload", 1); + _progressCollector.QueueProgressCompletion(1, "Multipart Upload Collection"); return new MultipartUploadResult { @@ -229,9 +241,21 @@ namespace PSMinIO.Core.S3 using var fileStream = fileInfo.OpenRead(); fileStream.Seek(offset, SeekOrigin.Begin); - + + // Create progress-tracking stream for chunk upload (Layer 3) + var progressStream = new ProgressTrackingStream(fileStream, size, + (bytesRead, totalBytes) => + { + var chunkProgress = (double)bytesRead / totalBytes * 100; + _progressCollector.QueueProgressUpdate(3, "Uploading Chunk", + $"Part {partNumber}: {SizeFormatter.FormatBytes(bytesRead)}/{SizeFormatter.FormatBytes(totalBytes)}", + (int)chunkProgress, 2); // Parent: File Upload (Layer 2) + }); + + // Read data for MD5 calculation (still need to buffer for MD5) var buffer = new byte[size]; var totalRead = 0; + var originalPosition = fileStream.Position; while (totalRead < size) { var bytesRead = fileStream.Read(buffer, totalRead, (int)(size - totalRead)); @@ -239,20 +263,26 @@ namespace PSMinIO.Core.S3 totalRead += bytesRead; } - using var content = new ByteArrayContent(buffer, 0, totalRead); - content.Headers.ContentLength = totalRead; - // Calculate MD5 for integrity using var md5 = MD5.Create(); var md5Hash = md5.ComputeHash(buffer, 0, totalRead); var md5String = Convert.ToBase64String(md5Hash); + + // Reset stream position for actual upload + fileStream.Seek(originalPosition, SeekOrigin.Begin); + + using var content = new StreamContent(progressStream); + content.Headers.ContentLength = size; content.Headers.Add("Content-MD5", md5String); var response = _httpClient.ExecuteRequest(HttpMethod.Put, $"/{bucketName}/{objectName}", queryParams, content: content); var etag = response.Headers.ETag?.Tag?.Trim('"') ?? ""; - + + // Complete chunk progress (Layer 3) + _progressCollector.QueueProgressCompletion(3, "Uploading Chunk", 2); + return new PartInfo { PartNumber = partNumber, diff --git a/src/Utils/ProgressTrackingStream.cs b/src/Utils/ProgressTrackingStream.cs new file mode 100644 index 0000000..f813a78 --- /dev/null +++ b/src/Utils/ProgressTrackingStream.cs @@ -0,0 +1,82 @@ +using System; +using System.IO; + +namespace PSMinIO.Utils +{ + /// + /// Stream wrapper that tracks read progress and reports it via callback + /// + public class ProgressTrackingStream : Stream + { + private readonly Stream _baseStream; + private readonly long _totalBytes; + private readonly Action _progressCallback; + private long _bytesRead = 0; + private readonly long _startPosition; + + public ProgressTrackingStream(Stream baseStream, long totalBytes, Action progressCallback) + { + _baseStream = baseStream ?? throw new ArgumentNullException(nameof(baseStream)); + _totalBytes = totalBytes; + _progressCallback = progressCallback ?? throw new ArgumentNullException(nameof(progressCallback)); + _startPosition = baseStream.Position; + } + + public override bool CanRead => _baseStream.CanRead; + public override bool CanSeek => _baseStream.CanSeek; + public override bool CanWrite => _baseStream.CanWrite; + public override long Length => _totalBytes; + public override long Position + { + get => _bytesRead; + set => throw new NotSupportedException("Setting position is not supported"); + } + + public override int Read(byte[] buffer, int offset, int count) + { + // Limit read to remaining bytes + var remainingBytes = _totalBytes - _bytesRead; + var bytesToRead = (int)Math.Min(count, remainingBytes); + + if (bytesToRead <= 0) + return 0; + + var bytesRead = _baseStream.Read(buffer, offset, bytesToRead); + _bytesRead += bytesRead; + + // Report progress + _progressCallback(_bytesRead, _totalBytes); + + return bytesRead; + } + + public override void Write(byte[] buffer, int offset, int count) + { + throw new NotSupportedException("Writing is not supported"); + } + + public override void Flush() + { + _baseStream.Flush(); + } + + public override long Seek(long offset, SeekOrigin origin) + { + throw new NotSupportedException("Seeking is not supported"); + } + + public override void SetLength(long value) + { + throw new NotSupportedException("Setting length is not supported"); + } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + // Don't dispose the base stream - let the caller handle it + } + base.Dispose(disposing); + } + } +}