<commit_before>﻿//
//      Copyright (C) 2012 DataStax Inc.
//
//   Licensed under the Apache License, Version 2.0 (the "License");
//   you may not use this file except in compliance with the License.
//   You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
//   Unless required by applicable law or agreed to in writing, software
//   distributed under the License is distributed on an "AS IS" BASIS,
//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//   See the License for the specific language governing permissions and
//   limitations under the License.
//

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Security;
using System.Net.Sockets;
using System.Security.Cryptography.X509Certificates;
using System.Threading;

namespace Cassandra
{
    internal class CassandraConnection : IDisposable
    {

#if ERRORINJECTION
        public void KillSocket()
        {
            try
            {
                _socket.Shutdown(SocketShutdown.Both);
            }
            catch { }
        }
#endif
 
        private static readonly Logger Logger = new Logger(typeof (CassandraConnection));
        private static readonly FrameParser FrameParser = new FrameParser();
        private readonly BoolSwitch _alreadyDisposed = new BoolSwitch();
        private readonly IAuthInfoProvider _authInfoProvider;
        private readonly IAuthProvider _authProvider;
        private readonly byte[][] _buffer;
        private readonly IBuffering _bufferingMode;
        private readonly IProtoBufComporessor _compressor;

        private readonly AtomicValue<Action<ResponseFrame>> _frameEventCallback = new AtomicValue<Action<ResponseFrame>>(null);
        private readonly AtomicArray<AsyncResult<IOutput>> _frameReadAsyncResult = new AtomicArray<AsyncResult<IOutput>>(sbyte.MaxValue + 1);
        private readonly AtomicArray<Action<ResponseFrame>> _frameReadCallback = new AtomicArray<Action<ResponseFrame>>(sbyte.MaxValue + 1);
        private readonly AtomicArray<Timer> _frameReadTimers = new AtomicArray<Timer>(sbyte.MaxValue + 1);
        private readonly ConcurrentStack<int> _freeStreamIDs = new ConcurrentStack<int>();
        private readonly AtomicValue<bool> _isStreamOpened = new AtomicValue<bool>(false);

        private readonly ISession _owner;
        private readonly int _port;
        private readonly Action<ErrorActionParam> _protocolErrorHandlerAction;
        private readonly int _queryAbortTimeout = Timeout.Infinite;
        private readonly AutoResetEvent _readerSocketStreamBusy = new AutoResetEvent(true);
        private readonly IPAddress _serverAddress;
        private readonly Socket _socket;
        private readonly BoolSwitch _socketExceptionOccured = new BoolSwitch();

        private readonly SocketOptions _socketOptions;
        private readonly Stream _socketStream;

        private readonly Dictionary<string, string> _startupOptions = new Dictionary<string, string>
        {
            {"CQL_VERSION", "3.0.0"}
        };

        private readonly AtomicValue<string> _currentKs = new AtomicValue<string>("");
        private readonly ConcurrentDictionary<byte[], string> _preparedQueries = new ConcurrentDictionary<byte[], string>();
        private readonly AtomicValue<string> _selectedKs = new AtomicValue<string>("");

        internal Guid Guid;
        private volatile byte _binaryProtocolRequestVersionByte = RequestFrame.ProtocolV2RequestVersionByte;
        private volatile byte _binaryProtocolResponseVersionByte = ResponseFrame.ProtocolV2ResponseVersionByte;

        private int _bufNo;
        private Action<ResponseFrame> _defaultFatalErrorAction;

        public bool IsHealthy
        {
            get { return !_alreadyDisposed.IsTaken() && !_socketExceptionOccured.IsTaken(); }
        }

        internal CassandraConnection(ISession owner, IPAddress serverAddress, ProtocolOptions protocolOptions,
                                     SocketOptions socketOptions, ClientOptions clientOptions,
                                     IAuthProvider authProvider, IAuthInfoProvider authInfoProvider, int protocolVersion)
        {
            if (protocolVersion == 1)
            {
                _binaryProtocolRequestVersionByte = RequestFrame.ProtocolV1RequestVersionByte;
                _binaryProtocolResponseVersionByte = ResponseFrame.ProtocolV1ResponseVersionByte;
            }

            Guid = Guid.NewGuid();
            _owner = owner;
            _bufferingMode = null;
            switch (protocolOptions.Compression)
            {
                case CompressionType.LZ4:
                    _bufferingMode = new FrameBuffering();
                    break;
                case CompressionType.Snappy:
                    _bufferingMode = new FrameBuffering();
                    break;
                case CompressionType.NoCompression:
                    _bufferingMode = clientOptions.WithoutRowSetBuffering ? new NoBuffering() : new FrameBuffering();
                    break;
                default:
                    throw new ArgumentException();
            }

            _authProvider = authProvider;
            _authInfoProvider = authInfoProvider;
            if (protocolOptions.Compression == CompressionType.Snappy)
            {
                _startupOptions.Add("COMPRESSION", "snappy");
                _compressor = new SnappyProtoBufCompressor();
            }
            else if (protocolOptions.Compression == CompressionType.LZ4)
            {
                _startupOptions.Add("COMPRESSION", "lz4");
                _compressor = new LZ4ProtoBufCompressor();
            }

            _serverAddress = serverAddress;
            _port = protocolOptions.Port;
            _queryAbortTimeout = clientOptions.QueryAbortTimeout;

            _socketOptions = socketOptions;

            for (int i = 0; i <= sbyte.MaxValue; i++)
                _freeStreamIDs.Push(i);

            _protocolErrorHandlerAction = param =>
            {
                if (param.AbstractResponse is ErrorResponse)
                    JobFinished(
                        param.Jar,
                        (param.AbstractResponse as ErrorResponse).Output);
            };

            _frameEventCallback.Value = EventOccured;

            _buffer = new[]
            {
                new byte[_bufferingMode.PreferedBufferSize()],
                new byte[_bufferingMode.PreferedBufferSize()]
            };

            var newSock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);

            if (_socketOptions.KeepAlive != null)
                newSock.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive,
                                        _socketOptions.KeepAlive.Value);

            newSock.SendTimeout = _socketOptions.ConnectTimeoutMillis;

            if (_socketOptions.SoLinger != null)
                newSock.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.Linger,
                                        new LingerOption(true, _socketOptions.SoLinger.Value));

            if (_socketOptions.ReceiveBufferSize != null)
                newSock.ReceiveBufferSize = _socketOptions.ReceiveBufferSize.Value;

            if (_socketOptions.SendBufferSize != null)
                newSock.ReceiveBufferSize = _socketOptions.SendBufferSize.Value;

            if (_socketOptions.TcpNoDelay != null)
                newSock.NoDelay = _socketOptions.TcpNoDelay.Value;

            newSock.Connect(new IPEndPoint(_serverAddress, _port));
            _socket = newSock;
            _bufferingMode.Reset();

            if (protocolOptions.SslOptions == null)
                _socketStream = new NetworkStream(_socket);
            else
            {<commit_msg>Added comment<commit_after>﻿//
//      Copyright (C) 2012 DataStax Inc.
//
//   Licensed under the Apache License, Version 2.0 (the "License");
//   you may not use this file except in compliance with the License.
//   You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
//   Unless required by applicable law or agreed to in writing, software
//   distributed under the License is distributed on an "AS IS" BASIS,
//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//   See the License for the specific language governing permissions and
//   limitations under the License.
//

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Security;
using System.Net.Sockets;
using System.Security.Cryptography.X509Certificates;
using System.Threading;

namespace Cassandra
{
    internal class CassandraConnection : IDisposable
    {

#if ERRORINJECTION
        public void KillSocket()
        {
            try
            {
                _socket.Shutdown(SocketShutdown.Both);
            }
            catch { }
        }
#endif
 
        private static readonly Logger Logger = new Logger(typeof (CassandraConnection));
        private static readonly FrameParser FrameParser = new FrameParser();
        private readonly BoolSwitch _alreadyDisposed = new BoolSwitch();
        private readonly IAuthInfoProvider _authInfoProvider;
        private readonly IAuthProvider _authProvider;
        private readonly byte[][] _buffer;
        private readonly IBuffering _bufferingMode;
        private readonly IProtoBufComporessor _compressor;

        private readonly AtomicValue<Action<ResponseFrame>> _frameEventCallback = new AtomicValue<Action<ResponseFrame>>(null);
        private readonly AtomicArray<AsyncResult<IOutput>> _frameReadAsyncResult = new AtomicArray<AsyncResult<IOutput>>(sbyte.MaxValue + 1);
        private readonly AtomicArray<Action<ResponseFrame>> _frameReadCallback = new AtomicArray<Action<ResponseFrame>>(sbyte.MaxValue + 1);
        private readonly AtomicArray<Timer> _frameReadTimers = new AtomicArray<Timer>(sbyte.MaxValue + 1);
        private readonly ConcurrentStack<int> _freeStreamIDs = new ConcurrentStack<int>();
        private readonly AtomicValue<bool> _isStreamOpened = new AtomicValue<bool>(false);

        private readonly ISession _owner;
        private readonly int _port;
        private readonly Action<ErrorActionParam> _protocolErrorHandlerAction;
        private readonly int _queryAbortTimeout = Timeout.Infinite;
        private readonly AutoResetEvent _readerSocketStreamBusy = new AutoResetEvent(true);
        private readonly IPAddress _serverAddress;
        private readonly Socket _socket;
        private readonly BoolSwitch _socketExceptionOccured = new BoolSwitch();

        private readonly SocketOptions _socketOptions;
        private readonly Stream _socketStream;

        private readonly Dictionary<string, string> _startupOptions = new Dictionary<string, string>
        {
            {"CQL_VERSION", "3.0.0"}
        };

        private readonly AtomicValue<string> _currentKs = new AtomicValue<string>("");
        private readonly ConcurrentDictionary<byte[], string> _preparedQueries = new ConcurrentDictionary<byte[], string>();
        private readonly AtomicValue<string> _selectedKs = new AtomicValue<string>("");

        internal Guid Guid;
        private volatile byte _binaryProtocolRequestVersionByte = RequestFrame.ProtocolV2RequestVersionByte;
        private volatile byte _binaryProtocolResponseVersionByte = ResponseFrame.ProtocolV2ResponseVersionByte;

        private int _bufNo;
        private Action<ResponseFrame> _defaultFatalErrorAction;

        public bool IsHealthy
        {
            get { return !_alreadyDisposed.IsTaken() && !_socketExceptionOccured.IsTaken(); }
        }

        internal CassandraConnection(ISession owner, IPAddress serverAddress, ProtocolOptions protocolOptions,
                                     SocketOptions socketOptions, ClientOptions clientOptions,
                                     IAuthProvider authProvider, IAuthInfoProvider authInfoProvider, int protocolVersion)
        {
            if (protocolVersion == 1)
            {
                _binaryProtocolRequestVersionByte = RequestFrame.ProtocolV1RequestVersionByte;
                _binaryProtocolResponseVersionByte = ResponseFrame.ProtocolV1ResponseVersionByte;
            }

            Guid = Guid.NewGuid();
            _owner = owner;
            _bufferingMode = null;
            switch (protocolOptions.Compression)
            {
                case CompressionType.LZ4:
                    _bufferingMode = new FrameBuffering();
                    break;
                case CompressionType.Snappy:
                    _bufferingMode = new FrameBuffering();
                    break;
                case CompressionType.NoCompression:
                    _bufferingMode = clientOptions.WithoutRowSetBuffering ? new NoBuffering() : new FrameBuffering();
                    break;
                default:
                    throw new ArgumentException();
            }

            _authProvider = authProvider;
            _authInfoProvider = authInfoProvider;
            if (protocolOptions.Compression == CompressionType.Snappy)
            {
                _startupOptions.Add("COMPRESSION", "snappy");
                _compressor = new SnappyProtoBufCompressor();
            }
            else if (protocolOptions.Compression == CompressionType.LZ4)
            {
                _startupOptions.Add("COMPRESSION", "lz4");
                _compressor = new LZ4ProtoBufCompressor();
            }

            _serverAddress = serverAddress;
            _port = protocolOptions.Port;
            _queryAbortTimeout = clientOptions.QueryAbortTimeout;

            _socketOptions = socketOptions;

            for (int i = 0; i <= sbyte.MaxValue; i++)
                _freeStreamIDs.Push(i);

            _protocolErrorHandlerAction = param =>
            {
                if (param.AbstractResponse is ErrorResponse)
                    JobFinished(
                        param.Jar,
                        (param.AbstractResponse as ErrorResponse).Output);
            };

            _frameEventCallback.Value = EventOccured;

            _buffer = new[]
            {
                new byte[_bufferingMode.PreferedBufferSize()],
                new byte[_bufferingMode.PreferedBufferSize()]
            };

            var newSock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);

            if (_socketOptions.KeepAlive != null)
                newSock.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive,
                                        _socketOptions.KeepAlive.Value);

            newSock.SendTimeout = _socketOptions.ConnectTimeoutMillis;

            if (_socketOptions.SoLinger != null)
                newSock.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.Linger,
                                        new LingerOption(true, _socketOptions.SoLinger.Value));

            if (_socketOptions.ReceiveBufferSize != null)
                newSock.ReceiveBufferSize = _socketOptions.ReceiveBufferSize.Value;

            if (_socketOptions.SendBufferSize != null)
                newSock.ReceiveBufferSize = _socketOptions.SendBufferSize.Value;

            if (_socketOptions.TcpNoDelay != null)
                newSock.NoDelay = _socketOptions.TcpNoDelay.Value;


            var connectionResult = newSock.BeginConnect(_serverAddress, _port, null, null);
            connectionResult.AsyncWaitHandle.WaitOne(_socketOptions.ConnectTimeoutMillis);

            if (!newSock.Connected)
            {
                //The socket is still open
                newSock.Close();
                throw new SocketException((int)SocketError.TimedOut);
            }

            _socket = newSock;
            _bufferingMode.Reset();

            if (protocolOptions.SslOptions == null)
                _socketStream = new NetworkStream(_socket);
            else
            {
{'fraction_non_alphanumeric': 0.07132234477630438, 'fraction_numerical': 0.0033461395464121947, 'mean_word_length': 2.3105641025641024, 'pattern_counts': {'":': 0, '<': 61, '<?xml version=': 0, '>': 61, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7041905', 'n_tokens_mistral': 3972, 'n_tokens_neox': 3734, 'n_words': 938}