<commit_before>﻿// 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.

namespace CassandraSharp
{
    using System;
    using System.IO;
    using System.Net.Sockets;
    using System.Threading;
    using Apache.Cassandra;
    using CassandraSharp.EndpointStrategy;
    using CassandraSharp.Utils;
    using Thrift.Transport;

    internal class Cluster : ICluster
    {
        private readonly IEndpointStrategy _endpointsManager;

        private readonly ILog _logger;

        private readonly ICluster _parentCluster;

        private readonly IPool<IConnection> _pool;

        private readonly IRecoveryService _recoveryService;

        private readonly ITransportFactory _transportFactory;

        public Cluster(IBehaviorConfig behaviorConfig, IPool<IConnection> pool, ITransportFactory transportFactory, IEndpointStrategy endpointsManager,
                       IRecoveryService recoveryService, ITimestampService timestampService, ILog logger)
            : this(behaviorConfig, pool, transportFactory, endpointsManager, recoveryService, timestampService, logger, null)
        {
        }

        private Cluster(IBehaviorConfig behaviorConfig, IPool<IConnection> pool, ITransportFactory transportFactory, IEndpointStrategy endpointsManager,
                        IRecoveryService recoveryService, ITimestampService timestampService, ILog logger, ICluster parentCluster)
        {
            BehaviorConfig = behaviorConfig;
            TimestampService = timestampService;
            _pool = pool;
            _endpointsManager = endpointsManager;
            _recoveryService = recoveryService;
            _transportFactory = transportFactory;
            _logger = logger;
            _parentCluster = parentCluster;
        }

        public void Dispose()
        {
            if (null == _parentCluster)
            {
                _pool.SafeDispose();
            }
        }

        public IBehaviorConfig BehaviorConfig { get; private set; }

        public ITimestampService TimestampService { get; private set; }

        public TResult ExecuteCommand<TResult>(Func<IConnection, TResult> func, Func<byte[]> keyFunc)
        {
            int tryCount = 1;
            while (true)
            {
                IConnection connection = null;
                try
                {
                    byte[] key = null;
                    if (null != keyFunc)
                    {
                        key = keyFunc();
                    }

                    connection = AcquireConnection(key);

                    TResult res = func(connection);

                    ReleaseConnection(connection, false);

                    return res;
                }
                catch (Exception ex)
                {
                    bool connectionDead;
                    bool retry;
                    DecipherException(ex, out connectionDead, out retry);

                    _logger.Error("Exception during command processing: connectionDead={0} retry={1} : {2}", connectionDead, retry, ex.Message);

                    ReleaseConnection(connection, connectionDead);
                    if (!retry || tryCount >= BehaviorConfig.MaxRetries)
                    {
                        _logger.Fatal("Max retry count reached");
                        throw;
                    }

                    if (BehaviorConfig.SleepBeforeRetry.HasValue)
                    {
                        Thread.Sleep(BehaviorConfig.SleepBeforeRetry.Value);
                    }
                }

                ++tryCount;
            }
        }

        public ICluster CreateChildCluster(IBehaviorConfig cfgOverride)
        {
            IBehaviorConfig childConfig = cfgOverride.Override(BehaviorConfig);
            return new Cluster(childConfig, _pool, _transportFactory, _endpointsManager, _recoveryService, TimestampService, _logger, this);
        }

        public IConnection AcquireConnection(byte[] key)
        {
            IConnection connection;
            if (_pool.Acquire(out connection))
            {
                return connection;
            }

            Endpoint endpoint = _endpointsManager.Pick(key);
            if (null == endpoint)
            {
                throw new ArgumentException("Can't find any valid endpoint");
            }

            Cassandra.Client client = _transportFactory.Create(endpoint.Address);
            connection = new Connection(client, endpoint);

            ChangeKeyspace(connection);

            return connection;
        }

        public void ReleaseConnection(IConnection connection, bool hasFailed)
        {
            // protect against exception during acquire connection
            if (null != connection)
            {
                if (hasFailed)
                {
                    if (null != _recoveryService)
                    {
                        _logger.Info("marking {0} for recovery", connection.Endpoint.Address);
                        _recoveryService.Recover(connection.Endpoint, _transportFactory, ClientRecoveredCallback);
                    }

                    _endpointsManager.Ban(connection.Endpoint);
                    connection.SafeDispose();
                }
                else
                {
                    _pool.Release(connection);
                }
            }
        }

        private void ChangeKeyspace(IConnection connection)
        {
            bool keyspaceChanged = connection.KeySpace != BehaviorConfig.KeySpace;
            if (keyspaceChanged && null != BehaviorConfig.KeySpace)
            {
                connection.CassandraClient.set_keyspace(BehaviorConfig.KeySpace);
                connection.KeySpace = BehaviorConfig.KeySpace;
            }
        }

        private void DecipherException(Exception ex, out bool connectionDead, out bool retry)
        {
            // connection dead exception handling
            if (ex is TTransportException)
            {
                connectionDead = true;
                retry = true;
            }
            else if (ex is IOException)
            {
                connectionDead = true;
                retry = true;
            }
            else if (ex is SocketException)
            {
                connectionDead = true;
                retry = true;
            }

                // functional exception handling
            else if (ex is TimedOutException)
            {
                connectionDead = false;
                retry = BehaviorConfig.RetryOnTimeout ?? false;
            }
            else if (ex is UnavailableException)<commit_msg>fixed keyspace change on existing connections<commit_after>﻿// 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.

namespace CassandraSharp
{
    using System;
    using System.IO;
    using System.Net.Sockets;
    using System.Threading;
    using Apache.Cassandra;
    using CassandraSharp.EndpointStrategy;
    using CassandraSharp.MadeSimple;
    using CassandraSharp.Utils;
    using Thrift.Transport;

    internal class Cluster : ICluster
    {
        private readonly IEndpointStrategy _endpointsManager;

        private readonly ILog _logger;

        private readonly ICluster _parentCluster;

        private readonly IPool<IConnection> _pool;

        private readonly IRecoveryService _recoveryService;

        private readonly ITransportFactory _transportFactory;

        public Cluster(IBehaviorConfig behaviorConfig, IPool<IConnection> pool, ITransportFactory transportFactory, IEndpointStrategy endpointsManager,
                       IRecoveryService recoveryService, ITimestampService timestampService, ILog logger)
            : this(behaviorConfig, pool, transportFactory, endpointsManager, recoveryService, timestampService, logger, null)
        {
        }

        private Cluster(IBehaviorConfig behaviorConfig, IPool<IConnection> pool, ITransportFactory transportFactory, IEndpointStrategy endpointsManager,
                        IRecoveryService recoveryService, ITimestampService timestampService, ILog logger, ICluster parentCluster)
        {
            BehaviorConfig = behaviorConfig;
            TimestampService = timestampService;
            _pool = pool;
            _endpointsManager = endpointsManager;
            _recoveryService = recoveryService;
            _transportFactory = transportFactory;
            _logger = logger;
            _parentCluster = parentCluster;
        }

        public void Dispose()
        {
            if (null == _parentCluster)
            {
                _pool.SafeDispose();
            }
        }

        public IBehaviorConfig BehaviorConfig { get; private set; }

        public ITimestampService TimestampService { get; private set; }

        public TResult ExecuteCommand<TResult>(Func<IConnection, TResult> func, Func<byte[]> keyFunc)
        {
            int tryCount = 1;
            while (true)
            {
                IConnection connection = null;
                try
                {
                    byte[] key = null;
                    if (null != keyFunc)
                    {
                        key = keyFunc();
                    }

                    connection = AcquireConnection(key);

                    TResult res = func(connection);

                    ReleaseConnection(connection, false);

                    return res;
                }
                catch (Exception ex)
                {
                    bool connectionDead;
                    bool retry;
                    DecipherException(ex, out connectionDead, out retry);

                    _logger.Error("Exception during command processing: connectionDead={0} retry={1} : {2}", connectionDead, retry, ex.Message);

                    ReleaseConnection(connection, connectionDead);
                    if (!retry || tryCount >= BehaviorConfig.MaxRetries)
                    {
                        _logger.Fatal("Max retry count reached");
                        throw;
                    }

                    if (BehaviorConfig.SleepBeforeRetry.HasValue)
                    {
                        Thread.Sleep(BehaviorConfig.SleepBeforeRetry.Value);
                    }
                }

                ++tryCount;
            }
        }

        public ICluster CreateChildCluster(IBehaviorConfig cfgOverride)
        {
            IBehaviorConfig childConfig = cfgOverride.Override(BehaviorConfig);
            return new Cluster(childConfig, _pool, _transportFactory, _endpointsManager, _recoveryService, TimestampService, _logger, this);
        }

        public IConnection AcquireConnection(byte[] key)
        {
            IConnection connection;
            if (!_pool.Acquire(out connection))
            {
                Endpoint endpoint = _endpointsManager.Pick(key);
                if (null == endpoint)
                {
                    throw new ArgumentException("Can't find any valid endpoint");
                }

                Cassandra.Client client = _transportFactory.Create(endpoint.Address);
                connection = new Connection(client, endpoint);
            }

            ChangeKeyspace(connection);

            return connection;
        }

        public void ReleaseConnection(IConnection connection, bool hasFailed)
        {
            // protect against exception during acquire connection
            if (null != connection)
            {
                if (hasFailed)
                {
                    if (null != _recoveryService)
                    {
                        _logger.Info("marking {0} for recovery", connection.Endpoint.Address);
                        _recoveryService.Recover(connection.Endpoint, _transportFactory, ClientRecoveredCallback);
                    }

                    _endpointsManager.Ban(connection.Endpoint);
                    connection.SafeDispose();
                }
                else
                {
                    _pool.Release(connection);
                }
            }
        }

        private void ChangeKeyspace(IConnection connection)
        {
            bool keyspaceChanged = connection.KeySpace != BehaviorConfig.KeySpace;
            if (keyspaceChanged && null != BehaviorConfig.KeySpace)
            {
                connection.CassandraClient.set_keyspace(BehaviorConfig.KeySpace);
                connection.KeySpace = BehaviorConfig.KeySpace;

                // UNDONE: CQL3 seems to be badly broken - USE does not work actually
                //string useKeyspace = string.Format("use {0}", connection.KeySpace);
                //byte[] bufUseKeyspace = new Utf8NameOrValue(useKeyspace).ToByteArray();
                //connection.CassandraClient.execute_cql_query(bufUseKeyspace, Compression.NONE);
            }
        }

        private void DecipherException(Exception ex, out bool connectionDead, out bool retry)
        {
            // connection dead exception handling
            if (ex is TTransportException)
            {
                connectionDead = true;
                retry = true;
            }
            else if (ex is IOException)
            {
                connectionDead = true;
                retry = true;
            }
            else if (ex is SocketException)
            {
                connectionDead = true;
                retry = true;
            }

                // functional exception handling
            else if (ex is TimedOutException)
            {
                connectionDead = false;
                retry = BehaviorConfig.RetryOnTimeout ?? false;
            }
            else if (ex is UnavailableException)
{'fraction_non_alphanumeric': 0.06614892022617344, 'fraction_numerical': 0.001430615164520744, 'mean_word_length': 1.6265879405976025, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '7555862', 'n_tokens_mistral': 3359, 'n_tokens_neox': 3112, 'n_words': 879}