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
54using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;

namespace Xtremegaida.DataStructures;

public class DataQueueMemoryReader : IDataQueueReader
{
	private readonly ReadOnlyMemory<byte> memory;
	private int totalBytesRead;
	private volatile bool disposed;

	public long UnreadLength => memory.Length - totalBytesRead;
	public bool IsReadClosed => disposed || totalBytesRead >= memory.Length;
	public long TotalBytesRead => totalBytesRead;

	public DataQueueMemoryReader(ReadOnlyMemory<byte> memory)
	{
		this.memory = memory;
	}

	public ValueTask<int> ReadAsync(Memory<byte> buffer, bool waitUntilFull = false, CancellationToken cancellationToken = default)
	{
		if (disposed) { return ValueTask.FromResult(0); }
		var canRead = memory.Length - totalBytesRead;
		if (canRead > buffer.Length) { canRead = buffer.Length; }
		memory.Slice(totalBytesRead, canRead).CopyTo(buffer);
		Interlocked.Add(ref totalBytesRead, canRead);
		return ValueTask.FromResult(canRead);
	}

	public ValueTask<int> ReadAsync(int skipBytes, CancellationToken cancellationToken = default)
	{
		if (disposed) { return ValueTask.FromResult(0); }
		var canRead = memory.Length - totalBytesRead;
		if (canRead > skipBytes) { canRead = skipBytes; }
		Interlocked.Add(ref totalBytesRead, canRead);
		return ValueTask.FromResult(canRead);
	}

	public ValueTask<int> ReadByteAsync(CancellationToken cancellationToken = default)
	{
		if (disposed || totalBytesRead >= memory.Length) { return ValueTask.FromResult(-1); }
		var result = memory.Span[totalBytesRead];
		Interlocked.Increment(ref totalBytesRead);
		return ValueTask.FromResult((int)result);
	}

	public void Dispose()
	{
		disposed = true;
	}
}