crysome/Crysome.Client/Crysome.Client.Handlers/ReverseProxyStream.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;

namespace Crysome.Client.Handlers;

internal sealed class ReverseProxyStream : Stream
{
	private readonly Queue<byte[]> _chunks = new Queue<byte[]>();

	private byte[] _currentChunk;

	private int _currentOffset;

	private readonly object _lock = new object();

	private bool _closed;

	private readonly Action<byte[]> _sendToServer;

	public override bool CanRead => true;

	public override bool CanWrite => true;

	public override bool CanSeek => false;

	public override long Length
	{
		get
		{
			throw new NotSupportedException();
		}
	}

	public override long Position
	{
		get
		{
			throw new NotSupportedException();
		}
		set
		{
			throw new NotSupportedException();
		}
	}

	public ReverseProxyStream(Action<byte[]> sendToServer)
	{
		_sendToServer = sendToServer ?? throw new ArgumentNullException("sendToServer");
	}

	public void Push(byte[] data)
	{
		if (data == null || data.Length == 0)
		{
			return;
		}
		lock (_lock)
		{
			if (!_closed)
			{
				_chunks.Enqueue(data);
				Monitor.Pulse(_lock);
			}
		}
	}

	public void CloseFromRemote()
	{
		lock (_lock)
		{
			_closed = true;
			Monitor.PulseAll(_lock);
		}
	}

	public override int Read(byte[] buffer, int offset, int count)
	{
		if (buffer == null || offset < 0 || count <= 0)
		{
			return 0;
		}
		lock (_lock)
		{
			int num = 0;
			while (num < count)
			{
				if (_currentChunk != null && _currentOffset < _currentChunk.Length)
				{
					int num2 = Math.Min(count - num, _currentChunk.Length - _currentOffset);
					Array.Copy(_currentChunk, _currentOffset, buffer, offset + num, num2);
					_currentOffset += num2;
					num += num2;
					if (_currentOffset >= _currentChunk.Length)
					{
						_currentChunk = null;
						_currentOffset = 0;
					}
					if (num > 0)
					{
						return num;
					}
				}
				if (_chunks.Count > 0)
				{
					_currentChunk = _chunks.Dequeue();
					_currentOffset = 0;
					continue;
				}
				if (_closed)
				{
					return (num > 0) ? num : 0;
				}
				Monitor.Wait(_lock);
			}
			return num;
		}
	}

	public override void Write(byte[] buffer, int offset, int count)
	{
		if (buffer != null && count > 0 && offset >= 0 && offset + count <= buffer.Length)
		{
			byte[] array = new byte[count];
			Array.Copy(buffer, offset, array, 0, count);
			_sendToServer(array);
		}
	}

	public override long Seek(long offset, SeekOrigin origin)
	{
		throw new NotSupportedException();
	}

	public override void SetLength(long value)
	{
		throw new NotSupportedException();
	}

	public override void Flush()
	{
	}
}