Merge pull request #7 from mwilde/support_checksum_algorithms

mlhpdx · web-flow · commit 4da619bc8f52 · 2024-12-19T12:23:05.000-08:00
Enable support for checksum calculation
diff --git a/Examples/DecompressGzAndUpload/DecompressGzAndUpload.csproj b/Examples/DecompressGzAndUpload/DecompressGzAndUpload.csproj
@@ -7,7 +7,7 @@
   <ItemGroup>
     <PackageReference Include="Amazon.Lambda.Core" Version="2.5.0" />
     <PackageReference Include="Amazon.Lambda.Serialization.SystemTextJson" Version="2.4.4" />
-    <PackageReference Include="AWSSDK.S3" Version="3.7.410.1" />
+    <PackageReference Include="AWSSDK.S3" Version="3.7.410.3" />
     <PackageReference Include="System.Text.Json" Version="9.0.0" />
   </ItemGroup>
   <ItemGroup>
diff --git a/Examples/DecryptAesAndUpload/DecryptAesAndUpload.csproj b/Examples/DecryptAesAndUpload/DecryptAesAndUpload.csproj
@@ -7,7 +7,7 @@
   <ItemGroup>
     <PackageReference Include="Amazon.Lambda.Core" Version="2.5.0" />
     <PackageReference Include="Amazon.Lambda.Serialization.SystemTextJson" Version="2.4.4" />
-    <PackageReference Include="AWSSDK.S3" Version="3.7.410.1" />
+    <PackageReference Include="AWSSDK.S3" Version="3.7.410.3" />
     <PackageReference Include="System.Text.Json" Version="9.0.0" />
   </ItemGroup>
   <ItemGroup>
diff --git a/Examples/DecryptPgpAndUpload/DecryptPgpAndUpload.csproj b/Examples/DecryptPgpAndUpload/DecryptPgpAndUpload.csproj
@@ -7,7 +7,7 @@
   <ItemGroup>
     <PackageReference Include="Amazon.Lambda.Core" Version="2.5.0" />
     <PackageReference Include="Amazon.Lambda.Serialization.SystemTextJson" Version="2.4.4" />
-    <PackageReference Include="AWSSDK.S3" Version="3.7.410.1" />
+    <PackageReference Include="AWSSDK.S3" Version="3.7.410.3" />
     <PackageReference Include="PgpCore" Version="6.5.1" />
     <PackageReference Include="SeekableS3Stream" Version="1.0.2" />
     <PackageReference Include="System.Text.Json" Version="9.0.0" />
diff --git a/Examples/ZipS3ListToS3/ZipS3ListToS3.csproj b/Examples/ZipS3ListToS3/ZipS3ListToS3.csproj
@@ -7,7 +7,7 @@
   <ItemGroup>
     <PackageReference Include="Amazon.Lambda.Core" Version="2.5.0" />
     <PackageReference Include="Amazon.Lambda.Serialization.SystemTextJson" Version="2.4.4" />
-    <PackageReference Include="AWSSDK.S3" Version="3.7.410.1" />
+    <PackageReference Include="AWSSDK.S3" Version="3.7.410.3" />
     <PackageReference Include="System.Text.Json" Version="9.0.0" />
   </ItemGroup>
   <ItemGroup>
diff --git a/S3UploadStream/S3UploadStream.cs b/S3UploadStream/S3UploadStream.cs
@@ -5,7 +5,9 @@
 using System.Collections.Generic;
 using System.IO;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
+using Amazon.Runtime;
 
 namespace Cppl.Utilities.AWS
 {
@@ -21,40 +23,77 @@ public class S3UploadStream : Stream
 
         internal class Metadata
         {
-            public string BucketName;
-            public string Key;
             public long PartLength = DEFAULT_PART_LENGTH;
 
             public int PartCount = 0;
             public string UploadId;
             public MemoryStream CurrentStream;
+            public CancellationToken CancellationToken;
 
             public long Position = 0; // based on bytes written
             public long Length = 0; // based on bytes written or SetLength, whichever is larger (no truncation)
 
             public List<Task> Tasks = new List<Task>();
-            public ConcurrentDictionary<int, string> PartETags = new ConcurrentDictionary<int, string>();
+            public ConcurrentDictionary<int, PartETag> PartETags = new ConcurrentDictionary<int, PartETag>();
+
+            public InitiateMultipartUploadRequest InitiateMultipartUploadRequest;
         }
 
-        Metadata _metadata = new Metadata();
-        IAmazonS3 _s3 = null;
+        private readonly IAmazonS3 _s3 = null;
+        private Metadata _metadata = new Metadata();
+
+        public event Action<InitiateMultipartUploadResponse> Initiated;
+        public event Action<UploadPartResponse> UploadedPart;
+        public event Action<StreamTransferProgressArgs> StreamTransfer;
+        public event Action<CompleteMultipartUploadResponse> Completed;
+
 
-        public S3UploadStream(IAmazonS3 s3, string s3uri, long partLength = DEFAULT_PART_LENGTH)
-            : this(s3, new Uri(s3uri), partLength)
+        public S3UploadStream(
+            IAmazonS3 s3,
+            string s3uri,
+            long partLength = DEFAULT_PART_LENGTH,
+            CancellationToken token = default)
+            : this(s3, new Uri(s3uri), partLength, token)
         {
         }
 
-        public S3UploadStream(IAmazonS3 s3, Uri s3uri, long partLength = DEFAULT_PART_LENGTH)
-            : this (s3, s3uri.Host, s3uri.LocalPath.Substring(1), partLength)
+        public S3UploadStream(
+            IAmazonS3 s3,
+            Uri s3uri,
+            long partLength = DEFAULT_PART_LENGTH,
+            CancellationToken token = default)
+            : this(s3, s3uri.Host, s3uri.LocalPath.Substring(1), partLength, token)
         {
         }
 
-        public S3UploadStream(IAmazonS3 s3, string bucket, string key, long partLength = DEFAULT_PART_LENGTH)
+        public S3UploadStream(
+            IAmazonS3 s3,
+            string bucket,
+            string key,
+            long partLength = DEFAULT_PART_LENGTH,
+            CancellationToken token = default)
+            : this(
+                s3,
+                new InitiateMultipartUploadRequest
+                {
+                    BucketName = bucket,
+                    Key = key
+                },
+                partLength,
+                token)
+        {
+        }
+
+        public S3UploadStream(
+            IAmazonS3 s3,
+            InitiateMultipartUploadRequest initiateMultipartUploadRequest,
+            long partLength = DEFAULT_PART_LENGTH,
+            CancellationToken token = default)
         {
             _s3 = s3;
-            _metadata.BucketName = bucket;
-            _metadata.Key = key;
             _metadata.PartLength = partLength;
+            _metadata.InitiateMultipartUploadRequest = initiateMultipartUploadRequest;
+            _metadata.CancellationToken = token;
         }
 
         protected override void Dispose(bool disposing)
@@ -67,10 +106,11 @@ protected override void Dispose(bool disposing)
                     CompleteUpload();
                 }
             }
+
             _metadata = null;
             base.Dispose(disposing);
         }
-    
+
         public override bool CanRead => false;
         public override bool CanSeek => false;
         public override bool CanWrite => true;
@@ -83,21 +123,26 @@ public override long Position
         }
 
         public override int Read(byte[] buffer, int offset, int count) => throw new NotImplementedException();
+
         public override long Seek(long offset, SeekOrigin origin) => throw new NotImplementedException();
 
         public override void SetLength(long value)
         {
             _metadata.Length = Math.Max(_metadata.Length, value);
-            _metadata.PartLength = Math.Max(MIN_PART_LENGTH, Math.Min(MAX_PART_LENGTH, _metadata.Length / MAX_PART_COUNT));
+            _metadata.PartLength =
+                Math.Max(MIN_PART_LENGTH, Math.Min(MAX_PART_LENGTH, _metadata.Length / MAX_PART_COUNT));
         }
 
         private void StartNewPart()
         {
-            if (_metadata.CurrentStream != null) {
+            if (_metadata.CurrentStream != null)
+            {
                 Flush(false);
             }
+
             _metadata.CurrentStream = new MemoryStream();
-            _metadata.PartLength = Math.Min(MAX_PART_LENGTH, Math.Max(_metadata.PartLength, (_metadata.PartCount / 2 + 1) * MIN_PART_LENGTH)); 
+            _metadata.PartLength = Math.Min(MAX_PART_LENGTH,
+                Math.Max(_metadata.PartLength, (_metadata.PartCount / 2 + 1) * MIN_PART_LENGTH));
         }
 
         public override void Flush()
@@ -109,39 +154,60 @@ private void Flush(bool disposing)
         {
             if ((_metadata.CurrentStream == null || _metadata.CurrentStream.Length < MIN_PART_LENGTH) &&
                 !disposing)
+            {
                 return;
+            }
 
-            if (_metadata.UploadId == null) {
-                _metadata.UploadId = _s3.InitiateMultipartUploadAsync(new InitiateMultipartUploadRequest()
-                {
-                    BucketName = _metadata.BucketName,
-                    Key = _metadata.Key
-                }).GetAwaiter().GetResult().UploadId;
+            if (_metadata.UploadId == null)
+            {
+                var response = _s3
+                    .InitiateMultipartUploadAsync(_metadata.InitiateMultipartUploadRequest, _metadata.CancellationToken)
+                    .GetAwaiter().GetResult();
+                _metadata.CancellationToken.ThrowIfCancellationRequested();
+                Initiated?.Invoke(response);
+                _metadata.UploadId = response.UploadId;
             }
-            
+
             if (_metadata.CurrentStream != null)
             {
                 var i = ++_metadata.PartCount;
 
                 _metadata.CurrentStream.Seek(0, SeekOrigin.Begin);
                 var request = new UploadPartRequest()
                 {
-                    BucketName = _metadata.BucketName,
-                    Key = _metadata.Key,
+                    BucketName = _metadata.InitiateMultipartUploadRequest.BucketName,
+                    Key = _metadata.InitiateMultipartUploadRequest.Key,
                     UploadId = _metadata.UploadId,
                     PartNumber = i,
                     IsLastPart = disposing,
-                    InputStream = _metadata.CurrentStream
+                    InputStream = _metadata.CurrentStream,
+                    ChecksumAlgorithm = _metadata.InitiateMultipartUploadRequest.ChecksumAlgorithm,
                 };
                 _metadata.CurrentStream = null;
+                request.StreamTransferProgress += (_, progressArgs) => { StreamTransfer?.Invoke(progressArgs); };
 
-                var upload = Task.Run(async () =>
-                {
-                    var response = await _s3.UploadPartAsync(request);
-                    _metadata.PartETags.AddOrUpdate(i, response.ETag,
-                        (n, s) => response.ETag);
-                    request.InputStream.Dispose();
-                });
+                var upload = Task.Run(
+                    async () =>
+                    {
+                        var response = await _s3.UploadPartAsync(request, _metadata.CancellationToken);
+
+                        _metadata.CancellationToken.ThrowIfCancellationRequested();
+
+                        UploadedPart?.Invoke(response);
+
+                        var partETag = new PartETag
+                        {
+                            PartNumber = response.PartNumber,
+                            ETag = response.ETag,
+                            ChecksumSHA1 = response.ChecksumSHA1,
+                            ChecksumSHA256 = response.ChecksumSHA256,
+                            ChecksumCRC32 = response.ChecksumCRC32,
+                            ChecksumCRC32C = response.ChecksumCRC32C,
+                        };
+
+                        _metadata.PartETags.AddOrUpdate(i, partETag, (n, s) => partETag);
+                        request.InputStream.Dispose();
+                    });
                 _metadata.Tasks.Add(upload);
             }
         }
@@ -150,14 +216,19 @@ private void CompleteUpload()
         {
             Task.WaitAll(_metadata.Tasks.ToArray());
 
-            if (Length > 0) {
-                _s3.CompleteMultipartUploadAsync(new CompleteMultipartUploadRequest()
-                {
-                    BucketName = _metadata.BucketName,
-                    Key = _metadata.Key,
-                    PartETags = _metadata.PartETags.Select(e => new PartETag(e.Key, e.Value)).ToList(),
-                    UploadId = _metadata.UploadId
-                }).GetAwaiter().GetResult();
+            if (Length > 0)
+            {
+                _metadata.CancellationToken.ThrowIfCancellationRequested();
+
+                var response = _s3.CompleteMultipartUploadAsync(
+                    new CompleteMultipartUploadRequest()
+                    {
+                        BucketName = _metadata.InitiateMultipartUploadRequest.BucketName,
+                        Key = _metadata.InitiateMultipartUploadRequest.Key,
+                        PartETags = _metadata.PartETags.Values.ToList(),
+                        UploadId = _metadata.UploadId,
+                    }, _metadata.CancellationToken).GetAwaiter().GetResult();
+                Completed?.Invoke(response);
             }
         }
 
@@ -171,8 +242,20 @@ public override void Write(byte[] buffer, int offset, int count)
             var c = Math.Min(count, buffer.Length - offset); // don't over-read the buffer, even if asked to
             do
             {
+                if (_metadata.CancellationToken.IsCancellationRequested)
+                {
+                    return;
+                }
+
                 if (_metadata.CurrentStream == null || _metadata.CurrentStream.Length >= _metadata.PartLength)
+                {
                     StartNewPart();
+                }
+
+                if (_metadata.CurrentStream == null)
+                {
+                    throw new ArgumentNullException(nameof(Metadata.CurrentStream));
+                }
 
                 var remaining = _metadata.PartLength - _metadata.CurrentStream.Length;
                 var w = Math.Min(c, (int)remaining);
@@ -184,4 +267,4 @@ public override void Write(byte[] buffer, int offset, int count)
             } while (c > 0);
         }
     }
-}
+}
diff --git a/S3UploadStream/S3UploadStream.csproj b/S3UploadStream/S3UploadStream.csproj
@@ -15,7 +15,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="AWSSDK.S3" Version="3.7.410.1" />
+    <PackageReference Include="AWSSDK.S3" Version="3.7.410.3" />
     <None Include="../Readme.md" Pack="true" PackagePath="\" />
   </ItemGroup>
 
