add BoundedStream to return only a subset of a stream.

Author: huiguangjun

src/AlibabaCloud.OSS.V2/IO/BoundedStream.cs

@@ -0,0 +1,200 @@
+using System;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace AlibabaCloud.OSS.V2.IO
+{
+    /// <summary>
+    /// Represents read-only view over the portion of underlying stream.
+    /// </summary>
+    public sealed class BoundedStream : Stream
+    {
+        private long _length;
+        private long _offset;
+
+        /// <summary>
+        /// Creates a new BoundedStream that wraps the given stream.
+        /// </summary>
+        /// <param name="stream">The wrapped input stream.</param>
+        public BoundedStream(Stream stream)
+        {
+            this._length = stream.Length;
+            BaseStream = stream;
+        }
+
+        /// <summary>
+        /// Creates a new BoundedStream that wraps the given stream and limits it to a certain size.
+        /// </summary>
+        /// <param name="stream">The wrapped input stream.</param>
+        /// <param name="offset">The offset in the underlying stream.</param>
+        /// <param name="length">Total number of bytes to allow to be read from the stream. </param>
+        public BoundedStream(Stream stream, long offset, long length)
+        {
+            BaseStream = stream;
+            Adjust(offset, length);
+        }
+
+        /// <summary>
+        /// Gets underlying stream.
+        /// </summary>
+        public Stream BaseStream { get; private set; }
+
+        /// <summary>
+        /// Adjust the stream bounds.
+        /// </summary>
+        /// <remarks>
+        /// This method modifies <see cref="Stream.Position"/> property of the underlying stream.
+        /// </remarks>
+        /// <param name="offset">The offset in the underlying stream.</param>
+        /// <param name="length">Total number of bytes to allow to be read from the stream. </param>
+        /// <exception cref="ArgumentOutOfRangeException"><paramref name="length"/> is larger than the remaining length of the underlying stream; or <paramref name="offset"/> if greater than the length of the underlying stream.</exception>
+        public void Adjust(long offset, long length)
+        {
+            ThrowIfGreaterThan((ulong)offset, (ulong)BaseStream.Length, nameof(offset));
+            ThrowIfGreaterThan((ulong)length, (ulong)(BaseStream.Length - offset), nameof(length));
+            this._length = length;
+            this._offset = offset;
+            BaseStream.Position = offset;
+        }
+
+        /// <summary>
+        /// Gets a value indicating whether the current stream supports reading.
+        /// </summary>
+        /// <value><see langword="true"/> if the stream supports reading; otherwise, <see langword="false"/>.</value>
+        public override bool CanRead => BaseStream.CanRead;
+
+        /// <summary>
+        /// Gets a value indicating whether the current stream supports seeking.
+        /// </summary>
+        /// <value><see langword="true"/> if the stream supports seeking; otherwise, <see langword="false"/>.</value>
+        public override bool CanSeek => BaseStream.CanSeek;
+
+        /// <summary>
+        /// Gets a value indicating whether the current stream supports writing.
+        /// </summary>
+        /// <value>Always <see langword="false"/>.</value>
+        public override bool CanWrite => false;
+
+        /// <inheritdoc/>
+        public override long Length => _length;
+
+        /// <inheritdoc/>
+        public override long Position
+        {
+            get => BaseStream.Position - _offset;
+            set
+            {
+                ThrowIfGreaterThan(value, _length, nameof(value));
+                BaseStream.Position = _offset + value;
+            }
+        }
+
+        private long RemainingBytes => _length - Position;
+
+        /// <inheritdoc/>
+        public override void Flush() => BaseStream.Flush();
+
+        /// <inheritdoc/>
+        public override Task FlushAsync(CancellationToken token = default) => BaseStream.FlushAsync(token);
+
+        /// <inheritdoc/>
+        public override bool CanTimeout => BaseStream.CanTimeout;
+
+        /// <inheritdoc/>
+        public override int ReadByte()
+            => Position < _length ? BaseStream.ReadByte() : -1;
+
+        /// <inheritdoc/>
+        public override void WriteByte(byte value) => throw new NotSupportedException();
+
+        /// <inheritdoc/>
+        public override int Read(byte[] buffer, int offset, int count)
+        {
+            return BaseStream.Read(buffer, offset, (int)Math.Min(count, RemainingBytes));
+        }
+
+        /// <inheritdoc/>
+        public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state)
+        {
+            count = (int)Math.Min(count, RemainingBytes);
+            return BaseStream.BeginRead(buffer, offset, count, callback, state);
+        }
+
+        /// <inheritdoc/>
+        public override int EndRead(IAsyncResult asyncResult) => BaseStream.EndRead(asyncResult);
+
+        /// <inheritdoc/>
+        public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken token = default)
+            => BaseStream.ReadAsync(buffer, offset, (int)Math.Min(count, RemainingBytes), token);
+
+        /// <inheritdoc/>
+        public override long Seek(long offset, SeekOrigin origin)
+        {
+            var newPosition = origin switch
+            {
+                SeekOrigin.Begin => offset,
+                SeekOrigin.Current => Position + offset,
+                SeekOrigin.End => _length + offset,
+                _ => throw new ArgumentOutOfRangeException(nameof(origin))
+            };
+
+            if (newPosition < 0L)
+                throw new IOException();
+
+            ThrowIfGreaterThan(newPosition, _length, nameof(offset));
+
+            Position = newPosition;
+            return newPosition;
+        }
+
+        /// <inheritdoc/>
+        public override void SetLength(long value)
+        {
+            ThrowIfGreaterThan((ulong)value, (ulong)(BaseStream.Length - BaseStream.Position), nameof(value));
+            _length = value;
+        }
+
+        /// <inheritdoc/>
+        public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
+
+        /// <inheritdoc/>
+        public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken token = default) => Task.FromException(new NotSupportedException());
+
+        /// <inheritdoc/>
+        public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state)
+            => throw new NotSupportedException();
+
+        /// <inheritdoc/>
+        public override void EndWrite(IAsyncResult asyncResult) => throw new InvalidOperationException();
+
+        /// <inheritdoc/>
+        public override int ReadTimeout
+        {
+            get => BaseStream.ReadTimeout;
+            set => BaseStream.ReadTimeout = value;
+        }
+
+        /// <inheritdoc/>
+        public override int WriteTimeout
+        {
+            get => BaseStream.WriteTimeout;
+            set => BaseStream.WriteTimeout = value;
+        }
+
+        /// <inheritdoc/>
+        protected override void Dispose(bool disposing)
+        {
+            base.Dispose(disposing);
+            BaseStream.Dispose();
+        }
+
+        private static void ThrowIfGreaterThan<T>(T value, T other, string? paramName = null) where T : IComparable<T>
+        {
+            if (value.CompareTo(other) > 0)
+            {
+                throw new ArgumentOutOfRangeException(paramName, value, $"{paramName} ('{value}')must be less than or equal to  '{other}'");
+            }
+        }
+    }
+}

test/AlibabaCloud.OSS.V2.IntegrationTests/ClientMiscTest.cs

@@ -1,5 +1,6 @@
 using System.Text;
 using AlibabaCloud.OSS.V2.Models;
+using AlibabaCloud.OSS.V2.IO;
 
 namespace AlibabaCloud.OSS.V2.IntegrationTests;
 
@@ -1005,4 +1006,106 @@ public async Task TestQueryWithSpecialChar()
         var got1 = await reader1.ReadToEndAsync();
         Assert.Equal(content, got1);
     }
+
+    [Fact]
+    public async Task TestUploadPartFromFile()
+    {
+        var client = Utils.GetDefaultClient();
+
+        //default
+        var bucketName = Utils.RandomBucketName(BucketNamePrefix);
+
+        var result = await client.PutBucketAsync(new()
+        {
+            Bucket = bucketName
+        });
+
+        Assert.NotNull(result);
+        Assert.Equal(200, result.StatusCode);
+        Assert.NotNull(result.RequestId);
+
+        var partSize = 120 * 1024;
+        var filepath = Utils.RandomFilePath(RootPath);
+        var saveFilepath = Utils.RandomFilePath(RootPath);
+        Utils.PrepareSampleFile(filepath, 512);
+        Assert.True(File.Exists(filepath));
+
+        // init
+        var objectName = Utils.RandomObjectName();
+        var initResult = await client.InitiateMultipartUploadAsync(new()
+        {
+            Bucket = bucketName,
+            Key = objectName
+        }
+        );
+        Assert.NotNull(initResult);
+        Assert.Equal(200, initResult.StatusCode);
+        Assert.NotNull(initResult.RequestId);
+        Assert.Equal("url", initResult.EncodingType);
+
+        // upload
+        using var file = File.OpenRead(filepath);
+        var fileSize = file.Length;
+        long partNumber = 1;
+        for (int offset = 0; offset < fileSize; offset += partSize)
+        {
+            var size = (long)Math.Min(partSize, fileSize - offset);
+            var upResult = await client.UploadPartAsync(new()
+            {
+                Bucket = bucketName,
+                Key = objectName,
+                PartNumber = partNumber,
+                UploadId = initResult.UploadId,
+                Body = new BoundedStream(file, (long)offset, size)
+            });
+
+            partNumber++;
+
+            Assert.NotNull(upResult);
+            Assert.Equal(200, upResult.StatusCode);
+            Assert.NotNull(upResult.RequestId);
+        }
+
+        // complete
+        var cmResult = await client.CompleteMultipartUploadAsync(new()
+        {
+            Bucket = bucketName,
+            Key = objectName,
+            UploadId = initResult.UploadId,
+            CompleteAll = "yes"
+        });
+        Assert.NotNull(cmResult);
+        Assert.Equal(200, cmResult.StatusCode);
+        Assert.NotNull(cmResult.RequestId);
+        Assert.Equal("url", cmResult.EncodingType);
+
+        // get object
+        var getObjectResult = await client.GetObjectAsync(
+            new()
+            {
+                Bucket = bucketName,
+                Key = objectName
+            }
+        );
+        Assert.NotNull(getObjectResult);
+        Assert.Equal(200, getObjectResult.StatusCode);
+        Assert.Equal((long)fileSize, getObjectResult.ContentLength);
+        Assert.Equal("Multipart", getObjectResult.ObjectType);
+        Assert.Null(getObjectResult.TaggingCount);
+        Assert.Null(getObjectResult.ContentMd5);
+        Assert.NotNull(getObjectResult.StorageClass);
+        Assert.NotNull(getObjectResult.Metadata);
+        Assert.Empty(getObjectResult.Metadata);
+        Assert.NotNull(getObjectResult.Body);
+        Assert.False(getObjectResult.Body.CanSeek);
+
+        using var saveStream = File.OpenWrite(saveFilepath);
+        await getObjectResult.Body.CopyToAsync(saveStream);
+        saveStream.Close();
+
+        var srcMd5 = Utils.ComputeContentMd5(filepath);
+        var dstMd5 = Utils.ComputeContentMd5(saveFilepath);
+        Assert.NotEmpty(srcMd5);
+        Assert.Equal(srcMd5, dstMd5);
+    }
 }