lancommander-ref/SharpCompress
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix up rewindable stream and use it more, add NonDisposingStream

Browse source
This commit is contained in:
Adam Hathcock 2026-01-29 09:08:40 +00:00
parent 58459bda12
commit e2cb9f39ab
12 changed files with 529 additions and 86 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/SharpCompress/Common/Volume.cs
View file

@ -18,7 +18,7 @@ public abstract partial class Volume : IVolume, IAsyncDisposable
_baseStream = stream;
if (ReaderOptions.LeaveStreamOpen)
{
stream = SharpCompressStream.Create(stream, leaveOpen: true);
stream = new NonDisposingStream(stream);
}
if (stream is IStreamStack ss)

2
src/SharpCompress/Factories/Factory.cs
View file

@ -87,7 +87,7 @@ public abstract class Factory : IFactory
stream.Rewind();
if (IsArchive(stream, options.Password))
{
stream.Rewind();
stream.StopRecording();
reader = readerFactory.OpenReader(stream, options);
return true;
}

4
src/SharpCompress/Factories/GZipFactory.cs
View file

@ -120,11 +120,11 @@ public class GZipFactory
var testStream = new GZipStream(rewindableStream, CompressionMode.Decompress);
if (TarArchive.IsTarFile(testStream))
{
rewindableStream.Rewind();
rewindableStream.StopRecording();
reader = new TarReader(rewindableStream, options, CompressionType.GZip);
return true;
}
rewindableStream.Rewind();
rewindableStream.StopRecording();
reader = OpenReader(rewindableStream, options);
return true;
}

13
src/SharpCompress/Factories/TarFactory.cs
View file

@ -168,7 +168,7 @@ public class TarFactory
var decompressedStream = wrapper.CreateStream(rewindableStream);
if (TarArchive.IsTarFile(decompressedStream))
{
rewindableStream.Rewind();
rewindableStream.StopRecording();
return new TarReader(rewindableStream, options, wrapper.CompressionType);
}
}
@ -185,18 +185,19 @@ public class TarFactory
{
cancellationToken.ThrowIfCancellationRequested();
options ??= new ReaderOptions();
var rewindableStream = new SharpCompressStream(stream);
var pos = rewindableStream.GetPosition();
var rewindableStream = new RewindableStream(stream);
rewindableStream.StartRecording();
foreach (var wrapper in TarWrapper.Wrappers)
{
rewindableStream.StackSeek(pos);
rewindableStream.Rewind();
if (await wrapper.IsMatchAsync(rewindableStream, cancellationToken))
{
rewindableStream.StackSeek(pos);
rewindableStream.Rewind();
var decompressedStream = wrapper.CreateStream(rewindableStream);
if (await TarArchive.IsTarFileAsync(decompressedStream, cancellationToken))
{
rewindableStream.StackSeek(pos);
rewindableStream.Rewind();
rewindableStream.StopRecording();
return new TarReader(rewindableStream, options, wrapper.CompressionType);
}
}

216
src/SharpCompress/IO/NonDisposingStream.cs Normal file
View file

@ -0,0 +1,216 @@
using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
namespace SharpCompress.IO;
/// <summary>
/// A stream wrapper that prevents disposal of the underlying stream.
/// This is useful when working with compression streams directly and you want
/// to keep the base stream open after the compression stream is disposed.
/// </summary>
internal class NonDisposingStream : Stream
{
private readonly Stream _stream;
private bool _isDisposed;
/// <summary>
/// Gets or sets a value indicating whether to throw an exception when the stream is disposed.
/// This is useful for testing to ensure streams are not disposed prematurely.
/// </summary>
public bool ThrowOnDispose { get; set; }
/// <summary>
/// Initializes a new instance of the <see cref="NonDisposingStream"/> class.
/// </summary>
/// <param name="stream">The stream to wrap. This stream will NOT be disposed when this wrapper is disposed.</param>
/// <exception cref="ArgumentNullException">Thrown when <paramref name="stream"/> is null.</exception>
public NonDisposingStream(Stream stream)
{
_stream = stream ?? throw new ArgumentNullException(nameof(stream));
}
public override bool CanRead => !_isDisposed && _stream.CanRead;
public override bool CanSeek => !_isDisposed && _stream.CanSeek;
public override bool CanWrite => !_isDisposed && _stream.CanWrite;
public override long Length
{
get
{
ThrowIfDisposed();
return _stream.Length;
}
}
public override long Position
{
get
{
ThrowIfDisposed();
return _stream.Position;
}
set
{
ThrowIfDisposed();
_stream.Position = value;
}
}
public override void Flush()
{
ThrowIfDisposed();
_stream.Flush();
}
public override int Read(byte[] buffer, int offset, int count)
{
ThrowIfDisposed();
return _stream.Read(buffer, offset, count);
}
#if !LEGACY_DOTNET
public override int Read(Span<byte> buffer)
{
ThrowIfDisposed();
return _stream.Read(buffer);
}
#endif
public override long Seek(long offset, SeekOrigin origin)
{
ThrowIfDisposed();
return _stream.Seek(offset, origin);
}
public override void SetLength(long value)
{
ThrowIfDisposed();
_stream.SetLength(value);
}
public override void Write(byte[] buffer, int offset, int count)
{
ThrowIfDisposed();
_stream.Write(buffer, offset, count);
}
#if !LEGACY_DOTNET
public override void Write(ReadOnlySpan<byte> buffer)
{
ThrowIfDisposed();
_stream.Write(buffer);
}
#endif
public override Task<int> ReadAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
)
{
ThrowIfDisposed();
return _stream.ReadAsync(buffer, offset, count, cancellationToken);
}
#if !LEGACY_DOTNET
public override ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default
)
{
ThrowIfDisposed();
return _stream.ReadAsync(buffer, cancellationToken);
}
#endif
public override Task WriteAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
)
{
ThrowIfDisposed();
return _stream.WriteAsync(buffer, offset, count, cancellationToken);
}
#if !LEGACY_DOTNET
public override ValueTask WriteAsync(
ReadOnlyMemory<byte> buffer,
CancellationToken cancellationToken = default
)
{
ThrowIfDisposed();
return _stream.WriteAsync(buffer, cancellationToken);
}
#endif
public override Task FlushAsync(CancellationToken cancellationToken)
{
ThrowIfDisposed();
return _stream.FlushAsync(cancellationToken);
}
public override Task CopyToAsync(
Stream destination,
int bufferSize,
CancellationToken cancellationToken
)
{
ThrowIfDisposed();
return _stream.CopyToAsync(destination, bufferSize, cancellationToken);
}
/// <summary>
/// Disposes this wrapper without disposing the underlying stream.
/// </summary>
protected override void Dispose(bool disposing)
{
if (!_isDisposed)
{
if (ThrowOnDispose)
{
throw new InvalidOperationException(
$"Attempt to dispose of a {nameof(NonDisposingStream)} when {nameof(ThrowOnDispose)} is true"
);
}
_isDisposed = true;
// Intentionally do NOT dispose _stream
}
base.Dispose(disposing);
}
#if !LEGACY_DOTNET
/// <summary>
/// Asynchronously disposes this wrapper without disposing the underlying stream.
/// </summary>
public override async ValueTask DisposeAsync()
{
if (!_isDisposed)
{
if (ThrowOnDispose)
{
throw new InvalidOperationException(
$"Attempt to dispose of a {nameof(NonDisposingStream)} when {nameof(ThrowOnDispose)} is true"
);
}
_isDisposed = true;
// Intentionally do NOT dispose _stream
}
await base.DisposeAsync();
}
#endif
private void ThrowIfDisposed()
{
if (_isDisposed)
{
throw new ObjectDisposedException(nameof(NonDisposingStream));
}
}
}

7
src/SharpCompress/IO/RewindableStream.cs
View file

@ -84,6 +84,13 @@ namespace SharpCompress.IO
IsRecording = true;
}
public void StopRecording()
{
_isRewound = true;
IsRecording = false;
_bufferPosition = 0;
}
public override bool CanRead => true;
public override bool CanSeek => stream.CanSeek;

16
src/SharpCompress/Readers/ReaderFactory.Async.cs
View file

@ -54,8 +54,8 @@ public static partial class ReaderFactory
stream.NotNull(nameof(stream));
options ??= new ReaderOptions() { LeaveStreamOpen = false };
var bStream = new SharpCompressStream(stream, bufferSize: options.BufferSize);
long pos = bStream.GetPosition();
var bStream = new RewindableStream(stream);
bStream.StartRecording();
var factories = Factory.Factories.OfType<Factory>();
@ -68,7 +68,7 @@ public static partial class ReaderFactory
);
if (testedFactory is IReaderFactory readerFactory)
{
bStream.StackSeek(pos);
bStream.Rewind();
if (
await testedFactory.IsArchiveAsync(
bStream,
@ -76,11 +76,12 @@ public static partial class ReaderFactory
)
)
{
bStream.StackSeek(pos);
bStream.Rewind();
bStream.StopRecording();
return await readerFactory.OpenAsyncReader(bStream, options, cancellationToken);
}
}
bStream.StackSeek(pos);
bStream.Rewind();
}
foreach (var factory in factories)
@ -89,13 +90,14 @@ public static partial class ReaderFactory
{
continue; // Already tested above
}
bStream.StackSeek(pos);
bStream.Rewind();
if (
factory is IReaderFactory readerFactory
&& await factory.IsArchiveAsync(bStream, cancellationToken: cancellationToken)
)
{
bStream.StackSeek(pos);
bStream.Rewind();
bStream.StopRecording();
return await readerFactory.OpenAsyncReader(bStream, options, cancellationToken);
}
}

2
src/SharpCompress/Readers/ReaderFactory.cs
View file

@ -52,7 +52,7 @@ public static partial class ReaderFactory
&& reader != null
)
{
bStream.Rewind();
bStream.StopRecording();
return reader;
}
}

176
tests/SharpCompress.Test/Mocks/ForwardOnlyStream.cs
View file

@ -2,113 +2,163 @@ using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using SharpCompress.Common;
using SharpCompress.IO;
namespace SharpCompress.Test.Mocks;
public class ForwardOnlyStream : SharpCompressStream, IStreamStack
/// <summary>
/// A forward-only stream wrapper that delegates directly to the underlying stream
/// without any buffering. Supports reading and writing but not seeking.
/// </summary>
public class ForwardOnlyStream : Stream
{
private readonly Stream stream;
#if DEBUG_STREAMS
long IStreamStack.InstanceId { get; set; }
#endif
Stream IStreamStack.BaseStream() => stream;
int IStreamStack.BufferSize
{
get => 0;
set { }
}
int IStreamStack.BufferPosition
{
get => 0;
set { }
}
void IStreamStack.SetPosition(long position) { }
public bool IsDisposed { get; private set; }
private readonly Stream _stream;
private bool _isDisposed;
/// <summary>
/// Initializes a new instance of the <see cref="ForwardOnlyStream"/> class.
/// </summary>
/// <param name="stream">The underlying stream to wrap.</param>
/// <param name="bufferSize">Buffer size parameter (ignored - this implementation does not buffer).</param>
/// <exception cref="ArgumentNullException">Thrown when <paramref name="stream"/> is null.</exception>
public ForwardOnlyStream(Stream stream, int? bufferSize = null)
: base(stream, bufferSize: bufferSize ?? Constants.BufferSize)
{
this.stream = stream;
#if DEBUG_STREAMS
this.DebugConstruct(typeof(ForwardOnlyStream));
#endif
_stream = stream ?? throw new ArgumentNullException(nameof(stream));
// bufferSize is ignored - this implementation does not buffer
}
protected override void Dispose(bool disposing)
{
if (!IsDisposed)
{
if (disposing)
{
#if DEBUG_STREAMS
this.DebugDispose(typeof(ForwardOnlyStream));
#endif
stream.Dispose();
IsDisposed = true;
base.Dispose(disposing);
}
}
}
public override bool CanRead => true;
public override bool CanRead => !_isDisposed && _stream.CanRead;
public override bool CanSeek => false;
public override bool CanWrite => true;
public override void Flush() { }
public override bool CanWrite => !_isDisposed && _stream.CanWrite;
public override long Length => throw new NotSupportedException();
public override long Length
{
get => throw new NotSupportedException("Length is not supported on a forward-only stream.");
}
public override long Position
{
get => stream.Position;
set => throw new NotSupportedException();
get =>
throw new NotSupportedException("Position is not supported on a forward-only stream.");
set =>
throw new NotSupportedException("Position is not supported on a forward-only stream.");
}
public override int Read(byte[] buffer, int offset, int count) =>
stream.Read(buffer, offset, count);
public override void Flush()
{
ThrowIfDisposed();
_stream.Flush();
}
public override int Read(byte[] buffer, int offset, int count)
{
ThrowIfDisposed();
return _stream.Read(buffer, offset, count);
}
public override Task<int> ReadAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
) => stream.ReadAsync(buffer, offset, count, cancellationToken);
)
{
ThrowIfDisposed();
return _stream.ReadAsync(buffer, offset, count, cancellationToken);
}
#if !LEGACY_DOTNET
public override ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default
) => stream.ReadAsync(buffer, cancellationToken);
)
{
ThrowIfDisposed();
return _stream.ReadAsync(buffer, cancellationToken);
}
#endif
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
public override long Seek(long offset, SeekOrigin origin) =>
throw new NotSupportedException("Seek is not supported on a forward-only stream.");
public override void SetLength(long value) => throw new NotSupportedException();
public override void SetLength(long value) =>
throw new NotSupportedException("SetLength is not supported on a forward-only stream.");
public override void Write(byte[] buffer, int offset, int count) =>
stream.Write(buffer, offset, count);
public override void Write(byte[] buffer, int offset, int count)
{
ThrowIfDisposed();
_stream.Write(buffer, offset, count);
}
public override Task WriteAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
) => stream.WriteAsync(buffer, offset, count, cancellationToken);
)
{
ThrowIfDisposed();
return _stream.WriteAsync(buffer, offset, count, cancellationToken);
}
#if !LEGACY_DOTNET
public override ValueTask WriteAsync(
ReadOnlyMemory<byte> buffer,
CancellationToken cancellationToken = default
) => stream.WriteAsync(buffer, cancellationToken);
)
{
ThrowIfDisposed();
return _stream.WriteAsync(buffer, cancellationToken);
}
#endif
public override Task FlushAsync(CancellationToken cancellationToken) =>
stream.FlushAsync(cancellationToken);
public override Task FlushAsync(CancellationToken cancellationToken)
{
ThrowIfDisposed();
return _stream.FlushAsync(cancellationToken);
}
public override Task CopyToAsync(
Stream destination,
int bufferSize,
CancellationToken cancellationToken
)
{
ThrowIfDisposed();
return _stream.CopyToAsync(destination, bufferSize, cancellationToken);
}
protected override void Dispose(bool disposing)
{
if (!_isDisposed)
{
if (disposing)
{
_stream.Dispose();
}
_isDisposed = true;
base.Dispose(disposing);
}
}
#if !LEGACY_DOTNET
public override async ValueTask DisposeAsync()
{
if (!_isDisposed)
{
await _stream.DisposeAsync();
_isDisposed = true;
}
await base.DisposeAsync();
}
#endif
private void ThrowIfDisposed()
{
if (_isDisposed)
{
throw new ObjectDisposedException(nameof(ForwardOnlyStream));
}
}
}

7
tests/SharpCompress.Test/ReaderTests.cs
View file

@ -164,11 +164,8 @@ public abstract class ReaderTests : TestBase
using var file = File.OpenRead(testArchive);
#if !LEGACY_DOTNET
await using var protectedStream = SharpCompressStream.Create(
new ForwardOnlyStream(file, options.BufferSize),
leaveOpen: true,
throwOnDispose: true,
bufferSize: options.BufferSize
await using var protectedStream = new NonDisposingStream(
new ForwardOnlyStream(file, options.BufferSize)
);
await using var testStream = new TestStream(protectedStream);
#else

67
tests/SharpCompress.Test/Streams/RewindableStreamAsyncTest.cs
View file

@ -1,6 +1,7 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using SharpCompress.IO;
@ -398,4 +399,70 @@ public class RewindableStreamAsyncTest
return result.ToArray();
}
#endif
[Fact]
public async ValueTask TestStopRecordingAsync()
{
var ms = new MemoryStream();
var bw = new BinaryWriter(ms);
bw.Write(1);
bw.Write(2);
bw.Write(3);
bw.Write(4);
bw.Write(5);
bw.Write(6);
bw.Write(7);
bw.Flush();
ms.Position = 0;
var stream = new RewindableStream(ms);
stream.StartRecording();
Assert.Equal(1, await ReadInt32Async(stream).ConfigureAwait(false));
Assert.Equal(2, await ReadInt32Async(stream).ConfigureAwait(false));
Assert.Equal(3, await ReadInt32Async(stream).ConfigureAwait(false));
Assert.Equal(4, await ReadInt32Async(stream).ConfigureAwait(false));
stream.StopRecording();
Assert.Equal(1, await ReadInt32Async(stream).ConfigureAwait(false));
Assert.Equal(2, await ReadInt32Async(stream).ConfigureAwait(false));
Assert.Equal(3, await ReadInt32Async(stream).ConfigureAwait(false));
Assert.Equal(4, await ReadInt32Async(stream).ConfigureAwait(false));
Assert.Equal(5, await ReadInt32Async(stream).ConfigureAwait(false));
Assert.Equal(6, await ReadInt32Async(stream).ConfigureAwait(false));
Assert.Equal(7, await ReadInt32Async(stream).ConfigureAwait(false));
Assert.False(stream.IsRecording);
}
[Fact]
public async ValueTask TestStopRecordingNoFurtherBufferingAsync()
{
var ms = new MemoryStream();
var bw = new BinaryWriter(ms);
bw.Write(1);
bw.Write(2);
bw.Write(3);
bw.Write(4);
bw.Flush();
ms.Position = 0;
var stream = new RewindableStream(ms);
stream.StartRecording();
var buffer = new byte[8];
await stream.ReadAsync(buffer, 0, 8).ConfigureAwait(false);
stream.StopRecording();
await stream.ReadAsync(buffer, 0, 8).ConfigureAwait(false);
Assert.Equal(BitConverter.GetBytes(1), buffer.Take(4).ToArray());
Assert.Equal(BitConverter.GetBytes(2), buffer.Skip(4).Take(4).ToArray());
int bytesRead = await stream.ReadAsync(buffer, 0, 8).ConfigureAwait(false);
Assert.Equal(8, bytesRead);
Assert.False(stream.IsRecording);
}
}

103
tests/SharpCompress.Test/Streams/RewindableStreamTest.cs
View file

@ -1,5 +1,6 @@
using System;
using System.IO;
using System.Linq;
using SharpCompress.IO;
using Xunit;
@ -194,4 +195,106 @@ public class RewindableStreamTest
stream.Dispose();
Assert.Throws<ObjectDisposedException>(() => stream.Read(new byte[4], 0, 4));
}
[Fact]
public void TestStopRecordingBasic()
{
var ms = new MemoryStream();
var bw = new BinaryWriter(ms);
bw.Write(1);
bw.Write(2);
bw.Write(3);
bw.Write(4);
bw.Write(5);
bw.Write(6);
bw.Write(7);
bw.Flush();
ms.Position = 0;
var stream = new RewindableStream(ms);
stream.StartRecording();
var br = new BinaryReader(stream);
Assert.Equal(1, br.ReadInt32());
Assert.Equal(2, br.ReadInt32());
Assert.Equal(3, br.ReadInt32());
Assert.Equal(4, br.ReadInt32());
stream.StopRecording();
Assert.Equal(1, br.ReadInt32());
Assert.Equal(2, br.ReadInt32());
Assert.Equal(3, br.ReadInt32());
Assert.Equal(4, br.ReadInt32());
Assert.Equal(5, br.ReadInt32());
Assert.Equal(6, br.ReadInt32());
Assert.Equal(7, br.ReadInt32());
Assert.False(stream.IsRecording);
}
[Fact]
public void TestStopRecordingNoFurtherBuffering()
{
var ms = new MemoryStream();
var bw = new BinaryWriter(ms);
bw.Write(1);
bw.Write(2);
bw.Write(3);
bw.Write(4);
bw.Flush();
ms.Position = 0;
var stream = new RewindableStream(ms);
stream.StartRecording();
var buffer = new byte[8];
stream.Read(buffer, 0, 8);
stream.StopRecording();
stream.Read(buffer, 0, 8);
Assert.Equal(BitConverter.GetBytes(1), buffer.Take(4).ToArray());
Assert.Equal(BitConverter.GetBytes(2), buffer.Skip(4).Take(4).ToArray());
int bytesRead = stream.Read(buffer, 0, 8);
Assert.Equal(8, bytesRead);
Assert.False(stream.IsRecording);
bytesRead = stream.Read(buffer, 0, 8);
Assert.Equal(0, bytesRead);
}
#if !LEGACY_DOTNET
[Fact]
public void TestStopRecordingWithSpan()
{
var ms = new MemoryStream();
var bw = new BinaryWriter(ms);
bw.Write(1);
bw.Write(2);
bw.Write(3);
bw.Write(4);
bw.Flush();
ms.Position = 0;
var stream = new RewindableStream(ms);
stream.StartRecording();
var buffer = new byte[8];
stream.Read(buffer);
stream.StopRecording();
stream.Read(buffer);
Assert.Equal(BitConverter.GetBytes(1), buffer.Take(4).ToArray());
Assert.Equal(BitConverter.GetBytes(2), buffer.Skip(4).Take(4).ToArray());
int bytesRead = stream.Read(buffer);
Assert.Equal(8, bytesRead);
Assert.Equal(BitConverter.GetBytes(3), buffer.Take(4).ToArray());
Assert.Equal(BitConverter.GetBytes(4), buffer.Skip(4).Take(4).ToArray());
}
#endif
}