Return cassandra client instance.

@param connection
the connection
@return the connection
765266
protected Cassandra.Client getConnection(Object connection)
    {
        if (connection != null)
        {
            return ((IPooledConnection) connection).getAPI();
        }

        throw new KunderaException("Invalid configuration!, no available pooled connection found for:"
                + this.getClass().getSimpleName());
    }
673162
public static synchronized BlockingCommandManager getInstanceFor(XMPPConnection connection) {
        BlockingCommandManager blockingCommandManager = INSTANCES.get(connection);

        if (blockingCommandManager == null) {
            blockingCommandManager = new BlockingCommandManager(connection);
            INSTANCES.put(connection, blockingCommandManager);
        }

        return blockingCommandManager;
    }
1013154
public static synchronized VCardManager getInstanceFor(XMPPConnection connection) {
        VCardManager vcardManager = INSTANCES.get(connection);
        if (vcardManager == null) {
            vcardManager = new VCardManager(connection);
            INSTANCES.put(connection, vcardManager);
        }
        return vcardManager;
    }
992270
@Override
    public Connection getConnection(RpcClient rpcClient, ClientTransportConfig transportConfig, Url url) {
        if (rpcClient == null || transportConfig == null || url == null) {
            return null;
        }
        Connection connection = urlConnectionMap.get(transportConfig);
        if (connection != null && !connection.isFine()) {
            closeConnection(rpcClient, transportConfig, url);
            connection = null;
        }
        if (connection == null) {
            try {
                connection = rpcClient.getConnection(url, url.getConnectTimeout());
            } catch (Exception e) {
                LOGGER.warn("get connection failed in url," + url);
            }
            if (connection == null) {
                return null;
            }
            // 保存唯一长连接
            Connection oldConnection = urlConnectionMap.putIfAbsent(transportConfig, connection);
            if (oldConnection != null) {
                if (LOGGER.isWarnEnabled()) {
                    LOGGER.warn("Multiple threads init ClientTransport with same key:" + url);
                }
                //only if new connection is not equals old connection,we can close it
                if (connection != oldConnection) {
                    rpcClient.closeStandaloneConnection(connection); //如果同时有人插入，则使用第一个
                    connection = oldConnection;
                }
            } else {

                // 增加计数器
                AtomicInteger counter = connectionRefCounter.get(connection);
                if (counter == null) {
                    counter = new AtomicInteger(0);
                    AtomicInteger oldCounter = connectionRefCounter.putIfAbsent(connection, counter);
                    if (oldCounter != null) {
                        counter = oldCounter;
                    }
                }
                int currentCount = counter.incrementAndGet();
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("Bolt client transport {} of {}, current ref count is: {}", url.toString(),
                        NetUtils.channelToString(connection.getLocalAddress(), connection.getRemoteAddress()),
                        currentCount);
                }
            }
        }
        return connection;
    }
1021295
public static synchronized ServiceDiscoveryManager getInstanceFor(XMPPConnection connection) {
        ServiceDiscoveryManager sdm = instances.get(connection);
        if (sdm == null) {
            sdm = new ServiceDiscoveryManager(connection);
            // Register the new instance and associate it with the connection
            instances.put(connection, sdm);
        }
        return sdm;
    }
1009159
Connection getConnection(ConnectionPool pool)
    {
        ConnectionPool connectionPool = pool;
        boolean success = false;
        while (!success)
        {
            try
            {
                success = true;
                Cassandra.Client client = connectionPool.getConnection();

                if (logger.isDebugEnabled())
                {
                    logger.debug("Returning connection of {} :{} .", pool.getPoolProperties().getHost(), pool
                            .getPoolProperties().getPort());
                }

                return new Connection(client, connectionPool);
            }
            catch (TException te)
            {
                success = false;
                logger.warn("{} :{}  host appears to be down, trying for next ", pool.getPoolProperties().getHost(),
                        pool.getPoolProperties().getPort());
                connectionPool = getNewPool(pool.getPoolProperties().getHost(), pool.getPoolProperties().getPort());
            }
        }

        throw new KunderaException("All hosts are down. please check servers manully.");
    }
376904
public static synchronized void getConnection(String poolName, Object mcp, Object cl, Object connection)
   {
      log.tracef("%s", new TraceEvent(poolName,
                                      Integer.toHexString(System.identityHashCode(mcp)),
                                      TraceEvent.GET_CONNECTION,
                                      Integer.toHexString(System.identityHashCode(cl)),
                                      Integer.toHexString(System.identityHashCode(connection))));
   }
1244819
public static synchronized DeliveryReceiptManager getInstanceFor(XMPPConnection connection) {
        DeliveryReceiptManager receiptManager = instances.get(connection);

        if (receiptManager == null) {
            receiptManager = new DeliveryReceiptManager(connection);
            instances.put(connection, receiptManager);
        }

        return receiptManager;
    }
120839
@Override
    protected Object createPoolOrConnection()
    {
        PersistenceUnitMetadata persistenceUnitMetadata = kunderaMetadata.getApplicationMetadata()
                .getPersistenceUnitMetadata(getPersistenceUnit());

        Properties props = persistenceUnitMetadata.getProperties();
        String keyspace = null;

        if (externalProperties != null)
        {
            keyspace = (String) externalProperties.get(PersistenceProperties.KUNDERA_KEYSPACE);
        }

        if (keyspace == null)
        {
            keyspace = (String) props.get(PersistenceProperties.KUNDERA_KEYSPACE);
        }

        for (Host host : ((CassandraHostConfiguration) configuration).getCassandraHosts())
        {
            PoolConfiguration prop = new PoolProperties();
            prop.setHost(host.getHost());
            prop.setPort(host.getPort());
            prop.setKeySpace(keyspace);

            CassandraUtilities.setPoolConfigPolicy((CassandraHost) host, prop);
            try
            {
                ConnectionPool pool = new ConnectionPool(prop);
                hostPools.put(host, pool);
            }
            catch (TException e)
            {
                logger.warn("Node {} are down, Caused by {} .", host.getHost(), e.getMessage());
                if (host.isRetryHost())
                {
                    logger.warn("Scheduling node for future retry");
                    ((CassandraRetryService) hostRetryService).add((CassandraHost) host);
                }
            }
        }
        return null;
    }
792203
public Connection getConnection(String name) throws ExecutionException {
        Connection connection = null;
        synchronized (connections) {
            if (connections.containsKey(name)) {
                connection = connections.get(name);
            } else {

                String msg = "The connection [" + name + "] does not exist";
                logger.error(msg);
                throw new ExecutionException(msg);
            }
        }
        return connection;
    }
73441
public <T> T execute(final String query, Object connection) {
        try {
            org.apache.cassandra.thrift.Cassandra.Client conn =
                (org.apache.cassandra.thrift.Cassandra.Client) connection;
            conn.set_cql_version(CassandraConstants.CQL_VERSION_3_0);
            KunderaCoreUtils.printQuery(query, showQuery);
            return (T) conn.execute_cql3_query(ByteBuffer.wrap(query.getBytes(Constants.CHARSET_UTF8)),
                Compression.NONE, getConsistencyLevel());
        } catch (Exception e) {
            log.error("Error while executing query {}", query);
            throw new KunderaException(e);
        }
    }
853271
public DataConnection getConnection() {
        DataConnection connection;

        for (connection = availableConnections.poll();
             connection != null && !connection.isUsable();
             connection = availableConnections.poll())
        {
            if (!connection.isUsable()) {
                connection.close();
            }
        }

        if (connection == null) {
            connection = new DataConnection(getDataUri());

            try {
                connection.connect();
            }
            catch (Exception e) {
                logger.error("[getConnection] could not connect to database", e);
                return null;
            }
        }

        connection.setLocale(threadLocale.getLocale());

        return connection;
    }
1188973
public <R> Connection<CL> getConnectionForOperation(BaseOperation<CL, R> baseOperation) {
	return selectionStrategy.getConnection(baseOperation, cpConfiguration.getMaxTimeoutWhenExhausted(),
		TimeUnit.MILLISECONDS);
    }
424546
public Cassandra.Client getRawClient(final String schema) {
        Cassandra.Client client = null;
        Object pooledConnection;
        pooledConnection = getConnection();
        client = (org.apache.cassandra.thrift.Cassandra.Client) getConnection(pooledConnection);
        try {
            client.set_cql_version(getCqlVersion());
        } catch (Exception e) {
            log.error("Error during borrowing a connection , Caused by: {}.", e);
            throw new KunderaException(e);
        } finally {
            releaseConnection(pooledConnection);
        }
        return client;

    }
540238
private static HttpEntity entityFromConnection(HttpURLConnection connection) {
        BasicHttpEntity entity = new BasicHttpEntity();
        InputStream inputStream;
        try {
            inputStream = connection.getInputStream();
        } catch (IOException ioe) {
            inputStream = connection.getErrorStream();
        }
        entity.setContent(inputStream);
        entity.setContentLength(connection.getContentLength());
        entity.setContentEncoding(connection.getContentEncoding());
        entity.setContentType(connection.getContentType());
        return entity;
    }
32200
public static PubSubManager getInstanceFor(XMPPConnection connection) {
    // CHECKSTYLE:ON:RegexpSingleline
        DomainBareJid pubSubService = null;
        if (connection.isAuthenticated()) {
            try {
                pubSubService = getPubSubService(connection);
            }
            catch (NoResponseException | XMPPErrorException | NotConnectedException e) {
                LOGGER.log(Level.WARNING, "Could not determine PubSub service", e);
            }
            catch (InterruptedException e) {
                LOGGER.log(Level.FINE, "Interrupted while trying to determine PubSub service", e);
            }
        }
        if (pubSubService == null) {
            try {
                // Perform an educated guess about what the PubSub service's domain bare JID may be
                pubSubService = JidCreate.domainBareFrom("pubsub." + connection.getXMPPServiceDomain());
            }
            catch (XmppStringprepException e) {
                throw new RuntimeException(e);
            }
        }
        return getInstanceFor(connection, pubSubService);
    }
985893
@Override
    protected Object createPoolOrConnection()
    {
        // AuthInfoProvider,LoadBalancingPolicy, ReconnectionPolicy,
        // RetryPolicy,ProtocolOptions.Compression, SSLOptions,
        // PoolingOptions,SocketOptions

        if (logger.isDebugEnabled())
        {
            logger.debug("Intiatilzing connection");
        }

        Properties connectionProperties = CassandraPropertyReader.csmd.getConnectionProperties();
        Builder connectionBuilder = Cluster.builder();

        // add host/port and AuthInfoProvider
        for (Host host : configuration.getCassandraHosts())
        {
            connectionBuilder.addContactPoint(host.getHost()).withPort(host.getPort());
            if (host.getUser() != null)
            {
                connectionBuilder.withCredentials(host.getUser(), host.getPassword());
            }
        }

        // add policy configuration
        String loadBalancingPolicy = connectionProperties.getProperty(Constants.LOADBALANCING_POLICY);
        if (!StringUtils.isBlank(loadBalancingPolicy))
        {
            LoadBalancingPolicy policy = getPolicyInstance(BalancingPolicy.getPolicy(loadBalancingPolicy),
                    connectionProperties);
            if (policy != null)
            {
                connectionBuilder.withLoadBalancingPolicy(policy);
            }
        }

        // compression
        String compression = connectionProperties.getProperty("compression");

        if (!StringUtils.isBlank(compression))
        {
            connectionBuilder.withCompression(Compression.valueOf(compression));
        }

        // ReconnectionPolicy

        String reconnectionPolicy = connectionProperties.getProperty("reconnection.policy");
        if (!StringUtils.isBlank(reconnectionPolicy))
        {
            com.datastax.driver.core.policies.ReconnectionPolicy policy = getPolicy(
                    ReconnectionPolicy.getPolicy(reconnectionPolicy), connectionProperties);

            if (policy != null)
            {
                connectionBuilder.withReconnectionPolicy(policy);
            }

        }

        // , RetryPolicy
        String retryPolicy = connectionProperties.getProperty("retry.policy");

        if (!StringUtils.isBlank(retryPolicy))
        {
            com.datastax.driver.core.policies.RetryPolicy policy = getPolicy(RetryPolicy.getPolicy(retryPolicy),
                    connectionProperties);

            if (policy != null)
            {
                connectionBuilder.withRetryPolicy(policy);
            }

        }

        // TODO::: SSLOptions? Not sure how to add it.

        // SocketOptions
        connectionBuilder.withSocketOptions(getSocketOptions(connectionProperties));

        // PoolingOptions,
        connectionBuilder.withPoolingOptions(getPoolingOptions(connectionProperties));

        // finally build cluster.
        Cluster cluster = connectionBuilder.build();

        PersistenceUnitMetadata persistenceUnitMetadata = kunderaMetadata.getApplicationMetadata()
                .getPersistenceUnitMetadata(getPersistenceUnit());
        Properties props = persistenceUnitMetadata.getProperties();

        if (externalProperties != null)
        {
            keyspace = (String) externalProperties.get(PersistenceProperties.KUNDERA_KEYSPACE);
        }

        if (keyspace == null)
        {
            keyspace = (String) props.get(PersistenceProperties.KUNDERA_KEYSPACE);
        }
        setSessionObject(cluster); // TODO custom session
        return cluster; // TODO custom cluster
    }