<commit_before>// Copyright (c) 2013 <NAME>
// See the file COPYING for license details.

using System;
using System.Diagnostics;
using System.Net;
using System.Net.Sockets;
using System.Threading;

using x2.Events;
using x2.Flows;

namespace x2.Links.SocketLink
{
    /// <summary>
    /// TCP/IP client link based on the enhanced SocketAsyncEventArgs pattern.
    /// </summary>
    public class AsyncTcpClient : TcpClientBase
    {
        private SocketAsyncEventArgs connectEventArgs;

        public AsyncTcpClient(string name)
            : base(name)
        {
        }

        protected override void ConnectImpl(EndPoint endpoint)
        {
            if (connectEventArgs == null)
            {
                connectEventArgs = new SocketAsyncEventArgs();
                connectEventArgs.Completed += OnConnectCompleted;
            }
            connectEventArgs.RemoteEndPoint = endpoint;

            bool pending = socket.ConnectAsync(connectEventArgs);
            if (!pending)
            {
                OnConnect(connectEventArgs);
            }
        }

        // Completed event handler for ConnectAsync
        private void OnConnectCompleted(object sender, SocketAsyncEventArgs e)
        {
            OnConnect(e);
        }

        // Completion callback for ConnectAsync
        private void OnConnect(SocketAsyncEventArgs e)
        {
            var noti = new LinkSessionConnected { LinkName = Name };

            if (e.SocketError == SocketError.Success)
            {
                // Adjust socket options.
                socket.NoDelay = NoDelay;

                Log.Info("{0} {1} connected to {2}", Name, socket.Handle, socket.RemoteEndPoint);

                noti.Result = true;

                ConnectInternal();

                connectEventArgs = null;

                session = new AsyncTcpLinkSession(this, socket);
                session.Polarity = true;

                if (BufferTransform != null)
                {
                    session.BufferTransform = BufferTransform;
                }

                noti.Context = session;
                Flow.Publish(noti);

                session.BeginReceive(true);
            }
            else
            {
                Log.Warn("{0} connect error {1}", Name, e.SocketError);

                noti.Context = e.RemoteEndPoint;
                Flow.Publish(noti);

                RetryInternal(e.RemoteEndPoint);
            }
        }
    }

    public class AsyncTcpClientFlow : FrameBasedFlow
    {
        protected AsyncTcpClient link;

        private volatile bool closeOnHeartbeatFailure;

        protected x2.Flows.Timer Timer { get; private set; }

        public bool CloseOnHeartbeatFailure
        {
            get { return closeOnHeartbeatFailure; }
            set { closeOnHeartbeatFailure = value; }
        }
        public double HeartbeatTimeout { get; set; }  // in seconds

        public bool AutoReconnect
        {
            get { return link.AutoReconnect; }
            set { link.AutoReconnect = value; }
        }
        public int MaxRetryCount  // 0 for unlimited
        {
            get { return link.MaxRetryCount; }
            set { link.MaxRetryCount = value; }
        }
        public long RetryInterval  // in millisec
        {
            get { return link.RetryInterval; }
            set { link.RetryInterval = value; }
        }

        public Action<Event, LinkSession> Preprocessor
        {
            get { return link.Preprocessor; }
            set { link.Preprocessor = value; }
        }

        public AsyncTcpClientFlow(string name)
        {
            link = new AsyncTcpClient(name);
            Add(link);

            link.HeartbeatEventHandler = OnHeartbeatEvent;

            this.name = name;

            Resolution = Time.TicksInSecond;  // 1-second frame resolution
            HeartbeatTimeout = 15;

            Timer = new x2.Flows.Timer(OnTimer);
        }

        public void Close()
        {
            Timer.Cancel(link.Session.HeartbeatTimeoutToken);

            link.Close();
        }

        public void Connect(string host, int port)
        {
            link.Connect(host, port);
        }

        protected override void SetUp()
        {
            base.SetUp();

            Subscribe(new LinkSessionConnected { LinkName = Name }, OnLinkSessionConnected);
            Subscribe(new LinkSessionDisconnected { LinkName = Name }, OnLinkSessionDisconnected);
        }<commit_msg>added additional constructor to AsyncTcpClient<commit_after>// Copyright (c) 2013 <NAME>
// See the file COPYING for license details.

using System;
using System.Diagnostics;
using System.Net;
using System.Net.Sockets;
using System.Threading;

using x2.Events;
using x2.Flows;

namespace x2.Links.SocketLink
{
    /// <summary>
    /// TCP/IP client link based on the enhanced SocketAsyncEventArgs pattern.
    /// </summary>
    public class AsyncTcpClient : TcpClientBase
    {
        private SocketAsyncEventArgs connectEventArgs;

        public AsyncTcpClient(string name)
            : base(name)
        {
        }

        protected override void ConnectImpl(EndPoint endpoint)
        {
            if (connectEventArgs == null)
            {
                connectEventArgs = new SocketAsyncEventArgs();
                connectEventArgs.Completed += OnConnectCompleted;
            }
            connectEventArgs.RemoteEndPoint = endpoint;

            bool pending = socket.ConnectAsync(connectEventArgs);
            if (!pending)
            {
                OnConnect(connectEventArgs);
            }
        }

        // Completed event handler for ConnectAsync
        private void OnConnectCompleted(object sender, SocketAsyncEventArgs e)
        {
            OnConnect(e);
        }

        // Completion callback for ConnectAsync
        private void OnConnect(SocketAsyncEventArgs e)
        {
            var noti = new LinkSessionConnected { LinkName = Name };

            if (e.SocketError == SocketError.Success)
            {
                // Adjust socket options.
                socket.NoDelay = NoDelay;

                Log.Info("{0} {1} connected to {2}", Name, socket.Handle, socket.RemoteEndPoint);

                noti.Result = true;

                ConnectInternal();

                connectEventArgs = null;

                session = new AsyncTcpLinkSession(this, socket);
                session.Polarity = true;

                if (BufferTransform != null)
                {
                    session.BufferTransform = BufferTransform;
                }

                noti.Context = session;
                Flow.Publish(noti);

                session.BeginReceive(true);
            }
            else
            {
                Log.Warn("{0} connect error {1}", Name, e.SocketError);

                noti.Context = e.RemoteEndPoint;
                Flow.Publish(noti);

                RetryInternal(e.RemoteEndPoint);
            }
        }
    }

    public class AsyncTcpClientFlow : FrameBasedFlow
    {
        protected AsyncTcpClient link;

        private volatile bool closeOnHeartbeatFailure;

        protected x2.Flows.Timer Timer { get; private set; }

        public bool CloseOnHeartbeatFailure
        {
            get { return closeOnHeartbeatFailure; }
            set { closeOnHeartbeatFailure = value; }
        }
        public double HeartbeatTimeout { get; set; }  // in seconds

        public bool AutoReconnect
        {
            get { return link.AutoReconnect; }
            set { link.AutoReconnect = value; }
        }
        public int MaxRetryCount  // 0 for unlimited
        {
            get { return link.MaxRetryCount; }
            set { link.MaxRetryCount = value; }
        }
        public long RetryInterval  // in millisec
        {
            get { return link.RetryInterval; }
            set { link.RetryInterval = value; }
        }

        public Action<Event, LinkSession> Preprocessor
        {
            get { return link.Preprocessor; }
            set { link.Preprocessor = value; }
        }

        public AsyncTcpClientFlow(string name)
            : this(name, new AsyncTcpClient(name))
        {
        }

        public AsyncTcpClientFlow(string name, AsyncTcpClient link)
        {
            this.link = link;
            Add(link);

            link.HeartbeatEventHandler = OnHeartbeatEvent;

            this.name = name;

            Resolution = Time.TicksInSecond;  // 1-second frame resolution
            HeartbeatTimeout = 15;

            Timer = new x2.Flows.Timer(OnTimer);
        }

        public void Close()
        {
            Timer.Cancel(link.Session.HeartbeatTimeoutToken);

            link.Close();
        }

        public void Connect(string host, int port)
        {
            link.Connect(host, port);
        }

        protected override void SetUp()
        {
            base.SetUp();

            Subscribe(new LinkSessionConnected { LinkName = Name }, OnLinkSessionConnected);
            Subscribe(new LinkSessionDisconnected { LinkName = Name }, OnLinkSessionDisconnected);
        }
{'fraction_non_alphanumeric': 0.07922304670449586, 'fraction_numerical': 0.00392841553906591, 'mean_word_length': 1.5941126521369942, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, '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': '2349847', 'n_tokens_mistral': 2364, 'n_tokens_neox': 2227, 'n_words': 547}