<commit_before>﻿using System;
using System.IO;
using System.Threading;
using Renci.SshNet.Channels;
using Renci.SshNet.Common;
using System.Collections.Generic;
using Renci.SshNet.Abstractions;

namespace Renci.SshNet
{
    /// <summary>
    /// Represents instance of the SSH shell object
    /// </summary>
    public class Shell : IDisposable
    {
        private readonly ISession _session;
        private IChannelSession _channel;
        private EventWaitHandle _channelClosedWaitHandle;
        private Stream _input;
        private readonly string _terminalName;
        private readonly uint _columns;
        private readonly uint _rows;
        private readonly uint _width;
        private readonly uint _height;
        private readonly IDictionary<TerminalModes, uint> _terminalModes;
        private EventWaitHandle _dataReaderTaskCompleted;
        private readonly Stream _outputStream;
        private readonly Stream _extendedOutputStream;
        private readonly int _bufferSize;

        /// <summary>
        /// Gets a value indicating whether this shell is started.
        /// </summary>
        /// <value>
        /// <c>true</c> if started is started; otherwise, <c>false</c>.
        /// </value>
        public bool IsStarted { get; private set; }

        /// <summary>
        /// Occurs when shell is starting.
        /// </summary>
        public event EventHandler<EventArgs> Starting;

        /// <summary>
        /// Occurs when shell is started.
        /// </summary>
        public event EventHandler<EventArgs> Started;

        /// <summary>
        /// Occurs when shell is stopping.
        /// </summary>
        public event EventHandler<EventArgs> Stopping;

        /// <summary>
        /// Occurs when shell is stopped.
        /// </summary>
        public event EventHandler<EventArgs> Stopped;

        /// <summary>
        /// Occurs when an error occurred.
        /// </summary>
        public event EventHandler<ExceptionEventArgs> ErrorOccurred;

        /// <summary>
        /// Initializes a new instance of the <see cref="Shell"/> class.
        /// </summary>
        /// <param name="session">The session.</param>
        /// <param name="input">The input.</param>
        /// <param name="output">The output.</param>
        /// <param name="extendedOutput">The extended output.</param>
        /// <param name="terminalName">Name of the terminal.</param>
        /// <param name="columns">The columns.</param>
        /// <param name="rows">The rows.</param>
        /// <param name="width">The width.</param>
        /// <param name="height">The height.</param>
        /// <param name="terminalModes">The terminal modes.</param>
        /// <param name="bufferSize">Size of the buffer for output stream.</param>
        internal Shell(ISession session, Stream input, Stream output, Stream extendedOutput, string terminalName, uint columns, uint rows, uint width, uint height, IDictionary<TerminalModes, uint> terminalModes, int bufferSize)
        {
            _session = session;
            _input = input;
            _outputStream = output;
            _extendedOutputStream = extendedOutput;
            _terminalName = terminalName;
            _columns = columns;
            _rows = rows;
            _width = width;
            _height = height;
            _terminalModes = terminalModes;
            _bufferSize = bufferSize;
        }

        /// <summary>
        /// Starts this shell.
        /// </summary>
        /// <exception cref="SshException">Shell is started.</exception>
        public void Start()
        {
            if (IsStarted)
            {
                throw new SshException("Shell is started.");
            }

            if (Starting != null)
            {
                Starting(this, new EventArgs());
            }

            _channel = _session.CreateChannelSession();
            _channel.DataReceived += Channel_DataReceived;
            _channel.ExtendedDataReceived += Channel_ExtendedDataReceived;
            _channel.Closed += Channel_Closed;
            _session.Disconnected += Session_Disconnected;
            _session.ErrorOccured += Session_ErrorOccured;

            _channel.Open();
            _channel.SendPseudoTerminalRequest(_terminalName, _columns, _rows, _width, _height, _terminalModes);
            _channel.SendShellRequest();

            _channelClosedWaitHandle = new AutoResetEvent(false);

            //  Start input stream listener
            _dataReaderTaskCompleted = new ManualResetEvent(false);
            ThreadAbstraction.ExecuteThread(() =>
            {
                try
                {
                    var buffer = new byte[_bufferSize];

                    while (_channel.IsOpen)
                    {
#if FEATURE_STREAM_TAP
                        var readTask = _input.ReadAsync(buffer, 0, buffer.Length);
                        var readWaitHandle = ((IAsyncResult) readTask).AsyncWaitHandle;

                        if (WaitHandle.WaitAny(new[] {readWaitHandle, _channelClosedWaitHandle}) == 0)
                        {
                            var read = readTask.Result;
                            _channel.SendData(buffer, 0, read);
                            continue;
                        }
#elif FEATURE_STREAM_APM
                        var asyncResult = _input.BeginRead(buffer, 0, buffer.Length, delegate(IAsyncResult result)
                        {
                            //  If input stream is closed and disposed already dont finish reading the stream
                            if (_input == null)
                                return;

                            var read = _input.EndRead(result);
                            if (read > 0)
                            {
                                _channel.SendData(buffer, 0, read);
                            }

                        }, null);

                        WaitHandle.WaitAny(new[] { asyncResult.AsyncWaitHandle, _channelClosedWaitHandle });

                        if (asyncResult.IsCompleted)
                            continue;
#else
                        #error Async receive is not implemented.
#endif
                        break;
                    }
                }
                catch (Exception exp)
                {
                    RaiseError(new ExceptionEventArgs(exp));
                }<commit_msg>Sync APM and TAP implementation.<commit_after>﻿using System;
using System.IO;
using System.Threading;
using Renci.SshNet.Channels;
using Renci.SshNet.Common;
using System.Collections.Generic;
using Renci.SshNet.Abstractions;

namespace Renci.SshNet
{
    /// <summary>
    /// Represents instance of the SSH shell object
    /// </summary>
    public class Shell : IDisposable
    {
        private readonly ISession _session;
        private IChannelSession _channel;
        private EventWaitHandle _channelClosedWaitHandle;
        private Stream _input;
        private readonly string _terminalName;
        private readonly uint _columns;
        private readonly uint _rows;
        private readonly uint _width;
        private readonly uint _height;
        private readonly IDictionary<TerminalModes, uint> _terminalModes;
        private EventWaitHandle _dataReaderTaskCompleted;
        private readonly Stream _outputStream;
        private readonly Stream _extendedOutputStream;
        private readonly int _bufferSize;

        /// <summary>
        /// Gets a value indicating whether this shell is started.
        /// </summary>
        /// <value>
        /// <c>true</c> if started is started; otherwise, <c>false</c>.
        /// </value>
        public bool IsStarted { get; private set; }

        /// <summary>
        /// Occurs when shell is starting.
        /// </summary>
        public event EventHandler<EventArgs> Starting;

        /// <summary>
        /// Occurs when shell is started.
        /// </summary>
        public event EventHandler<EventArgs> Started;

        /// <summary>
        /// Occurs when shell is stopping.
        /// </summary>
        public event EventHandler<EventArgs> Stopping;

        /// <summary>
        /// Occurs when shell is stopped.
        /// </summary>
        public event EventHandler<EventArgs> Stopped;

        /// <summary>
        /// Occurs when an error occurred.
        /// </summary>
        public event EventHandler<ExceptionEventArgs> ErrorOccurred;

        /// <summary>
        /// Initializes a new instance of the <see cref="Shell"/> class.
        /// </summary>
        /// <param name="session">The session.</param>
        /// <param name="input">The input.</param>
        /// <param name="output">The output.</param>
        /// <param name="extendedOutput">The extended output.</param>
        /// <param name="terminalName">Name of the terminal.</param>
        /// <param name="columns">The columns.</param>
        /// <param name="rows">The rows.</param>
        /// <param name="width">The width.</param>
        /// <param name="height">The height.</param>
        /// <param name="terminalModes">The terminal modes.</param>
        /// <param name="bufferSize">Size of the buffer for output stream.</param>
        internal Shell(ISession session, Stream input, Stream output, Stream extendedOutput, string terminalName, uint columns, uint rows, uint width, uint height, IDictionary<TerminalModes, uint> terminalModes, int bufferSize)
        {
            _session = session;
            _input = input;
            _outputStream = output;
            _extendedOutputStream = extendedOutput;
            _terminalName = terminalName;
            _columns = columns;
            _rows = rows;
            _width = width;
            _height = height;
            _terminalModes = terminalModes;
            _bufferSize = bufferSize;
        }

        /// <summary>
        /// Starts this shell.
        /// </summary>
        /// <exception cref="SshException">Shell is started.</exception>
        public void Start()
        {
            if (IsStarted)
            {
                throw new SshException("Shell is started.");
            }

            if (Starting != null)
            {
                Starting(this, new EventArgs());
            }

            _channel = _session.CreateChannelSession();
            _channel.DataReceived += Channel_DataReceived;
            _channel.ExtendedDataReceived += Channel_ExtendedDataReceived;
            _channel.Closed += Channel_Closed;
            _session.Disconnected += Session_Disconnected;
            _session.ErrorOccured += Session_ErrorOccured;

            _channel.Open();
            _channel.SendPseudoTerminalRequest(_terminalName, _columns, _rows, _width, _height, _terminalModes);
            _channel.SendShellRequest();

            _channelClosedWaitHandle = new AutoResetEvent(false);

            //  Start input stream listener
            _dataReaderTaskCompleted = new ManualResetEvent(false);
            ThreadAbstraction.ExecuteThread(() =>
            {
                try
                {
                    var buffer = new byte[_bufferSize];

                    while (_channel.IsOpen)
                    {
#if FEATURE_STREAM_TAP
                        var readTask = _input.ReadAsync(buffer, 0, buffer.Length);
                        var readWaitHandle = ((IAsyncResult) readTask).AsyncWaitHandle;

                        if (WaitHandle.WaitAny(new[] {readWaitHandle, _channelClosedWaitHandle}) == 0)
                        {
                            var read = readTask.Result;
                            _channel.SendData(buffer, 0, read);
                            continue;
                        }
#elif FEATURE_STREAM_APM
                        var asyncResult = _input.BeginRead(buffer, 0, buffer.Length, result =>
                            {
                                //  If input stream is closed and disposed already don't finish reading the stream
                                if (_input == null)
                                    return;

                                var read = _input.EndRead(result);
                                _channel.SendData(buffer, 0, read);
                            }, null);

                        WaitHandle.WaitAny(new[] { asyncResult.AsyncWaitHandle, _channelClosedWaitHandle });

                        if (asyncResult.IsCompleted)
                            continue;
#else
                        #error Async receive is not implemented.
#endif
                        break;
                    }
                }
                catch (Exception exp)
                {
                    RaiseError(new ExceptionEventArgs(exp));
                }
{'fraction_non_alphanumeric': 0.09563718962951359, 'fraction_numerical': 0.0008615963029685909, 'mean_word_length': 1.5820020222446916, 'pattern_counts': {'":': 0, '<': 115, '<?xml version=': 0, '>': 119, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5276215', 'n_tokens_mistral': 3082, 'n_tokens_neox': 2936, 'n_words': 808}