blobs

Author: KSemenenko

Directory.Build.props

@@ -13,8 +13,8 @@
         <PackageLicenseExpression>MIT</PackageLicenseExpression>
         <PackageReadmeFile>README.md</PackageReadmeFile>
         <Product>Managed Code - Storage</Product>
-        <Version>2.0.10</Version>
-        <PackageVersion>2.0.10</PackageVersion>
+        <Version>2.0.11</Version>
+        <PackageVersion>2.0.11</PackageVersion>
     </PropertyGroup>
     <PropertyGroup Condition="'$(GITHUB_ACTIONS)' == 'true'">
         <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>

ManagedCode.Storage.Azure/BlobStream.cs

@@ -1,8 +1,6 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
-using System.Threading;
-using System.Threading.Tasks;
 using Azure.Storage.Blobs;
 using Azure.Storage.Blobs.Specialized;
 
@@ -14,7 +12,28 @@ public class BlobStream : Stream
     private const int PageSizeInBytes = 512;
     public const int DefaultBufferSize = 1024 * 1024 * 4;
     private readonly PageBlobClient _pageBlob;
+    private object _syncRoot = new ();
 
+    public static BlobStream OpenStream(PageBlobClient pageBlob)
+    {
+        if (!pageBlob.Exists())
+        {
+            pageBlob.Create(0);
+        }
+
+        return new BlobStream(pageBlob);
+    }
+    
+    public static BufferedStream OpenBufferedStream(PageBlobClient pageBlob)
+    {
+        if (!pageBlob.Exists())
+        {
+            pageBlob.Create(0);
+        }
+
+        return new BufferedStream( new BlobStream(pageBlob), DefaultBufferSize);
+    }
+    
     public BlobStream(string connectionString, string container, string fileName)
         : this(GetClient(connectionString, container, fileName))
     {
@@ -26,7 +45,7 @@ public BlobStream(PageBlobClient pageBlob)
         _pageBlob.CreateIfNotExists(0);
     }
 
-    private long BlobLength => _pageBlob.GetProperties().Value.ContentLength;
+   private long BlobLength => _pageBlob.GetProperties().Value.ContentLength;
 
     public override bool CanRead => true;
 
@@ -80,52 +99,45 @@ public override long Seek(long offset, SeekOrigin origin)
                 Position += offset;
                 break;
             case SeekOrigin.End:
-                Position = Length - offset;
+                Position = Length + offset;
                 break;
         }
-
+        
         return Position;
     }
 
     public override void SetLength(long value)
     {
-        if (value > Length)
+        var length = Length;
+        if (value != length)
         {
             var newSize = NextPageAddress(value);
             _pageBlob.Resize(newSize);
         }
+        
+        SetLengthInternal(value);
     }
 
     public override int Read(byte[] buffer, int offset, int count)
     {
-        var bytesRead = 0;
-        using (var stream = _pageBlob.OpenRead(false, Position))
+        lock (_syncRoot)
         {
-            bytesRead = stream.Read(buffer, offset, count);
-            if (Position + count > Length)
+            int bytesRead = 0;
+            var length = Length;
+            
+            if (Position + count > length)
             {
-                bytesRead = (int)(Length - Position);
+                count = (int)(length - Position);
             }
-        }
-
-        Position += bytesRead;
-        return bytesRead;
-    }
-
-    public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
-    {
-        var bytesRead = 0;
-        using (var stream = await _pageBlob.OpenReadAsync(false, Position, cancellationToken: cancellationToken))
-        {
-            bytesRead = await stream.ReadAsync(buffer, offset, count, cancellationToken);
-            if (Position + count > Length)
+            
+            using (var stream = _pageBlob.OpenRead(false, Position))
             {
-                bytesRead = (int)(Length - Position);
+                bytesRead = stream.Read(buffer, offset, count);
             }
-        }
 
-        Position += bytesRead;
-        return bytesRead;
+            Position += bytesRead;
+            return bytesRead;
+        }
     }
 
     private void EnsureCapacity(long position)
@@ -137,84 +149,40 @@ private void EnsureCapacity(long position)
         }
     }
 
-    private async Task EnsureCapacityAsync(long position)
-    {
-        if (BlobLength < position)
-        {
-            var newSize = NextPageAddress(position);
-            await _pageBlob.ResizeAsync(newSize);
-        }
-    }
-    
     public override void Write(byte[] buffer, int offset, int count)
     {
-        EnsureCapacity(Position + count);
-
-        var pageStartAddress = PreviousPageAddress(Position);
-        var pageBytes = NextPageAddress(Position + count) - pageStartAddress;
-        var offsetInFirstPage = (int)(Position % PageSizeInBytes);
-        var offsetInLastPage = (offsetInFirstPage + count) % PageSizeInBytes;
-
-        var bufferToMerge = new byte[pageBytes];
-        if (offsetInFirstPage > 0 || (pageBytes > PageSizeInBytes && offsetInLastPage > 0))
+        lock (_syncRoot)
         {
-            var localCount = (int)(pageBytes - PageSizeInBytes);
-            using (var stream = _pageBlob.OpenRead(false, pageStartAddress))
+            var pageStartAddress = PreviousPageAddress(Position);
+            var pageBytes = NextPageAddress(Position + count) - pageStartAddress;
+            var offsetInFirstPage = (int)(Position % PageSizeInBytes);
+            var offsetInLastPage = (offsetInFirstPage + count) % PageSizeInBytes;
+
+            var bufferToMerge = new byte[pageBytes];
+            if (offsetInFirstPage > 0 || (pageBytes > PageSizeInBytes && offsetInLastPage > 0))
             {
-                _ = stream.Read(bufferToMerge, 0, localCount);
+                //var localCount = (int)(pageBytes - PageSizeInBytes);
+                using (var stream = _pageBlob.OpenRead(false, pageStartAddress))
+                {
+                    _ = stream.Read(bufferToMerge, 0, (int)pageBytes);
+                }
             }
-        }
 
-        Buffer.BlockCopy(buffer, offset, bufferToMerge, offsetInFirstPage, count);
-        
-        EnsureCapacity(pageStartAddress + bufferToMerge.Length);
-
-        using (var stream = _pageBlob.OpenWrite(false, pageStartAddress))
-        {
-            stream.Write(bufferToMerge, 0, bufferToMerge.Length);
-            stream.Flush();
-        }
-        
-        Position += count;
-        if (Position > Length)
-        {
-            SetLengthInternal(Position);
-        }
-    }
+            Buffer.BlockCopy(buffer, offset, bufferToMerge, offsetInFirstPage, count);
+            
+            EnsureCapacity(pageStartAddress + bufferToMerge.Length);
 
-    public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
-    {
-        //await EnsureCapacityAsync(Position + count);
-        
-        var pageStartAddress = PreviousPageAddress(Position);
-        var pageBytes = NextPageAddress(Position + count) - pageStartAddress;
-        var offsetInFirstPage = (int)(Position % PageSizeInBytes);
-        var offsetInLastPage = (offsetInFirstPage + count) % PageSizeInBytes;
-
-        var bufferToMerge = new byte[pageBytes];
-        if (offsetInFirstPage > 0 || (pageBytes > PageSizeInBytes && offsetInLastPage > 0))
-        {
-            var localCount = (int)(pageBytes - PageSizeInBytes);
-            using (var stream = await _pageBlob.OpenReadAsync(false, pageStartAddress, cancellationToken: cancellationToken))
+            using (var stream = _pageBlob.OpenWrite(false, pageStartAddress))
             {
-                _ = await stream.ReadAsync(bufferToMerge, 0, localCount, cancellationToken);
+                stream.Write(bufferToMerge, 0, bufferToMerge.Length);
+                stream.Flush();
             }
-        }
-
-        Buffer.BlockCopy(buffer, offset, bufferToMerge, offsetInFirstPage, count);
 
-        await EnsureCapacityAsync(pageStartAddress + bufferToMerge.Length);
-      
-        using (var stream = await _pageBlob.OpenWriteAsync(false, pageStartAddress, cancellationToken: cancellationToken))
-        {
-            await stream.WriteAsync(bufferToMerge, 0, bufferToMerge.Length, cancellationToken);
-            await stream.FlushAsync(cancellationToken);
-        }
-
-        Position += count;
-        if (Position > Length)
-        {
-            await SetLengthInternalAsync(Position, cancellationToken);
+            Position += count;
+            if (Position > Length)
+            {
+                SetLengthInternal(Position);
+            }
         }
     }
 
@@ -229,30 +197,12 @@ private long PreviousPageAddress(long position)
         var previousPageAddress = position - position % PageSizeInBytes;
         return previousPageAddress;
     }
-
-    public static BlobStream Open(PageBlobClient pageBlob)
-    {
-        if (!pageBlob.Exists())
-        {
-            pageBlob.Create(0);
-        }
-
-        return new BlobStream(pageBlob);
-    }
-
+    
     private void SetLengthInternal(long newLength)
     {
         _pageBlob.SetMetadata(new Dictionary<string, string>
         {
             [MetadataLengthKey] = newLength.ToString()
         });
     }
-
-    private Task SetLengthInternalAsync(long newLength, CancellationToken cancellationToken = default)
-    {
-        return _pageBlob.SetMetadataAsync(new Dictionary<string, string>
-        {
-            [MetadataLengthKey] = newLength.ToString()
-        }, cancellationToken: cancellationToken);
-    }
 }