ChecksumCalculatingStream
Checksums stream contents as the stream is progressed through.
Limited seek support on read. No seek support on write.
All properties pass through except CanRead,
CanWrite, CanSeek, and
Positionset (get will pass through).
Passes thorugh properties traditionally locked behind seeking (e.g.
Length) regardless of read or write mode, ensuring
expected stream checks are not interrupted and that exceptions only
throw when unsupported behavior is actually attempted. This is done
to maintain existing stream behavior as closely as possible, as this
class is meant to only observe the stream contents as they flow.
using Azure.Core.Pipeline;
using Azure.Storage.Common;
using System;
using System.IO;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
namespace Azure.Storage
{
internal class ChecksumCalculatingStream : Stream
{
public delegate void AppendChecksumCalculation (ReadOnlySpan<byte> checksum);
private readonly Stream _stream;
private readonly AppendChecksumCalculation _appendChecksumCalculation;
private readonly long _initialPosition;
private long _nextToBeChecksummedPosition;
public override bool CanRead { get; }
public override bool CanWrite { get; }
public override bool CanSeek {
get {
if (!CanWrite)
return _stream.CanSeek;
return false;
}
}
public override long Length => _stream.Length;
public override long Position {
get {
return _stream.Position;
}
set {
Seek(value, SeekOrigin.Begin);
}
}
public override bool CanTimeout => _stream.CanTimeout;
public override int ReadTimeout {
get {
return _stream.ReadTimeout;
}
set {
_stream.ReadTimeout = value;
}
}
public override int WriteTimeout {
get {
return _stream.WriteTimeout;
}
set {
_stream.WriteTimeout = value;
}
}
public static Stream GetReadStream(Stream stream, AppendChecksumCalculation appendChecksumCalculation)
{
return new ChecksumCalculatingStream(stream, appendChecksumCalculation, true);
}
public static Stream GetWriteStream(Stream stream, AppendChecksumCalculation appendChecksumCalculation)
{
return new ChecksumCalculatingStream(stream, appendChecksumCalculation, false);
}
private ChecksumCalculatingStream(Stream stream, AppendChecksumCalculation appendChecksumCalculation, bool isReadMode)
{
Azure.Storage.Common.Argument.AssertNotNull(stream, "stream");
Azure.Storage.Common.Argument.AssertNotNull(appendChecksumCalculation, "appendChecksumCalculation");
CanRead = isReadMode;
CanWrite = !isReadMode;
if (CanRead && !stream.CanRead)
throw new ArgumentException("Created stream in read mode but base stream cannot read.");
if (CanWrite && !stream.CanWrite)
throw new ArgumentException("Created stream in write mode but base stream cannot write.");
_stream = stream;
_appendChecksumCalculation = appendChecksumCalculation;
if (_stream.CanSeek) {
_initialPosition = _stream.Position;
_nextToBeChecksummedPosition = _stream.Position;
} else {
_initialPosition = -1;
_nextToBeChecksummedPosition = -1;
}
}
public override int Read(byte[] buffer, int offset, int count)
{
if (!CanSeek)
return ReadUnseekableInternal(buffer, offset, count, false, default(CancellationToken)).EnsureCompleted();
return ReadSeekableInternal(buffer, offset, count, false, default(CancellationToken)).EnsureCompleted();
}
[AsyncStateMachine(typeof(<ReadAsync>d__30))]
public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
<ReadAsync>d__30 stateMachine = default(<ReadAsync>d__30);
stateMachine.<>t__builder = AsyncTaskMethodBuilder<int>.Create();
stateMachine.<>4__this = this;
stateMachine.buffer = buffer;
stateMachine.offset = offset;
stateMachine.count = count;
stateMachine.cancellationToken = cancellationToken;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
[AsyncStateMachine(typeof(<ReadSeekableInternal>d__31))]
private Task<int> ReadSeekableInternal(byte[] buffer, int offset, int count, bool async, CancellationToken cancellationToken)
{
<ReadSeekableInternal>d__31 stateMachine = default(<ReadSeekableInternal>d__31);
stateMachine.<>t__builder = AsyncTaskMethodBuilder<int>.Create();
stateMachine.<>4__this = this;
stateMachine.buffer = buffer;
stateMachine.offset = offset;
stateMachine.count = count;
stateMachine.async = async;
stateMachine.cancellationToken = cancellationToken;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
[AsyncStateMachine(typeof(<ReadUnseekableInternal>d__32))]
private Task<int> ReadUnseekableInternal(byte[] buffer, int offset, int count, bool async, CancellationToken cancellationToken)
{
<ReadUnseekableInternal>d__32 stateMachine = default(<ReadUnseekableInternal>d__32);
stateMachine.<>t__builder = AsyncTaskMethodBuilder<int>.Create();
stateMachine.<>4__this = this;
stateMachine.buffer = buffer;
stateMachine.offset = offset;
stateMachine.count = count;
stateMachine.async = async;
stateMachine.cancellationToken = cancellationToken;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
public override void Write(byte[] buffer, int offset, int count)
{
WriteInternal(buffer, offset, count, false, default(CancellationToken)).EnsureCompleted();
}
[AsyncStateMachine(typeof(<WriteAsync>d__34))]
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
<WriteAsync>d__34 stateMachine = default(<WriteAsync>d__34);
stateMachine.<>t__builder = AsyncTaskMethodBuilder.Create();
stateMachine.<>4__this = this;
stateMachine.buffer = buffer;
stateMachine.offset = offset;
stateMachine.count = count;
stateMachine.cancellationToken = cancellationToken;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
[AsyncStateMachine(typeof(<WriteInternal>d__35))]
private Task WriteInternal(byte[] buffer, int offset, int count, bool async, CancellationToken cancellationToken)
{
<WriteInternal>d__35 stateMachine = default(<WriteInternal>d__35);
stateMachine.<>t__builder = AsyncTaskMethodBuilder.Create();
stateMachine.<>4__this = this;
stateMachine.buffer = buffer;
stateMachine.offset = offset;
stateMachine.count = count;
stateMachine.async = async;
stateMachine.cancellationToken = cancellationToken;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
public override long Seek(long offset, SeekOrigin origin)
{
if (CanWrite)
throw new NotSupportedException("No seek support in write mode.");
if (_stream.CanSeek) {
long num;
switch (origin) {
case SeekOrigin.Begin:
num = offset;
break;
case SeekOrigin.Current:
num = _stream.Position + offset;
break;
case SeekOrigin.End:
num = _stream.Length + offset;
break;
default:
throw new ArgumentException($"""{origin}""");
}
long num2 = num;
if (num2 < _initialPosition || _nextToBeChecksummedPosition < num2)
throw new NotSupportedException("Cannot seek past current checksum calculation point.");
}
return _stream.Seek(offset, origin);
}
public override void Flush()
{
_stream.Flush();
}
[AsyncStateMachine(typeof(<FlushAsync>d__38))]
public override Task FlushAsync(CancellationToken cancellationToken)
{
<FlushAsync>d__38 stateMachine = default(<FlushAsync>d__38);
stateMachine.<>t__builder = AsyncTaskMethodBuilder.Create();
stateMachine.<>4__this = this;
stateMachine.cancellationToken = cancellationToken;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
public override void Close()
{
_stream.Close();
}
public override void SetLength(long value)
{
_stream.SetLength(value);
}
private void AssertCanRead()
{
if (!CanRead)
throw new NotSupportedException("Cannot read from stream.");
}
private void AssertCanWrite()
{
if (!CanWrite)
throw new NotSupportedException("Cannot write to stream.");
}
}
}