<commit_before>﻿using System;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using ByteNom.Protocol;
using ProtoBuf;

namespace ByteNom
{
    /// <summary>
    ///     Represents the abstract base class for all ByteNom connections.
    /// </summary>
    public abstract class Connection : IDisposable
    {
        private bool _disposed;
        private Thread _thread;

        /// <summary>
        ///     Gets a value indicating whether this <see cref="Connection" /> is connected.
        /// </summary>
        /// <value>
        ///     <c>true</c> if connected; otherwise, <c>false</c>.
        /// </value>
        public bool Connected
        {
            get
            {
                if (this.Client == null)
                    return false;

                return this.Client.Connected;
            }
        }

        /// <summary>
        ///     Gets the remote end point.
        /// </summary>
        /// <value>
        ///     The remote end point.
        /// </value>
        public EndPoint EndPoint
        {
            get
            {
                if (this.Client == null)
                    return null;

                return this.Client.Client.RemoteEndPoint;
            }
        }

        /// <summary>
        ///     Gets the tcp client.
        /// </summary>
        /// <value>
        ///     The tcp client.
        /// </value>
        protected TcpClient Client { get; private set; }

        /// <summary>
        ///     Gets the network stream.
        /// </summary>
        /// <value>
        ///     The network stream.
        /// </value>
        protected NetworkStream Stream { get; private set; }

        void IDisposable.Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        /// <summary>
        ///     Sets the internal tcp client for this connection.
        /// </summary>
        /// <param name="client">The client.</param>
        protected void SetClient(TcpClient client)
        {
            this.Client = client;
            this.Stream = client.GetStream();
        }

        /// <summary>
        ///     Starts listening for messages.
        /// </summary>
        protected void Start()
        {
            this._thread = new Thread(this.ThreadRun)
            {
                Name = "ByteNom.Connection",
                IsBackground = true
            };
            this._thread.Start();
        }

        private void ThreadRun()
        {
            try
            {
                this.Work();
            }
            catch (IOException)
            {
            }
            catch (SocketException)
            {
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex);
            }
            finally
            {
                this.Disconnect();
                this.OnDisconnected();
            }
        }

        /// <summary>
        ///     This method is run from the external thread of the connection. Override this to perform startup/shutdown tasks.
        /// </summary>
        protected virtual void Work()
        {
            while (this.Connected)
            {
                var dataMsg = this.ProtoGet<DataMessage>();

                // If we received 0 bytes, the connection is closed
                if (dataMsg == null) break;

                Message msg = MessageSerializer.Deserialize(dataMsg);
                this.OnMessageReceived(msg);
            }
        }

        /// <summary>
        ///     Creates and sends a message with the specified data.
        /// </summary>
        /// <param name="type">The message type.</param>
        /// <param name="args">The message arguments.</param>
        public void Send(string type, params object[] args)
        {
            this.Send(new Message(type, args));
        }

        /// <summary>
        ///     Sends the specified message.
        /// </summary>
        /// <param name="message">The message.</param>
        public void Send(Message message)
        {
            DataMessage dataMsg = MessageSerializer.Serialize(message);

            try
            {
                if (!this.Connected) return;
                this.ProtoSend(dataMsg);
            }
            catch (IOException)
            {
                this.Disconnect();
            }
        }

        /// <summary>
        ///     Disconnects from the server.
        /// </summary>
        public void Disconnect()
        {
            this.Stream.Close();
            this.Client.Close();
        }

        internal T ProtoGet<T>()
        {
            return Serializer.DeserializeWithLengthPrefix<T>(this.Stream, PrefixStyle.Base128);
        }

        internal void ProtoSend<T>(T messageObj)
        {
            Serializer.SerializeWithLengthPrefix(this.Stream, messageObj, PrefixStyle.Base128);
        }

        /// <summary>
        ///     Occurs when a message is received.
        /// </summary>
        public event MessageEventHandler MessageReceived;

        /// <summary>
        ///     Called when a message is received.
        /// </summary>
        /// <param name="message">The message.</param><commit_msg>Fix exception if the connection is aborted while awaiting Hello<commit_after>﻿using System;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using ByteNom.Protocol;
using ProtoBuf;

namespace ByteNom
{
    /// <summary>
    ///     Represents the abstract base class for all ByteNom connections.
    /// </summary>
    public abstract class Connection : IDisposable
    {
        private bool _disposed;
        private Thread _thread;

        /// <summary>
        ///     Gets a value indicating whether this <see cref="Connection" /> is connected.
        /// </summary>
        /// <value>
        ///     <c>true</c> if connected; otherwise, <c>false</c>.
        /// </value>
        public bool Connected
        {
            get
            {
                if (this.Client == null)
                    return false;

                return this.Client.Connected;
            }
        }

        /// <summary>
        ///     Gets the remote end point.
        /// </summary>
        /// <value>
        ///     The remote end point.
        /// </value>
        public EndPoint EndPoint
        {
            get
            {
                if (this.Client == null)
                    return null;

                return this.Client.Client.RemoteEndPoint;
            }
        }

        /// <summary>
        ///     Gets the tcp client.
        /// </summary>
        /// <value>
        ///     The tcp client.
        /// </value>
        protected TcpClient Client { get; private set; }

        /// <summary>
        ///     Gets the network stream.
        /// </summary>
        /// <value>
        ///     The network stream.
        /// </value>
        protected NetworkStream Stream { get; private set; }

        void IDisposable.Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        /// <summary>
        ///     Sets the internal tcp client for this connection.
        /// </summary>
        /// <param name="client">The client.</param>
        protected void SetClient(TcpClient client)
        {
            this.Client = client;
            this.Stream = client.GetStream();
        }

        /// <summary>
        ///     Starts listening for messages.
        /// </summary>
        protected void Start()
        {
            this._thread = new Thread(this.ThreadRun)
            {
                Name = "ByteNom.Connection",
                IsBackground = true
            };
            this._thread.Start();
        }

        private void ThreadRun()
        {
            try
            {
                this.Work();
            }
            catch (IOException)
            {
            }
            catch (SocketException)
            {
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex);
            }
            finally
            {
                this.Disconnect();
                this.OnDisconnected();
            }
        }

        /// <summary>
        ///     This method is run from the external thread of the connection. Override this to perform startup/shutdown tasks.
        /// </summary>
        protected virtual void Work()
        {
            while (this.Connected)
            {
                var dataMsg = this.ProtoGet<DataMessage>();
                Message msg = MessageSerializer.Deserialize(dataMsg);
                this.OnMessageReceived(msg);
            }
        }

        /// <summary>
        ///     Creates and sends a message with the specified data.
        /// </summary>
        /// <param name="type">The message type.</param>
        /// <param name="args">The message arguments.</param>
        public void Send(string type, params object[] args)
        {
            this.Send(new Message(type, args));
        }

        /// <summary>
        ///     Sends the specified message.
        /// </summary>
        /// <param name="message">The message.</param>
        public void Send(Message message)
        {
            DataMessage dataMsg = MessageSerializer.Serialize(message);

            try
            {
                if (!this.Connected) return;
                this.ProtoSend(dataMsg);
            }
            catch (IOException)
            {
                this.Disconnect();
            }
        }

        /// <summary>
        ///     Disconnects from the server.
        /// </summary>
        public void Disconnect()
        {
            this.Stream.Close();
            this.Client.Close();
        }

        internal T ProtoGet<T>() where T : class
        {
            var value = Serializer.DeserializeWithLengthPrefix<T>(this.Stream, PrefixStyle.Base128);
            if (value == null)
                throw new IOException("The connection was closed.");

            return value;
        }

        internal void ProtoSend<T>(T messageObj)
        {
            Serializer.SerializeWithLengthPrefix(this.Stream, messageObj, PrefixStyle.Base128);
        }

        /// <summary>
        ///     Occurs when a message is received.
        /// </summary>
        public event MessageEventHandler MessageReceived;

        /// <summary>
        ///     Called when a message is received.
        /// </summary>
        /// <param name="message">The message.</param>
{'fraction_non_alphanumeric': 0.11281809613572102, 'fraction_numerical': 0.0012252591894439209, 'mean_word_length': 1.171270718232044, 'pattern_counts': {'":': 0, '<': 109, '<?xml version=': 0, '>': 109, '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': '6847339', 'n_tokens_mistral': 2632, 'n_tokens_neox': 2549, 'n_words': 668}