using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;
using Minio;
using Minio.DataModel;
using PSMinIO.Models;
using PSMinIO.Utils;
namespace PSMinIO.Utils
{
///
/// Synchronous wrapper for MinIO client operations
/// Converts async MinIO operations to synchronous calls for PowerShell compatibility
///
public class MinIOClientWrapper : IDisposable
{
private readonly IMinioClient _client;
private readonly CancellationTokenSource _cancellationTokenSource;
private bool _disposed = false;
///
/// Creates a new MinIOClientWrapper instance
///
/// MinIO configuration
public MinIOClientWrapper(MinIOConfiguration configuration)
{
if (configuration == null)
throw new ArgumentNullException(nameof(configuration));
if (!configuration.IsValid)
throw new ArgumentException("Invalid MinIO configuration", nameof(configuration));
_cancellationTokenSource = new CancellationTokenSource();
// Create MinIO client with configuration
var clientBuilder = new MinioClient()
.WithEndpoint(configuration.Endpoint)
.WithCredentials(configuration.AccessKey, configuration.SecretKey);
if (configuration.UseSSL)
{
clientBuilder = clientBuilder.WithSSL();
// Configure custom HttpClient for certificate validation if needed
if (configuration.SkipCertificateValidation)
{
var httpClientHandler = new HttpClientHandler()
{
ServerCertificateCustomValidationCallback = (message, cert, chain, errors) => true
};
var httpClient = new HttpClient(httpClientHandler);
clientBuilder = clientBuilder.WithHttpClient(httpClient);
}
}
if (configuration.TimeoutSeconds > 0)
{
clientBuilder = clientBuilder.WithTimeout(configuration.TimeoutSeconds * 1000);
}
_client = clientBuilder.Build();
}
///
/// Gets the cancellation token for operations
///
public CancellationToken CancellationToken => _cancellationTokenSource.Token;
///
/// Lists all buckets synchronously
///
/// List of bucket information
public List ListBuckets()
{
try
{
var bucketsResult = Task.Run(async () =>
await _client.ListBucketsAsync(CancellationToken)).GetAwaiter().GetResult();
return bucketsResult.Buckets
.Select(MinIOBucketInfo.FromMinioBucket)
.ToList();
}
catch (Exception ex)
{
throw new InvalidOperationException($"Failed to list buckets: {ex.Message}", ex);
}
}
///
/// Checks if a bucket exists synchronously
///
/// Name of the bucket
/// True if bucket exists, false otherwise
public bool BucketExists(string bucketName)
{
if (string.IsNullOrWhiteSpace(bucketName))
throw new ArgumentException("Bucket name cannot be null or empty", nameof(bucketName));
try
{
var args = new BucketExistsArgs().WithBucket(bucketName);
return Task.Run(async () =>
await _client.BucketExistsAsync(args, CancellationToken)).GetAwaiter().GetResult();
}
catch (Exception ex)
{
throw new InvalidOperationException($"Failed to check if bucket '{bucketName}' exists: {ex.Message}", ex);
}
}
///
/// Creates a bucket synchronously
///
/// Name of the bucket to create
/// Optional region for the bucket
public void CreateBucket(string bucketName, string? region = null)
{
if (string.IsNullOrWhiteSpace(bucketName))
throw new ArgumentException("Bucket name cannot be null or empty", nameof(bucketName));
try
{
var args = new MakeBucketArgs().WithBucket(bucketName);
if (!string.IsNullOrWhiteSpace(region))
{
args = args.WithLocation(region);
}
Task.Run(async () =>
await _client.MakeBucketAsync(args, CancellationToken)).GetAwaiter().GetResult();
}
catch (Exception ex)
{
throw new InvalidOperationException($"Failed to create bucket '{bucketName}': {ex.Message}", ex);
}
}
///
/// Deletes a bucket synchronously
///
/// Name of the bucket to delete
public void DeleteBucket(string bucketName)
{
if (string.IsNullOrWhiteSpace(bucketName))
throw new ArgumentException("Bucket name cannot be null or empty", nameof(bucketName));
try
{
var args = new RemoveBucketArgs().WithBucket(bucketName);
Task.Run(async () =>
await _client.RemoveBucketAsync(args, CancellationToken)).GetAwaiter().GetResult();
}
catch (Exception ex)
{
throw new InvalidOperationException($"Failed to delete bucket '{bucketName}': {ex.Message}", ex);
}
}
///
/// Lists objects in a bucket synchronously
///
/// Name of the bucket
/// Optional prefix to filter objects
/// Whether to list objects recursively
/// Whether to include all versions of objects
/// List of object information
public List ListObjects(string bucketName, string? prefix = null, bool recursive = true, bool includeVersions = false)
{
if (string.IsNullOrWhiteSpace(bucketName))
throw new ArgumentException("Bucket name cannot be null or empty", nameof(bucketName));
try
{
var args = new ListObjectsArgs()
.WithBucket(bucketName)
.WithRecursive(recursive);
if (!string.IsNullOrWhiteSpace(prefix))
{
args = args.WithPrefix(prefix);
}
// Add version support if requested
if (includeVersions)
{
args = args.WithVersions(true);
}
var objects = new List();
var observable = _client.ListObjectsAsync(args, CancellationToken);
// Convert observable to synchronous list
var task = Task.Run(() =>
{
var tcs = new TaskCompletionSource();
observable.Subscribe(
onNext: item => objects.Add(MinIOObjectInfo.FromMinioItem(item, bucketName)),
onError: ex => tcs.SetException(ex),
onCompleted: () => tcs.SetResult(true)
);
return tcs.Task;
});
task.GetAwaiter().GetResult();
return objects;
}
catch (Exception ex)
{
throw new InvalidOperationException($"Failed to list objects in bucket '{bucketName}': {ex.Message}", ex);
}
}
///
/// Gets bucket policy synchronously
///
/// Name of the bucket
/// Bucket policy as JSON string
public string GetBucketPolicy(string bucketName)
{
if (string.IsNullOrWhiteSpace(bucketName))
throw new ArgumentException("Bucket name cannot be null or empty", nameof(bucketName));
try
{
var args = new GetPolicyArgs().WithBucket(bucketName);
return Task.Run(async () =>
await _client.GetPolicyAsync(args, CancellationToken)).GetAwaiter().GetResult();
}
catch (Exception ex)
{
throw new InvalidOperationException($"Failed to get policy for bucket '{bucketName}': {ex.Message}", ex);
}
}
///
/// Sets bucket policy synchronously
///
/// Name of the bucket
/// Policy JSON string
public void SetBucketPolicy(string bucketName, string policy)
{
if (string.IsNullOrWhiteSpace(bucketName))
throw new ArgumentException("Bucket name cannot be null or empty", nameof(bucketName));
if (string.IsNullOrWhiteSpace(policy))
throw new ArgumentException("Policy cannot be null or empty", nameof(policy));
try
{
var args = new SetPolicyArgs()
.WithBucket(bucketName)
.WithPolicy(policy);
Task.Run(async () =>
await _client.SetPolicyAsync(args, CancellationToken)).GetAwaiter().GetResult();
}
catch (Exception ex)
{
throw new InvalidOperationException($"Failed to set policy for bucket '{bucketName}': {ex.Message}", ex);
}
}
///
/// Deletes an object synchronously
///
/// Name of the bucket
/// Name of the object to delete
public void DeleteObject(string bucketName, string objectName)
{
if (string.IsNullOrWhiteSpace(bucketName))
throw new ArgumentException("Bucket name cannot be null or empty", nameof(bucketName));
if (string.IsNullOrWhiteSpace(objectName))
throw new ArgumentException("Object name cannot be null or empty", nameof(objectName));
try
{
var args = new RemoveObjectArgs()
.WithBucket(bucketName)
.WithObject(objectName);
Task.Run(async () =>
await _client.RemoveObjectAsync(args, CancellationToken)).GetAwaiter().GetResult();
}
catch (Exception ex)
{
throw new InvalidOperationException($"Failed to delete object '{objectName}' from bucket '{bucketName}': {ex.Message}", ex);
}
}
///
/// Deletes multiple objects synchronously
///
/// Name of the bucket
/// List of object names to delete
public void DeleteObjects(string bucketName, IEnumerable objectNames)
{
if (string.IsNullOrWhiteSpace(bucketName))
throw new ArgumentException("Bucket name cannot be null or empty", nameof(bucketName));
if (objectNames == null)
throw new ArgumentNullException(nameof(objectNames));
var objectList = objectNames.ToList();
if (objectList.Count == 0)
return;
try
{
var deleteObjectsArgs = new RemoveObjectsArgs()
.WithBucket(bucketName)
.WithObjects(objectList);
var observableTask = _client.RemoveObjectsAsync(deleteObjectsArgs, CancellationToken);
// Convert observable to synchronous operation
var task = Task.Run(async () =>
{
var observable = await observableTask;
var tcs = new TaskCompletionSource();
observable.Subscribe(
onNext: deleteError =>
{
// In MinIO 5.0.0, DeleteError might have different properties
// For now, just check if there's an error and report it
if (!string.IsNullOrEmpty(deleteError.Message))
{
tcs.SetException(new InvalidOperationException($"Failed to delete object '{deleteError.Key}': {deleteError.Message}"));
}
},
onError: ex => tcs.SetException(ex),
onCompleted: () => tcs.SetResult(true)
);
return await tcs.Task;
});
task.GetAwaiter().GetResult();
}
catch (Exception ex)
{
throw new InvalidOperationException($"Failed to delete objects from bucket '{bucketName}': {ex.Message}", ex);
}
}
///
/// Uploads a file to MinIO synchronously with progress reporting
///
/// Name of the bucket
/// Name of the object
/// Path to the file to upload
/// Content type of the file (optional)
/// Progress callback for reporting upload progress
/// ETag of the uploaded object
public string UploadFile(string bucketName, string objectName, string filePath,
string? contentType = null, Action? progressCallback = null)
{
if (string.IsNullOrWhiteSpace(bucketName))
throw new ArgumentException("Bucket name cannot be null or empty", nameof(bucketName));
if (string.IsNullOrWhiteSpace(objectName))
throw new ArgumentException("Object name cannot be null or empty", nameof(objectName));
if (string.IsNullOrWhiteSpace(filePath))
throw new ArgumentException("File path cannot be null or empty", nameof(filePath));
if (!File.Exists(filePath))
throw new FileNotFoundException($"File not found: {filePath}");
try
{
var fileInfo = new FileInfo(filePath);
var fileSize = fileInfo.Length;
// Determine content type if not provided
if (string.IsNullOrWhiteSpace(contentType))
{
contentType = GetContentType(filePath);
}
// Use explicit file stream management to ensure proper handle release
using var fileStream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.Read);
var args = new PutObjectArgs()
.WithBucket(bucketName)
.WithObject(objectName)
.WithStreamData(fileStream)
.WithObjectSize(fileSize)
.WithContentType(contentType);
// Progress tracking not available in MinIO 4.0.7
// progressCallback is ignored for now
Task.Run(async () =>
await _client.PutObjectAsync(args, CancellationToken)).GetAwaiter().GetResult();
return string.Empty; // MinIO 4.0.7 PutObjectAsync returns void
}
catch (Exception ex)
{
throw new InvalidOperationException($"Failed to upload file '{filePath}' to bucket '{bucketName}' as '{objectName}': {ex.Message}", ex);
}
}
///
/// Downloads an object from MinIO synchronously with progress reporting
///
/// Name of the bucket
/// Name of the object
/// Path where the file should be saved
/// Progress callback for reporting download progress
public void DownloadFile(string bucketName, string objectName, string filePath,
Action? progressCallback = null)
{
if (string.IsNullOrWhiteSpace(bucketName))
throw new ArgumentException("Bucket name cannot be null or empty", nameof(bucketName));
if (string.IsNullOrWhiteSpace(objectName))
throw new ArgumentException("Object name cannot be null or empty", nameof(objectName));
if (string.IsNullOrWhiteSpace(filePath))
throw new ArgumentException("File path cannot be null or empty", nameof(filePath));
try
{
// Ensure the directory exists
var directory = Path.GetDirectoryName(filePath);
if (!string.IsNullOrEmpty(directory) && !Directory.Exists(directory))
{
Directory.CreateDirectory(directory);
}
var args = new GetObjectArgs()
.WithBucket(bucketName)
.WithObject(objectName)
.WithFile(filePath);
// Progress tracking not available in MinIO 5.0.0
// progressCallback is ignored for now
Task.Run(async () =>
await _client.GetObjectAsync(args, CancellationToken)).GetAwaiter().GetResult();
}
catch (Exception ex)
{
throw new InvalidOperationException($"Failed to download object '{objectName}' from bucket '{bucketName}' to '{filePath}': {ex.Message}", ex);
}
}
///
/// Creates a directory object in MinIO (zero-byte object with trailing slash)
///
/// Name of the bucket
/// Path of the directory (should end with /)
/// ETag of the created directory object
public string CreateDirectory(string bucketName, string directoryPath)
{
if (string.IsNullOrWhiteSpace(bucketName))
throw new ArgumentException("Bucket name cannot be null or empty", nameof(bucketName));
if (string.IsNullOrWhiteSpace(directoryPath))
throw new ArgumentException("Directory path cannot be null or empty", nameof(directoryPath));
// Ensure directory path ends with /
if (!directoryPath.EndsWith("/"))
directoryPath += "/";
try
{
// Use a properly configured empty stream
using var emptyStream = new MemoryStream(new byte[0]);
emptyStream.Position = 0;
var args = new PutObjectArgs()
.WithBucket(bucketName)
.WithObject(directoryPath)
.WithStreamData(emptyStream)
.WithObjectSize(0)
.WithContentType("application/x-directory");
Task.Run(async () =>
await _client.PutObjectAsync(args, CancellationToken)).GetAwaiter().GetResult();
return ""; // Directory objects don't have meaningful ETags
}
catch (Exception ex)
{
throw new InvalidOperationException($"Failed to create directory '{directoryPath}' in bucket '{bucketName}': {ex.Message}", ex);
}
}
///
/// Uploads a stream to MinIO synchronously
///
/// Name of the bucket
/// Name of the object
/// Stream containing the data to upload
/// Content type of the data
/// Progress callback for reporting upload progress
/// ETag of the uploaded object
public string UploadStream(string bucketName, string objectName, Stream data,
string contentType = "application/octet-stream", Action? progressCallback = null)
{
if (string.IsNullOrWhiteSpace(bucketName))
throw new ArgumentException("Bucket name cannot be null or empty", nameof(bucketName));
if (string.IsNullOrWhiteSpace(objectName))
throw new ArgumentException("Object name cannot be null or empty", nameof(objectName));
if (data == null)
throw new ArgumentNullException(nameof(data));
try
{
// Ensure stream is at the beginning
if (data.CanSeek)
{
data.Position = 0;
}
var args = new PutObjectArgs()
.WithBucket(bucketName)
.WithObject(objectName)
.WithStreamData(data)
.WithObjectSize(data.Length)
.WithContentType(contentType);
// Progress tracking not available in MinIO 5.0.0
// progressCallback is ignored for now
Task.Run(async () =>
await _client.PutObjectAsync(args, CancellationToken)).GetAwaiter().GetResult();
return string.Empty; // MinIO 4.0.7 PutObjectAsync returns void
}
catch (Exception ex)
{
throw new InvalidOperationException($"Failed to upload stream to bucket '{bucketName}' as '{objectName}': {ex.Message}", ex);
}
}
///
/// Gets the content type for a file based on its extension
///
/// Path to the file
/// Content type string
private static string GetContentType(string filePath)
{
var extension = Path.GetExtension(filePath).ToLowerInvariant();
return extension switch
{
".txt" => "text/plain",
".html" => "text/html",
".css" => "text/css",
".js" => "application/javascript",
".json" => "application/json",
".xml" => "application/xml",
".pdf" => "application/pdf",
".zip" => "application/zip",
".jpg" or ".jpeg" => "image/jpeg",
".png" => "image/png",
".gif" => "image/gif",
".svg" => "image/svg+xml",
".mp4" => "video/mp4",
".mp3" => "audio/mpeg",
".wav" => "audio/wav",
_ => "application/octet-stream"
};
}
///
/// Lists object versions in a bucket synchronously
///
/// Name of the bucket
/// Optional prefix to filter objects
/// Whether to list objects recursively
/// Maximum number of objects to return (0 = unlimited)
/// List of object information including versions
private List ListObjectVersions(string bucketName, string? prefix, bool recursive, int maxObjects)
{
try
{
// For now, fall back to regular object listing since version listing
// may not be available in all MinIO SDK versions
// This can be enhanced when the SDK supports it
return ListObjects(bucketName, prefix, recursive, false);
}
catch (Exception ex)
{
throw new InvalidOperationException($"Failed to list object versions in bucket '{bucketName}': {ex.Message}", ex);
}
}
///
/// Generates a presigned URL for an object
///
/// Name of the bucket
/// Name of the object
/// URL expiry time
/// Presigned URL
public string GetPresignedUrl(string bucketName, string objectName, TimeSpan expiry)
{
if (string.IsNullOrWhiteSpace(bucketName))
throw new ArgumentException("Bucket name cannot be null or empty", nameof(bucketName));
if (string.IsNullOrWhiteSpace(objectName))
throw new ArgumentException("Object name cannot be null or empty", nameof(objectName));
try
{
var args = new PresignedGetObjectArgs()
.WithBucket(bucketName)
.WithObject(objectName)
.WithExpiry((int)expiry.TotalSeconds);
var result = Task.Run(async () =>
await _client.PresignedGetObjectAsync(args)).GetAwaiter().GetResult();
return result ?? string.Empty;
}
catch (Exception ex)
{
throw new InvalidOperationException($"Failed to generate presigned URL for object '{objectName}' in bucket '{bucketName}': {ex.Message}", ex);
}
}
///
/// Uploads a file using chunked transfer with resume capability
///
/// Transfer state for resume functionality
/// Thread-safe progress reporter for updates
/// Maximum retry attempts per chunk
/// MinIOObjectInfo of uploaded object or null if failed
public MinIOObjectInfo? UploadFileChunked(
ChunkedTransferState transferState,
ThreadSafeChunkedProgressReporter progressReporter,
int maxRetries = 3)
{
if (transferState == null)
throw new ArgumentNullException(nameof(transferState));
try
{
// For now, implement chunked upload using regular PutObject with progress tracking
// This simulates chunked behavior by reading the file in chunks and reporting progress
using var fileStream = new FileStream(transferState.FilePath, FileMode.Open, FileAccess.Read, FileShare.Read);
var totalChunks = (int)Math.Ceiling((double)transferState.TotalSize / transferState.ChunkSize);
var buffer = new byte[transferState.ChunkSize];
long totalBytesRead = 0;
// Create a progress-tracking stream wrapper
var progressStream = new ProgressTrackingStream(fileStream, (bytesRead) =>
{
var currentChunk = (int)(totalBytesRead / transferState.ChunkSize) + 1;
var chunkProgress = bytesRead % transferState.ChunkSize;
if (currentChunk <= totalChunks)
{
progressReporter.StartNewChunk(currentChunk, Math.Min(transferState.ChunkSize, transferState.TotalSize - totalBytesRead));
progressReporter.UpdateChunkProgress(chunkProgress);
if (chunkProgress == 0 && bytesRead > 0) // Chunk completed
{
progressReporter.CompleteChunk();
}
}
totalBytesRead = bytesRead;
});
// Upload the file
var putArgs = new PutObjectArgs()
.WithBucket(transferState.BucketName)
.WithObject(transferState.ObjectName)
.WithStreamData(progressStream)
.WithObjectSize(transferState.TotalSize);
Task.Run(async () =>
{
await _client.PutObjectAsync(putArgs, CancellationToken);
}).GetAwaiter().GetResult();
// Return object information
return new MinIOObjectInfo(
transferState.ObjectName,
transferState.TotalSize,
DateTime.UtcNow,
"simulated-etag", // MinIO 5.0.0 PutObject doesn't return ETag directly
transferState.BucketName);
}
catch (Exception ex)
{
throw new InvalidOperationException($"Chunked upload failed for object '{transferState.ObjectName}': {ex.Message}", ex);
}
}
///
/// Progress tracking stream wrapper
///
private class ProgressTrackingStream : Stream
{
private readonly Stream _baseStream;
private readonly Action _progressCallback;
private long _totalBytesRead = 0;
public ProgressTrackingStream(Stream baseStream, Action progressCallback)
{
_baseStream = baseStream;
_progressCallback = progressCallback;
}
public override bool CanRead => _baseStream.CanRead;
public override bool CanSeek => _baseStream.CanSeek;
public override bool CanWrite => _baseStream.CanWrite;
public override long Length => _baseStream.Length;
public override long Position
{
get => _baseStream.Position;
set => _baseStream.Position = value;
}
public override int Read(byte[] buffer, int offset, int count)
{
var bytesRead = _baseStream.Read(buffer, offset, count);
_totalBytesRead += bytesRead;
_progressCallback(_totalBytesRead);
return bytesRead;
}
public override void Flush() => _baseStream.Flush();
public override long Seek(long offset, SeekOrigin origin) => _baseStream.Seek(offset, origin);
public override void SetLength(long value) => _baseStream.SetLength(value);
public override void Write(byte[] buffer, int offset, int count) => _baseStream.Write(buffer, offset, count);
protected override void Dispose(bool disposing)
{
if (disposing)
{
_baseStream?.Dispose();
}
base.Dispose(disposing);
}
}
///
/// Downloads a file using chunked transfer with resume capability
///
/// Transfer state for resume functionality
/// Thread-safe progress reporter for updates
/// Maximum retry attempts per chunk
/// Number of parallel chunk downloads
/// True if download succeeded, false otherwise
public bool DownloadFileChunked(
ChunkedTransferState transferState,
ThreadSafeChunkedProgressReporter progressReporter,
int maxRetries = 3,
int parallelDownloads = 3)
{
if (transferState == null)
throw new ArgumentNullException(nameof(transferState));
try
{
// Create or open the target file
using var fileStream = new FileStream(transferState.FilePath, FileMode.Create, FileAccess.Write, FileShare.None);
fileStream.SetLength(transferState.TotalSize);
// Get list of chunks to download (with safety limit)
var chunksToDownload = new List();
var maxChunks = Math.Min(transferState.TotalChunks, 10000); // Safety limit of 10,000 chunks
for (int i = 0; i < maxChunks && !transferState.IsComplete; i++)
{
var nextChunk = transferState.GetNextChunk();
if (nextChunk == null)
break;
chunksToDownload.Add(nextChunk);
}
if (chunksToDownload.Count == 0)
{
return true; // Already complete
}
// Download chunks (with limited parallelism)
var semaphore = new SemaphoreSlim(parallelDownloads, parallelDownloads);
var downloadTasks = chunksToDownload.Select(chunk =>
DownloadChunkAsync(transferState, chunk, fileStream, progressReporter, maxRetries, semaphore)).ToArray();
var results = Task.WhenAll(downloadTasks).GetAwaiter().GetResult();
// Check if all chunks downloaded successfully
var allSucceeded = results.All(r => r);
if (allSucceeded)
{
// Mark all chunks as completed
foreach (var chunk in chunksToDownload)
{
transferState.MarkChunkCompleted(chunk);
}
}
return allSucceeded;
}
catch (Exception ex)
{
throw new InvalidOperationException($"Chunked download failed for object '{transferState.ObjectName}': {ex.Message}", ex);
}
}
///
/// Downloads a single chunk asynchronously with retry logic
///
/// Transfer state
/// Chunk to download
/// Target file stream
/// Thread-safe progress reporter
/// Maximum retry attempts
/// Semaphore for controlling parallelism
/// True if chunk downloaded successfully
private async Task DownloadChunkAsync(
ChunkedTransferState transferState,
ChunkInfo chunk,
FileStream fileStream,
ThreadSafeChunkedProgressReporter progressReporter,
int maxRetries,
SemaphoreSlim semaphore)
{
await semaphore.WaitAsync(CancellationToken);
try
{
progressReporter.StartNewChunk(chunk.ChunkNumber + 1, chunk.Size);
for (int attempt = 1; attempt <= maxRetries; attempt++)
{
try
{
using var chunkStream = new MemoryStream();
// Use GetObjectAsync with offset and length for proper chunked download
var getArgs = new GetObjectArgs()
.WithBucket(transferState.BucketName)
.WithObject(transferState.ObjectName)
.WithCallbackStream((stream) =>
{
// Skip to the start position and read only the chunk size
var buffer = new byte[8192]; // 8KB buffer
long totalRead = 0;
long skipBytes = chunk.StartByte;
// Skip to start position
while (skipBytes > 0)
{
var toSkip = (int)Math.Min(skipBytes, buffer.Length);
var skipped = stream.Read(buffer, 0, toSkip);
if (skipped == 0) break; // End of stream
skipBytes -= skipped;
}
// Read the chunk data
while (totalRead < chunk.Size)
{
var toRead = (int)Math.Min(chunk.Size - totalRead, buffer.Length);
var bytesRead = stream.Read(buffer, 0, toRead);
if (bytesRead == 0) break; // End of stream
chunkStream.Write(buffer, 0, bytesRead);
totalRead += bytesRead;
}
});
await _client.GetObjectAsync(getArgs, CancellationToken);
// Write chunk to file at correct position
lock (fileStream)
{
fileStream.Seek(chunk.StartByte, SeekOrigin.Begin);
chunkStream.Seek(0, SeekOrigin.Begin);
chunkStream.CopyTo(fileStream);
fileStream.Flush();
}
progressReporter.UpdateChunkProgress(chunk.Size);
progressReporter.CompleteChunk();
chunk.IsCompleted = true;
return true;
}
catch (Exception ex) when (attempt < maxRetries)
{
progressReporter.ReportChunkError(ex, attempt, maxRetries);
// Exponential backoff
var delay = TimeSpan.FromSeconds(Math.Pow(2, attempt));
await Task.Delay(delay, CancellationToken);
}
catch (Exception ex)
{
progressReporter.ReportChunkError(ex, attempt, maxRetries);
chunk.LastError = ex.Message;
chunk.RetryCount = attempt;
return false;
}
}
return false;
}
finally
{
semaphore.Release();
}
}
///
/// Cancels all ongoing operations
///
public void CancelOperations()
{
_cancellationTokenSource.Cancel();
}
///
/// Disposes the wrapper and underlying client
///
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
///
/// Protected dispose method
///
/// Whether disposing from Dispose method
protected virtual void Dispose(bool disposing)
{
if (!_disposed && disposing)
{
_cancellationTokenSource?.Cancel();
_cancellationTokenSource?.Dispose();
_client?.Dispose();
_disposed = true;
}
}
}
}