<commit_before>using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Net.Sockets;
using System.Net;
using Manos.IO;
using Manos.Collections;

namespace Manos.Managed
{
	public class Socket : Manos.IO.Socket
	{
		System.Net.Sockets.Socket socket;
		Action connectedCallback;
		Action<Socket> acceptedCallback;
		Stream stream;
		IOLoop loop;

		public Socket (IOLoop loop)
		{
			if (loop == null)
				throw new ArgumentNullException ("loop");
			this.loop = loop;
		}

		Socket (IOLoop loop, System.Net.Sockets.Socket socket) : this (loop)
		{
			this.socket = socket;
			this.address = ((IPEndPoint) socket.RemoteEndPoint).Address.ToString ();
			this.port = ((IPEndPoint) socket.RemoteEndPoint).Port;
            this.state = SocketState.Open;
		}
		
		class SocketStream : Manos.IO.Stream
		{
			Socket parent;
			bool readAllowed, writeAllowed;
			long readLimit, position;
			byte [] receiveBuffer = new byte [4096];
			System.Timers.Timer readTimer, writeTimer;

			public SocketStream (Socket parent)
			{
				this.parent = parent;
			}

            public override bool Managed
            {
                get { return true; }
            }

			public override long Position {
				get { return position; }
				set { SeekTo (value); }
			}

			public override bool CanRead {
				get { return true; }
			}

			public override bool CanWrite {
				get { return true; }
			}

			public override bool CanTimeout {
				get { return true; }
			}

			public override TimeSpan ReadTimeout {
				get { return readTimer == null ? TimeSpan.Zero : TimeSpan.FromMilliseconds (readTimer.Interval); }
				set {
					if (value < TimeSpan.Zero)
						throw new ArgumentException ("value");
					if (readTimer == null) {
						readTimer = new System.Timers.Timer (value.TotalMilliseconds);
						readTimer.Elapsed += HandleReadTimerElapsed;
					}
					if (value == TimeSpan.Zero) {
						readTimer.Stop ();
					} else {
						readTimer.Interval = value.TotalMilliseconds;
						readTimer.Start ();
					}
				}
			}

			public override TimeSpan WriteTimeout {
				get { return writeTimer == null ? TimeSpan.Zero : TimeSpan.FromMilliseconds (writeTimer.Interval); }
				set {
					if (value < TimeSpan.Zero)
						throw new ArgumentException ("value");
					if (writeTimer == null) {
						writeTimer = new System.Timers.Timer (value.TotalMilliseconds);
						writeTimer.Elapsed += HandleWriteTimerElapsed;
					}
					if (value == TimeSpan.Zero) {
						writeTimer.Stop ();
					} else {
						writeTimer.Interval = value.TotalMilliseconds;
						writeTimer.Start ();
					}
				}
			}

			void HandleReadTimerElapsed (object sender, System.Timers.ElapsedEventArgs e)
			{
				if (readAllowed) {
					RaiseError (new TimeoutException ());
					PauseReading ();
				}
			}

			void HandleWriteTimerElapsed (object sender, System.Timers.ElapsedEventArgs e)
			{
				if (writeAllowed) {
					RaiseError (new TimeoutException ());
					PauseWriting ();
				}
			}

			public override void Write (IEnumerable<ByteBuffer> data)
			{
				base.Write (data);
				ResumeWriting ();
			}

			public override void ResumeReading ()
			{
				ResumeReading (long.MaxValue);
			}

			public override void ResumeReading (long forBytes)
			{
				if (forBytes < 0)
					throw new ArgumentException ("forBytes");
				readLimit = forBytes;
                readAllowed = true;
                HandleRead ();
			}

			public override void ResumeWriting ()
			{
				writeAllowed = true;
				HandleWrite ();
			}

			public override void PauseReading ()
			{
				readAllowed = false;
			}

			public override void PauseWriting ()
			{
				writeAllowed = false;
			}

			public override void Flush ()
			{<commit_msg>Last fix; now sending files works on the managed stream too<commit_after>using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Net.Sockets;
using System.Net;
using Manos.IO;
using Manos.Collections;

namespace Manos.Managed
{
	public class Socket : Manos.IO.Socket
	{
		System.Net.Sockets.Socket socket;
		Action connectedCallback;
		Action<Socket> acceptedCallback;
		Stream stream;
		IOLoop loop;

		public Socket (IOLoop loop)
		{
			if (loop == null)
				throw new ArgumentNullException ("loop");
			this.loop = loop;
		}

		Socket (IOLoop loop, System.Net.Sockets.Socket socket) : this (loop)
		{
			this.socket = socket;
			this.address = ((IPEndPoint) socket.RemoteEndPoint).Address.ToString ();
			this.port = ((IPEndPoint) socket.RemoteEndPoint).Port;
            this.state = SocketState.Open;
		}
		
		class SocketStream : Manos.IO.Stream
		{
			Socket parent;
			bool readAllowed, writeAllowed;
			long readLimit, position;
			byte [] receiveBuffer = new byte [4096];
			System.Timers.Timer readTimer, writeTimer;

			public SocketStream (Socket parent)
			{
				this.parent = parent;
			}

            public override bool Managed
            {
                get { return true; }
            }

			public override long Position {
				get { return position; }
				set { SeekTo (value); }
			}

			public override bool CanRead {
				get { return true; }
			}

			public override bool CanWrite {
				get { return true; }
			}

			public override bool CanTimeout {
				get { return true; }
			}

			public override TimeSpan ReadTimeout {
				get { return readTimer == null ? TimeSpan.Zero : TimeSpan.FromMilliseconds (readTimer.Interval); }
				set {
					if (value < TimeSpan.Zero)
						throw new ArgumentException ("value");
					if (readTimer == null) {
						readTimer = new System.Timers.Timer (value.TotalMilliseconds);
						readTimer.Elapsed += HandleReadTimerElapsed;
					}
					if (value == TimeSpan.Zero) {
						readTimer.Stop ();
					} else {
						readTimer.Interval = value.TotalMilliseconds;
						readTimer.Start ();
					}
				}
			}

			public override TimeSpan WriteTimeout {
				get { return writeTimer == null ? TimeSpan.Zero : TimeSpan.FromMilliseconds (writeTimer.Interval); }
				set {
					if (value < TimeSpan.Zero)
						throw new ArgumentException ("value");
					if (writeTimer == null) {
						writeTimer = new System.Timers.Timer (value.TotalMilliseconds);
						writeTimer.Elapsed += HandleWriteTimerElapsed;
					}
					if (value == TimeSpan.Zero) {
						writeTimer.Stop ();
					} else {
						writeTimer.Interval = value.TotalMilliseconds;
						writeTimer.Start ();
					}
				}
			}

			void HandleReadTimerElapsed (object sender, System.Timers.ElapsedEventArgs e)
			{
				if (readAllowed) {
					RaiseError (new TimeoutException ());
					PauseReading ();
				}
			}

			void HandleWriteTimerElapsed (object sender, System.Timers.ElapsedEventArgs e)
			{
				if (writeAllowed) {
					RaiseError (new TimeoutException ());
					PauseWriting ();
				}
			}

			public override void Write (IEnumerable<ByteBuffer> data)
			{
				base.Write (data);
				ResumeWriting ();
			}

			public override void ResumeReading ()
			{
				ResumeReading (long.MaxValue);
			}

			public override void ResumeReading (long forBytes)
			{
				if (forBytes < 0)
					throw new ArgumentException ("forBytes");
				readLimit = forBytes;
                readAllowed = true;
                HandleRead ();
			}

			public override void ResumeWriting ()
			{
                if (writeAllowed) return;
				writeAllowed = true;
				HandleWrite ();
			}

			public override void PauseReading ()
			{
				readAllowed = false;
			}

			public override void PauseWriting ()
			{
				writeAllowed = false;
			}

			public override void Flush ()
			{
{'fraction_non_alphanumeric': 0.09994697773064687, 'fraction_numerical': 0.001325556733828208, 'mean_word_length': 2.603151862464183, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 7, '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': '4991048', 'n_tokens_mistral': 2918, 'n_tokens_neox': 2215, 'n_words': 567}