1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;

namespace Xtremegaida.DataStructures;

public class DataQueueStreamReader : IDataQueueReader, IAsyncDisposable
{
	private readonly Stream baseStream;
	private readonly bool keepStreamOpen;
	private readonly long maxReadBytes = -1;
	private volatile bool disposed;
	private long totalBytesRead;
	private byte[] byteRead;

	public long UnreadLength
	{
		get
		{
			var canRead = !baseStream.CanSeek ? (disposed ? 0 : long.MaxValue) : (baseStream.Length - baseStream.Position);
			if (maxReadBytes >= 0)
			{
				var maxRead = maxReadBytes - totalBytesRead;
				if (maxRead < canRead) { canRead = maxRead; }
			}
			return canRead;
		}
	}

	public bool IsReadClosed => disposed;

	public long TotalBytesRead => totalBytesRead;

	public DataQueueStreamReader(Stream baseStream, bool keepStreamOpen = false)
	{
		this.baseStream = baseStream;
		this.keepStreamOpen = keepStreamOpen;
	}

	public DataQueueStreamReader(Stream baseStream, long maxReadBytes, bool keepStreamOpen = false) : this(baseStream, keepStreamOpen)
	{
		if (maxReadBytes < 0) { throw new ArgumentOutOfRangeException(nameof(maxReadBytes)); }
		this.maxReadBytes = maxReadBytes;
	}

	public async ValueTask<int> ReadAsync(Memory<byte> buffer, bool waitUntilFull = false, CancellationToken cancellationToken = default)
	{
		int readBytes = 0;
		if (disposed) { return 0; }
		if (maxReadBytes >= 0)
		{
			var canRead = maxReadBytes - totalBytesRead;
			if (canRead < buffer.Length) { buffer = buffer.Slice(0, (int)canRead); }
		}
		if (!waitUntilFull)
		{
			readBytes = await baseStream.ReadAsync(buffer, cancellationToken);
		}
		else
		{
			while (!buffer.IsEmpty)
			{
				var read = await baseStream.ReadAsync(buffer, cancellationToken);
				if (read <= 0) { break; }
				buffer = buffer.Slice(read);
				readBytes += read;
			}
			;
		}
		Interlocked.Add(ref totalBytesRead, readBytes);
		return readBytes;
	}

	public async ValueTask<int> ReadAsync(int skipBytes, CancellationToken cancellationToken = default)
	{
		if (disposed) { return 0; }
		if (maxReadBytes >= 0)
		{
			var canRead = maxReadBytes - totalBytesRead;
			if (canRead < skipBytes) { skipBytes = (int)canRead; }
		}
		if (skipBytes <= 0) { return 0; }
		if (baseStream.CanSeek)
		{
			var length = baseStream.Length;
			var pos = baseStream.Position;
			var canRead = length - pos;
			if (canRead > skipBytes) { canRead = skipBytes; }
			if (canRead <= 0) { canRead = 0; }
			else { baseStream.Seek(canRead, SeekOrigin.Current); }
			Interlocked.Add(ref totalBytesRead, canRead);
			return (int)canRead;
		}
		int readBytes = 0;
		if (skipBytes <= 4)
		{
			while (readBytes < skipBytes)
			{
				if (await ReadByteAsync(cancellationToken) < 0) { break; }
				readBytes++;
			}
		}
		else
		{
			var buffer = new byte[Math.Min(skipBytes, 4096)];
			while (readBytes < skipBytes)
			{
				var canRead = skipBytes - readBytes;
				if (canRead > buffer.Length) { canRead = buffer.Length; }
				var read = await baseStream.ReadAsync(buffer, 0, canRead, cancellationToken);
				if (read <= 0) { break; }
				readBytes += read;
			}
			;
			Interlocked.Add(ref totalBytesRead, readBytes);
		}
		return readBytes;
	}

	public async ValueTask<int> ReadByteAsync(CancellationToken cancellationToken = default)
	{
		if (disposed || (maxReadBytes >= 0 && totalBytesRead >= maxReadBytes)) { return -1; }
		if (byteRead == null) { byteRead = new byte[1]; }
		var read = await baseStream.ReadAsync(byteRead, cancellationToken);
		if (read == 0) { return -1; }
		Interlocked.Increment(ref totalBytesRead);
		return byteRead[0];
	}

	public void Dispose()
	{
		if (disposed) { return; }
		disposed = true;
		if (!keepStreamOpen) { baseStream.Dispose(); }
	}

	public async ValueTask DisposeAsync()
	{
		if (disposed) { return; }
		disposed = true;
		if (!keepStreamOpen) { await baseStream.DisposeAsync(); }
	}
}