<commit_before>﻿using System;
using System.IO;
using System.Threading.Tasks;
using System.Collections.Generic;
using System.Threading;

namespace Ceen.Httpd
{
	/// <summary>
	/// Helper class that reads a stream,
	/// but also provides a buffer to allow peek/read-ahead
	/// It contains a special method to read data until the first 
	/// CR LF CR LF entry (end of HTTP header), and leaves the rest of
	/// the stream untouched.
	/// </summary>
	public class BufferedStreamReader : Stream
	{
		/// <summary>
		/// The underlying network stream
		/// </summary>
		private Stream m_parent;
		/// <summary>
		/// The internal read-ahead buffer
		/// </summary>
		private byte[] m_buffer;
		/// <summary>
		/// The number of bytes in the internal buffer
		/// </summary>
		private int m_buffercount = 0;
		/// <summary>
		/// The offset in the internal buffer from where the data starts
		/// </summary>
		private int m_bufferoffset = 0;
		/// <summary>
		/// The CR ASCII value.
		/// </summary>
		private const byte CR = 13;
		/// <summary>
		/// The LF ASCII value.
		/// </summary>
		private const byte LF = 10;
		/// <summary>
		/// The number of bytes we are allowed to read
		/// </summary>
		private long m_remainingbytes = 0;
		/// <summary>
		/// The number of bytes read
		/// </summary>
		private long m_maxread = 0;

		/// <summary>
		/// Initializes a new instance of the <see cref="Ceen.Httpd.BufferedStreamReader"/> class.
		/// </summary>
		/// <param name="parent">The parent stream.</param>
		/// <param name="initialbuffersize">Intitial size of the read-ahead buffer.</param>
		public BufferedStreamReader(Stream parent, int initialbuffersize = 1024)
		{
			if (parent == null)
				throw new ArgumentNullException(nameof(parent));
			
			m_parent = parent;
			m_buffer = new byte[initialbuffersize];
		}

		/// <summary>
		/// Resets the number of bytes read.
		/// </summary>
		/// <param name="maxsize">The maximum number of bytes to allow reading</param>
		internal void ResetReadLength(long maxsize)
		{
			m_maxread = m_remainingbytes = maxsize;
		}


		/// <summary>
		/// Reads all lines until an empty line (i.e. it reads until CR LF CR LF).
		/// Each read line, except the empty line is send to the linehandler function
		/// </summary>
		/// <returns>An awaitable task.</returns>
		/// <param name="maxlinesize">The maximum size of a single line.</param>
		/// <param name="maxheadersize">The maximum size of the headers.</param>
		/// <param name="idletimeout">The maximum time to remain idle.</param>
		/// <param name="linehandler">Callback method that processes each line.</param>
		/// <param name="timeouttask">Awaitable task for stopping the request.</param>
		/// <param name="stoptask">Awaitable task for stopping the request.</param>
		public async Task ReadHeaders(int maxlinesize, int maxheadersize, TimeSpan idletimeout, Action<string> linehandler, Task timeouttask, Task stoptask)
		{
			var buf = new byte[maxlinesize];
			var bufoffset = 0;
			var bufsize = 0;
			var totalread = 0;
			var lastlookoffset = 0;

			// Repeat the parsing until we time out or get an empty line
			while (true)
			{
				if (totalread == maxheadersize)
					throw new HttpException(HttpStatusCode.RequestHeaderFieldsTooLarge);

                Task<int> rtask;
                Task rt;

                using (var cs = new CancellationTokenSource(idletimeout))
                {
                    rtask = ReadAsync(buf, bufoffset, Math.Min(buf.Length - bufoffset, maxheadersize - totalread), cs.Token);
                    rt = await Task.WhenAny( stoptask, timeouttask, rtask);
                }

                // Timeout or stop has happened
				if (rt != rtask)
				{
					if (rt == stoptask)
						throw new TaskCanceledException();
					else if (totalread == 0)
						throw new EmptyStreamClosedException();
					else
						throw new HttpException(HttpStatusCode.RequestTimeout); 
				}

				var r = rtask.Result;
				totalread += r;
				bufsize += r;

				// Stream closed
				if (r == 0)
				{
					if (totalread == 0)
						throw new EmptyStreamClosedException();
					else
						throw new HttpException(HttpStatusCode.BadRequest);
				}

				// Look for first CR
				var ix = Array.IndexOf(buf, CR, lastlookoffset, bufsize - lastlookoffset);
				while (ix >= 0 && ix < bufsize - 1)
				{
					// Check if the next byte is LF
					if (buf[ix + 1] == LF)
					{
						var datalen = ix - bufoffset;

						// Check if the line is empty
						if (datalen == 0)
						{
							bufoffset = ix + 2;

							// If we have more data, "unread it"
							if (bufsize - bufoffset != 0)
								UnreadBytesIntoBuffer(buf, bufoffset, bufsize - bufoffset);

							return;
						}
						else
						{
							linehandler(System.Text.Encoding.ASCII.GetString(buf, bufoffset, datalen));
							bufoffset = ix + 2;
							lastlookoffset = bufoffset;
						}
					}
					else
					{
						// False hit, or char hit a split with blocks
						lastlookoffset = ix + 1;
					}

					// Get the next entry
					ix = Array.IndexOf(buf, CR, lastlookoffset, bufsize - lastlookoffset);
				}

				// Move trailing bytes to the start of the buffer to reduce space use
				// and make sure we do not search more than once for the CRLF
				bufsize = bufsize - bufoffset;
				if (bufsize != 0)
					Array.Copy(buf, bufoffset, buf, 0, bufsize);

				bufoffset = 0;
				lastlookoffset = Math.Max(0, bufsize - 1);
			}
		}

		/// <summary>
		/// Keeps reading from the stream until the supplied number of bytes are read
		/// </summary>
		/// <returns>The awaitable task.</returns>
		/// <param name="count">The number of bytes to read.</param>
		/// <param name="idletimeout">The time to wait while idle</param>
		/// <param name="timeouttask">The task that signals request timeout</param>
		/// <param name="stoptask">The task that signals stop for the server</param>
		internal async Task<byte[]> RepeatReadAsync(int count, TimeSpan idletimeout, Task timeouttask, Task stoptask)
		{
			var buf = new byte[count];
			await RepeatReadAsync(buf, 0, count, idletimeout, timeouttask, stoptask);
			return buf;
		}

		/// <summary>
		/// Gets the delimited sub stream.
		/// </summary>
		/// <returns>The delimited sub stream.</returns>
		/// <param name="delimiter">Delimiter.</param>
		/// <param name="idletime">The time to wait while idle</param><commit_msg>fix buffer overwrite<commit_after>﻿using System;
using System.IO;
using System.Threading.Tasks;
using System.Threading;

namespace Ceen.Httpd
{
	/// <summary>
	/// Helper class that reads a stream,
	/// but also provides a buffer to allow peek/read-ahead
	/// It contains a special method to read data until the first 
	/// CR LF CR LF entry (end of HTTP header), and leaves the rest of
	/// the stream untouched.
	/// </summary>
	public class BufferedStreamReader : Stream
	{
		/// <summary>
		/// The underlying network stream
		/// </summary>
		private Stream m_parent;
		/// <summary>
		/// The internal read-ahead buffer
		/// </summary>
		private byte[] m_buffer;
		/// <summary>
		/// The number of bytes in the internal buffer
		/// </summary>
		private int m_buffercount = 0;
		/// <summary>
		/// The offset in the internal buffer from where the data starts
		/// </summary>
		private int m_bufferoffset = 0;
		/// <summary>
		/// The CR ASCII value.
		/// </summary>
		private const byte CR = 13;
		/// <summary>
		/// The LF ASCII value.
		/// </summary>
		private const byte LF = 10;
		/// <summary>
		/// The number of bytes we are allowed to read
		/// </summary>
		private long m_remainingbytes = 0;
		/// <summary>
		/// The number of bytes read
		/// </summary>
		private long m_maxread = 0;

		/// <summary>
		/// Initializes a new instance of the <see cref="Ceen.Httpd.BufferedStreamReader"/> class.
		/// </summary>
		/// <param name="parent">The parent stream.</param>
		/// <param name="initialbuffersize">Intitial size of the read-ahead buffer.</param>
		public BufferedStreamReader(Stream parent, int initialbuffersize = 1024)
		{
			if (parent == null)
				throw new ArgumentNullException(nameof(parent));
			
			m_parent = parent;
			m_buffer = new byte[initialbuffersize];
		}

		/// <summary>
		/// Resets the number of bytes read.
		/// </summary>
		/// <param name="maxsize">The maximum number of bytes to allow reading</param>
		internal void ResetReadLength(long maxsize)
		{
			m_maxread = m_remainingbytes = maxsize;
		}


		/// <summary>
		/// Reads all lines until an empty line (i.e. it reads until CR LF CR LF).
		/// Each read line, except the empty line is send to the linehandler function
		/// </summary>
		/// <returns>An awaitable task.</returns>
		/// <param name="maxlinesize">The maximum size of a single line.</param>
		/// <param name="maxheadersize">The maximum size of the headers.</param>
		/// <param name="idletimeout">The maximum time to remain idle.</param>
		/// <param name="linehandler">Callback method that processes each line.</param>
		/// <param name="timeouttask">Awaitable task for stopping the request.</param>
		/// <param name="stoptask">Awaitable task for stopping the request.</param>
		public async Task ReadHeaders(int maxlinesize, int maxheadersize, TimeSpan idletimeout, Action<string> linehandler, Task timeouttask, Task stoptask)
		{
			var buf = new byte[maxlinesize];
			var bufoffset = 0;
			var bufsize = 0;
			var totalread = 0;
			var lastlookoffset = 0;

			// Repeat the parsing until we time out or get an empty line
			while (true)
			{
				if (totalread == maxheadersize)
					throw new HttpException(HttpStatusCode.RequestHeaderFieldsTooLarge);

                Task<int> rtask;
                Task rt;

                using (var cs = new CancellationTokenSource(idletimeout))
                {
                    rtask = ReadAsync(buf, bufsize, Math.Min(buf.Length - bufsize, maxheadersize - totalread), cs.Token);
                    rt = await Task.WhenAny( stoptask, timeouttask, rtask);
                }

                // Timeout or stop has happened
				if (rt != rtask)
				{
					if (rt == stoptask)
						throw new TaskCanceledException();
					else if (totalread == 0)
						throw new EmptyStreamClosedException();
					else
						throw new HttpException(HttpStatusCode.RequestTimeout); 
				}

				var r = rtask.Result;
				totalread += r;
				bufsize += r;

				// Stream closed
				if (r == 0)
				{
					if (totalread == 0)
						throw new EmptyStreamClosedException();
					else
						throw new HttpException(HttpStatusCode.BadRequest);
				}

				// Look for first CR
				var ix = Array.IndexOf(buf, CR, lastlookoffset, bufsize - lastlookoffset);
				while (ix >= 0 && ix < bufsize - 1)
				{
					// Check if the next byte is LF
					if (buf[ix + 1] == LF)
					{
						var datalen = ix - bufoffset;

						// Check if the line is empty
						if (datalen == 0)
						{
							bufoffset = ix + 2;

							// If we have more data, "unread it"
							if (bufsize - bufoffset != 0)
								UnreadBytesIntoBuffer(buf, bufoffset, bufsize - bufoffset);

							return;
						}
						else
						{
							linehandler(System.Text.Encoding.ASCII.GetString(buf, bufoffset, datalen));
							bufoffset = ix + 2;
							lastlookoffset = bufoffset;
						}
					}
					else
					{
						// False hit, or char hit a split with blocks
						lastlookoffset = ix + 1;
					}

					// Get the next entry
					ix = Array.IndexOf(buf, CR, lastlookoffset, bufsize - lastlookoffset);
				}

				// Move trailing bytes to the start of the buffer to reduce space use
				// and make sure we do not search more than once for the CRLF
				bufsize = bufsize - bufoffset;
				if (bufoffset > 0 && bufsize > 0)
					Array.Copy(buf, bufoffset, buf, 0, bufsize);

				bufoffset = 0;
				lastlookoffset = Math.Max(0, bufsize - 1);
			}
		}

		/// <summary>
		/// Keeps reading from the stream until the supplied number of bytes are read
		/// </summary>
		/// <returns>The awaitable task.</returns>
		/// <param name="count">The number of bytes to read.</param>
		/// <param name="idletimeout">The time to wait while idle</param>
		/// <param name="timeouttask">The task that signals request timeout</param>
		/// <param name="stoptask">The task that signals stop for the server</param>
		internal async Task<byte[]> RepeatReadAsync(int count, TimeSpan idletimeout, Task timeouttask, Task stoptask)
		{
			var buf = new byte[count];
			await RepeatReadAsync(buf, 0, count, idletimeout, timeouttask, stoptask);
			return buf;
		}

		/// <summary>
		/// Gets the delimited sub stream.
		/// </summary>
		/// <returns>The delimited sub stream.</returns>
		/// <param name="delimiter">Delimiter.</param>
		/// <param name="idletime">The time to wait while idle</param>
{'fraction_non_alphanumeric': 0.12276557209229073, 'fraction_numerical': 0.005276005984723207, 'mean_word_length': 2.4976590470944644, 'pattern_counts': {'":': 0, '<': 141, '<?xml version=': 0, '>': 143, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7228798', 'n_tokens_mistral': 4707, 'n_tokens_neox': 3752, 'n_words': 1289}