Skip to content

ProgressStream/ThrottledStream Dont Dispose Inner Stream (L7 Fix) #390

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Shivansps merged 1 commit into from
Apr 12, 2026
Merged

ProgressStream/ThrottledStream Dont Dispose Inner Stream (L7 Fix) #390

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 26 additions & 26 deletions Knossos.NET/Classes/ProgressStream.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,57 +9,57 @@ namespace Knossos.NET.Classes
/// </summary>
public class ProgressStream : Stream
{
private Stream stream;
private Action<int>? progressCallback;

private readonly Stream _stream;
private readonly Action<int>? _progressCallback;

public ProgressStream(Stream stream, Action<int>? progressCallback)
{
this.stream = stream;
this.progressCallback = progressCallback;
_stream = stream ?? throw new ArgumentNullException(nameof(stream));
_progressCallback = progressCallback;
}

public override bool CanRead => stream.CanRead;

public override bool CanSeek => stream.CanSeek;

public override bool CanWrite => stream.CanWrite;

public override long Length => stream.Length;

public override long Position { get => stream.Position; set => stream.Position = value; }

public override void Flush()
public override bool CanRead => _stream.CanRead;
public override bool CanSeek => _stream.CanSeek;
public override bool CanWrite => _stream.CanWrite;
public override long Length => _stream.Length;
public override long Position
{
stream.Flush();
get => _stream.Position;
set => _stream.Position = value;
}

public override void Flush() => _stream.Flush();

public override int Read(byte[] buffer, int offset, int count)
{
int n = stream.Read(buffer, offset, count);
progressCallback?.Invoke((int)((100 * stream.Position) / stream.Length));
int n = _stream.Read(buffer, offset, count);
_progressCallback?.Invoke((int)((100 * _stream.Position) / _stream.Length));
return n;
}

public override long Seek(long offset, SeekOrigin origin)
{
stream.Seek(offset, origin);
_stream.Seek(offset, origin);
return Position;
}

public override void SetLength(long value)
{
stream.SetLength(value);
}
public override void SetLength(long value) => _stream.SetLength(value);

public override void Write(byte[] buffer, int offset, int count)
=> _stream.Write(buffer, offset, count);

protected override void Dispose(bool disposing)
{
stream.Write(buffer, offset, count);
if (disposing)
{
_stream.Dispose();
}
base.Dispose(disposing);
}

public override void Close()
{
stream.Close();
_stream.Close();
base.Close();
}
}
Expand Down
140 changes: 43 additions & 97 deletions Knossos.NET/Classes/ThrottledStream.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
using System.Threading;
using System.Threading.Tasks;


namespace Knossos.NET.Classes
{
/// <summary>
Expand All @@ -12,150 +11,97 @@ namespace Knossos.NET.Classes
/// </summary>
public class ThrottledStream : Stream
{
private Stream stream;
private long maxBytesPerSecond = 0; //0 is disabled
private long bytes = 0;
private long time;
private readonly Stream _stream;
private long _maxBytesPerSecond;
private long _bytes;
private long _time;

public ThrottledStream(Stream stream, long maxBytesPerSecond)
{
if (stream == null || maxBytesPerSecond < 0)
{
throw new Exception("Base stream cant be null and maxBytesPerSecond needs to be 0 or higher.");
}
_stream = stream ?? throw new ArgumentNullException(nameof(stream));
if (maxBytesPerSecond < 0)
throw new ArgumentOutOfRangeException(nameof(maxBytesPerSecond));

this.stream = stream;
this.maxBytesPerSecond = maxBytesPerSecond;
time = Environment.TickCount; //Register the starting time
_maxBytesPerSecond = maxBytesPerSecond;
_time = Environment.TickCount;
}

private void Limit(int inBytes)
{
//If we havent read anything from the stream or max bytes per second is 0 (disabled), just return
if (inBytes <= 0 || maxBytesPerSecond <= 0 )
{
if (inBytes <= 0 || _maxBytesPerSecond <= 0)
return;
}

bytes += inBytes;
var timePassed = Environment.TickCount - time;
_bytes += inBytes;
var timePassed = Environment.TickCount - _time;

if (timePassed > 0)
{
if ((bytes * 1000L / timePassed) > maxBytesPerSecond)
if ((_bytes * 1000L / timePassed) > _maxBytesPerSecond)
{
var wait = (int)((bytes * 1000L / maxBytesPerSecond) - timePassed);
var wait = (int)((_bytes * 1000L / _maxBytesPerSecond) - timePassed);
if (wait > 1)
{
try
{
Thread.Sleep(wait);
}
catch { }
try { Thread.Sleep(wait); } catch { }
timePassed += wait;
}
}

if (timePassed >= 1000)
{
time = Environment.TickCount;
bytes = 0;
_time = Environment.TickCount;
_bytes = 0;
}
}
}

/* Possibility to change the Max Bytes per Second on the fly */
public void SetMaxBytesPerSecond(long maxBytesPerSecond)
{
if (maxBytesPerSecond != 0 && this.maxBytesPerSecond != maxBytesPerSecond)
if (maxBytesPerSecond != 0 && _maxBytesPerSecond != maxBytesPerSecond)
{
this.maxBytesPerSecond = maxBytesPerSecond;
time = Environment.TickCount;
bytes = 0;
_maxBytesPerSecond = maxBytesPerSecond;
_time = Environment.TickCount;
_bytes = 0;
}
}

/* Needed Stream Override Implementations */

public override long Position
{
get
{
return stream.Position;
}
set
{
stream.Position = value;
}
}

public override long Length
{
get
{
return stream.Length;
}
}

public override bool CanSeek
{
get
{
return stream.CanSeek;
}
get => _stream.Position;
set => _stream.Position = value;
}

public override bool CanRead
{
get
{
return stream.CanRead;
}
}

public override bool CanWrite
{
get
{
return stream.CanWrite;
}
}

public override string ToString()
{
return stream.ToString()!;
}

public override void Flush()
{
stream.Flush();
}

public override void SetLength(long value)
{
stream.SetLength(value);
}
public override long Length => _stream.Length;
public override bool CanSeek => _stream.CanSeek;
public override bool CanRead => _stream.CanRead;
public override bool CanWrite => _stream.CanWrite;

public override void Flush() => _stream.Flush();
public override void SetLength(long value) => _stream.SetLength(value);
public override void Write(byte[] buffer, int offset, int count)
{
stream.Write(buffer, offset, count);
}
=> _stream.Write(buffer, offset, count);

public override int Read(byte[] buffer, int offset, int count)
{
Limit(count);
return stream.Read(buffer, offset, count);
return _stream.Read(buffer, offset, count);
}

public override long Seek(long offset, SeekOrigin origin)
{
return stream.Seek(offset, origin);
}
=> _stream.Seek(offset, origin);

public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
{
Limit(buffer.Length);
return stream.ReadAsync(buffer, cancellationToken);
return _stream.ReadAsync(buffer, cancellationToken);
}

protected override void Dispose(bool disposing)
{
if (disposing)
{
_stream.Dispose();
}
base.Dispose(disposing);
}
}
}
}