<commit_before>// File: TcpTransport.cs
// Project: ROS_C-Sharp
// 
// ROS.NET
// <NAME> <<EMAIL>>
// UMass Lowell Robotics Laboratory
// 
// Reimplementation of the ROS (ros.org) ros_cpp client in C#.
// 
// Created: 09/01/2015
// Updated: 02/10/2016

#region USINGZ

using System;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using Socket = Ros_CSharp.CustomSocket.Socket;

#endregion

namespace Ros_CSharp
{
    public class TcpTransport
    {
        #region Delegates

        public delegate void AcceptCallback(TcpTransport trans);

        public delegate void DisconnectFunc(TcpTransport trans);

        public delegate void HeaderReceivedFunc(TcpTransport trans, Header header);

        public delegate void ReadFinishedFunc(TcpTransport trans);

        public delegate void WriteFinishedFunc(TcpTransport trans);

        #endregion

        #region Flags enum

        public enum Flags
        {
            SYNCHRONOUS = 1 << 0
        }

        #endregion

        private const int bytesperlong = 4; // 32 / 8
        private const int bitsperbyte = 8;
        public const int POLLERR = 0x008;
        public const int POLLHUP = 0x010;
        public const int POLLNVAL = 0x020;
        public const int POLLIN = 0x001;
        public const int POLLOUT = 0x004;

        public static bool use_keepalive;
        public IPEndPoint LocalEndPoint;
        public string _topic;
        public string cached_remote_host = "";
        public object close_mutex = new object();
        public bool closed;
        public string connected_host;
        public int connected_port;
        public int events;
        public bool expecting_read;
        public bool expecting_write;
        public int flags;
        public bool is_server;
        public bool no_delay;
        public PollSet poll_set;
        public IPEndPoint server_address;
        public int server_port = -1;

        private Socket sock;

        public TcpTransport()
        {
        }


        public TcpTransport(System.Net.Sockets.Socket s, PollSet pollset) : this(s, pollset, 0)
        {
        }

        public TcpTransport(System.Net.Sockets.Socket s, PollSet pollset, int flags) : this(pollset, flags)
        {
            setSocket(new Socket(s));
        }

        public TcpTransport(PollSet pollset)
            : this(pollset, 0)
        {
        }

        public TcpTransport(PollSet pollset, int flags) : this()
        {
            if (pollset != null)
            {
                poll_set = pollset;
                poll_set.DisposingEvent += close;
            }
            else
            {
                throw new Exception("Null pollset in tcptransport ctor");
            }
            this.flags = flags;
        }

        public string ClientURI
        {
#if !TRACE
            [DebuggerStepThrough]
#endif
                get
            {
                if (connected_host == null || connected_port == 0)
                    return "[NOT CONNECTED]";
                return "http://" + connected_host + ":" + connected_port + "/";
            }
        }

        public string Topic
        {
            get { return _topic != null ? _topic : "?!?!?!"; }
        }

        public virtual bool getRequiresHeader()
        {
            return true;
        }

        public event AcceptCallback accept_cb;
        public event DisconnectFunc disconnect_cb;
        public event WriteFinishedFunc write_cb;
        public event ReadFinishedFunc read_cb;

        public bool setNonBlocking()
        {
            if ((flags & (int) Flags.SYNCHRONOUS) == 0)
            {
                try
                {
                    sock.Blocking = false;
                }
                catch (Exception e)
                {
                    EDB.WriteLine(e);
                    close();
                    return false;
                }
            }

            return true;
        }

        public void setNoDelay(bool nd)
        {
            try
            {
                sock.NoDelay = nd;
            }
            catch (Exception e)
            {
                EDB.WriteLine(e);
            }
        }

        public void enableRead()
        {
            if (sock == null)
                throw new Exception("TcpTransport: Attempted to enable read before setSocket called");
            if (!sock.Connected)
                close();
            lock (close_mutex)
            {
                if (closed) return;
            }
            if (!expecting_read && poll_set != null)
            {
                //Console.WriteLine("ENABLE READ:   " + Topic + "(" + sock.FD + ")");
                expecting_read = true;
                poll_set.addEvents(sock, POLLIN);
            }
        }

        public void disableRead()
        {
            if (sock == null)
                return;
            if (!sock.Connected)
                close();
            lock (close_mutex)
            {
                if (closed) return;
            }
            if (expecting_read && poll_set != null)
            {
                //Console.WriteLine("DISABLE READ:  " + Topic + "(" + sock.FD + ")");
                poll_set.delEvents(sock, POLLIN);
                expecting_read = false;
            }
        }

        public void enableWrite()
        {
            if (!sock.Connected) close();
            lock (close_mutex)
            {
                if (closed) return;
            }
            if (!expecting_write && poll_set != null)
            {
                //Console.WriteLine("ENABLE WRITE:  " + Topic + "(" + sock.FD + ")");
                expecting_write = true;
                poll_set.addEvents(sock, POLLOUT);
            }
        }

        public void disableWrite()
        {
            if (!sock.Connected) close();
            lock (close_mutex)
            {
                if (closed) return;
            }
            if (expecting_write && poll_set != null)
            {
                //Console.WriteLine("DISABLE WRITE: " + Topic + "(" + sock.FD + ")");
                poll_set.delEvents(sock, POLLOUT);
                expecting_write = false;
            }
        }

        public bool connect(string host, int port)
        {
            sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            connected_host = host;
            connected_port = port;
            if (!setNonBlocking())<commit_msg>TcpTransport: Prevent NPE (MAY occur during ROS shutdown)<commit_after>// File: TcpTransport.cs
// Project: ROS_C-Sharp
// 
// ROS.NET
// <NAME> <<EMAIL>>
// UMass Lowell Robotics Laboratory
// 
// Reimplementation of the ROS (ros.org) ros_cpp client in C#.
// 
// Created: 09/01/2015
// Updated: 02/10/2016

#region USINGZ

using System;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using Socket = Ros_CSharp.CustomSocket.Socket;

#endregion

namespace Ros_CSharp
{
    public class TcpTransport
    {
        #region Delegates

        public delegate void AcceptCallback(TcpTransport trans);

        public delegate void DisconnectFunc(TcpTransport trans);

        public delegate void HeaderReceivedFunc(TcpTransport trans, Header header);

        public delegate void ReadFinishedFunc(TcpTransport trans);

        public delegate void WriteFinishedFunc(TcpTransport trans);

        #endregion

        #region Flags enum

        public enum Flags
        {
            SYNCHRONOUS = 1 << 0
        }

        #endregion

        private const int bytesperlong = 4; // 32 / 8
        private const int bitsperbyte = 8;
        public const int POLLERR = 0x008;
        public const int POLLHUP = 0x010;
        public const int POLLNVAL = 0x020;
        public const int POLLIN = 0x001;
        public const int POLLOUT = 0x004;

        public static bool use_keepalive;
        public IPEndPoint LocalEndPoint;
        public string _topic;
        public string cached_remote_host = "";
        public object close_mutex = new object();
        public bool closed;
        public string connected_host;
        public int connected_port;
        public int events;
        public bool expecting_read;
        public bool expecting_write;
        public int flags;
        public bool is_server;
        public bool no_delay;
        public PollSet poll_set;
        public IPEndPoint server_address;
        public int server_port = -1;

        private Socket sock;

        public TcpTransport()
        {
        }


        public TcpTransport(System.Net.Sockets.Socket s, PollSet pollset) : this(s, pollset, 0)
        {
        }

        public TcpTransport(System.Net.Sockets.Socket s, PollSet pollset, int flags) : this(pollset, flags)
        {
            setSocket(new Socket(s));
        }

        public TcpTransport(PollSet pollset)
            : this(pollset, 0)
        {
        }

        public TcpTransport(PollSet pollset, int flags) : this()
        {
            if (pollset != null)
            {
                poll_set = pollset;
                poll_set.DisposingEvent += close;
            }
            else
            {
                throw new Exception("Null pollset in tcptransport ctor");
            }
            this.flags = flags;
        }

        public string ClientURI
        {
#if !TRACE
            [DebuggerStepThrough]
#endif
                get
            {
                if (connected_host == null || connected_port == 0)
                    return "[NOT CONNECTED]";
                return "http://" + connected_host + ":" + connected_port + "/";
            }
        }

        public string Topic
        {
            get { return _topic != null ? _topic : "?!?!?!"; }
        }

        public virtual bool getRequiresHeader()
        {
            return true;
        }

        public event AcceptCallback accept_cb;
        public event DisconnectFunc disconnect_cb;
        public event WriteFinishedFunc write_cb;
        public event ReadFinishedFunc read_cb;

        public bool setNonBlocking()
        {
            if ((flags & (int) Flags.SYNCHRONOUS) == 0)
            {
                try
                {
                    sock.Blocking = false;
                }
                catch (Exception e)
                {
                    EDB.WriteLine(e);
                    close();
                    return false;
                }
            }

            return true;
        }

        public void setNoDelay(bool nd)
        {
            try
            {
                sock.NoDelay = nd;
            }
            catch (Exception e)
            {
                EDB.WriteLine(e);
            }
        }

        public void enableRead()
        {
            if (sock == null)
                return;
            if (!sock.Connected)
                close();
            lock (close_mutex)
            {
                if (closed) return;
            }
            if (!expecting_read && poll_set != null)
            {
                //Console.WriteLine("ENABLE READ:   " + Topic + "(" + sock.FD + ")");
                expecting_read = true;
                poll_set.addEvents(sock, POLLIN);
            }
        }

        public void disableRead()
        {
            if (sock == null)
                return;
            if (!sock.Connected)
                close();
            lock (close_mutex)
            {
                if (closed) return;
            }
            if (expecting_read && poll_set != null)
            {
                //Console.WriteLine("DISABLE READ:  " + Topic + "(" + sock.FD + ")");
                poll_set.delEvents(sock, POLLIN);
                expecting_read = false;
            }
        }

        public void enableWrite()
        {
            if (sock == null)
                return;
            if (!sock.Connected) close();
            lock (close_mutex)
            {
                if (closed) return;
            }
            if (!expecting_write && poll_set != null)
            {
                //Console.WriteLine("ENABLE WRITE:  " + Topic + "(" + sock.FD + ")");
                expecting_write = true;
                poll_set.addEvents(sock, POLLOUT);
            }
        }

        public void disableWrite()
        {
            if (sock == null)
                return;
            if (!sock.Connected) close();
            lock (close_mutex)
            {
                if (closed) return;
            }
            if (expecting_write && poll_set != null)
            {
                //Console.WriteLine("DISABLE WRITE: " + Topic + "(" + sock.FD + ")");
                poll_set.delEvents(sock, POLLOUT);
                expecting_write = false;
            }
        }

        public bool connect(string host, int port)
        {
            sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            connected_host = host;
            connected_port = port;
            if (!setNonBlocking())
{'fraction_non_alphanumeric': 0.0881433473122379, 'fraction_numerical': 0.007319862752573389, 'mean_word_length': 1.3417246920192822, 'pattern_counts': {'":': 2, '<': 13, '<?xml version=': 0, '>': 9, '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': '778388', 'n_tokens_mistral': 3633, 'n_tokens_neox': 3399, 'n_words': 883}