public static Session getClientSession(String hostAddr) {
    if(REGISTRY.containsKey(hostAddr)) {
        return REGISTRY.get(hostAddr);
    } else {
        Cluster.Builder clientClusterBuilder = new Cluster.Builder()
                .addContactPoint(hostAddr)
                .withQueryOptions(new QueryOptions()
                        .setConsistencyLevel(ConsistencyLevel.ONE)
                        .setSerialConsistencyLevel(ConsistencyLevel.LOCAL_SERIAL))
                .withoutJMXReporting()
                .withoutMetrics()
                .withReconnectionPolicy(new ConstantReconnectionPolicy(RECONNECT_DELAY_IN_MS));
        long startTimeInMillis = System.currentTimeMillis();
        Cluster clientCluster = clientClusterBuilder.build();
        Session clientSession = clientCluster.connect();

        LOG.info("Client session established after {} ms.",System.currentTimeMillis() - startTimeInMillis);
        REGISTRY.putIfAbsent(hostAddr,clientSession);
        return clientSession;
    }
}

public CassandraConfigDb(List<String> contactPoints,int port) {

    this.contactPoints = new ArrayList<InetAddress> (contactPoints.size());

    for (String contactPoint : contactPoints) {
        try {
            this.contactPoints.add(InetAddress.getByName(contactPoint));
        } catch (UnkNownHostException e) {
               throw new IllegalArgumentException(e.getMessage());
        }
    }

    this.port = port;

    cluster = (new Cluster.Builder()).withPort (this.port)
            .addContactPoints(this.contactPoints)
            .withSocketoptions(new Socketoptions().setReadTimeoutMillis(60000).setKeepAlive(true).setReuseAddress(true))
            .withLoadBalancingPolicy(new RoundRobinPolicy())
            .withReconnectionPolicy(new ConstantReconnectionPolicy(500L))
            .withQueryOptions(new QueryOptions().setConsistencyLevel(ConsistencyLevel.ONE))
            .build ();

    session = cluster.newSession();
    preparedStatements = new ConcurrentHashMap<StatementName,PreparedStatement> ();
    prepareStatementCreateLock = new Object();
}

public PersistentCassandraDrain( String username,String password,String[] seeds,String keyspace,RetentionConfiguration retentions ) {
    Cluster.Builder builder = Cluster.builder().addContactPoints( seeds );
    builder.withReconnectionPolicy( new ConstantReconnectionPolicy( 500 ) );
    if ( username != null ) {
        if ( password != null ) {
            builder = builder.withCredentials( username,password );
        } else {
            log.warn( "username was set,password was NOT set - IGnorING username!" );
        }
    }

    this.cluster = builder.build();
    this.keyspace = keyspace;
    this.session = cluster.connect( keyspace );
    this.retentions = retentions;
}

private static Cluster createCluster(CentralConfiguration centralConfig,TimestampGenerator defaultTimestampGenerator) {
    Cluster.Builder builder = Cluster.builder()
            .addContactPoints(
                    centralConfig.cassandraContactPoint().toArray(new String[0]))
            // aggressive reconnect policy seems ok since not many clients
            .withReconnectionPolicy(new ConstantReconnectionPolicy(1000))
            // let driver kNow that only idempotent queries are used so it will retry on timeout
            .withQueryOptions(new QueryOptions()
                    .setDefaultIdempotence(true)
                    .setConsistencyLevel(centralConfig.cassandraConsistencyLevel()))
            // central runs lots of parallel async queries and is very spiky since all
            // aggregates come in right after each minute marker
            .withPoolingOptions(
                    new PoolingOptions().setMaxQueueSize(Session.MAX_CONCURRENT_QUERIES))
            .withTimestampGenerator(defaultTimestampGenerator);
    String cassandraUsername = centralConfig.cassandraUsername();
    if (!cassandraUsername.isEmpty()) {
        // empty password is strange but valid
        builder.withCredentials(cassandraUsername,centralConfig.cassandraPassword());
    }
    return builder.build();
}

/**
 * Set the reconnection policy to define how often and in what interval to retry setup connections.
 * 
 * @param policy
 *            The reconnection policy as string "constant" {@link ConstantReconnectionPolicy} or "exponential"
 *            {@link ExponentialReconnectionPolicy}
 * @param delay
 *            The initial or constant delay
 * @param max
 *            The maximum delay (only required for {@link ExponentialReconnectionPolicy})
 */
public void setReconnectionPolicy(String policy,int delay,int... max) {
    switch(policy) {
        case "constant":
            setReconnectionPolicy(new ConstantReconnectionPolicy(delay));
            break;

        case "exponential":
            setReconnectionPolicy(new ExponentialReconnectionPolicy(delay,max[0]));
            break;

        default:
            setReconnectionPolicy(Policies.defaultReconnectionPolicy());
            break;
    }
}

@Test
public void cassandraSessionWithConfiguration() throws Exception {
    ApplicationContext testContext = getTestApplicationContext(
            "cloud-cassandra-with-config.xml",createService("my-service"));
    Cluster cluster = testContext.getBean("cassandra-full-config",getConnectorType());

    assertNotNull(cluster.getConfiguration().getSocketoptions());
    assertEquals(15000,cluster.getConfiguration().getSocketoptions().getConnectTimeoutMillis());
    assertTrue(DefaultRetryPolicy.class.isAssignableFrom(
            cluster.getConfiguration().getPolicies().getRetryPolicy().getClass()));
    assertTrue(RoundRobinPolicy.class.isAssignableFrom(cluster.getConfiguration()
            .getPolicies().getLoadBalancingPolicy().getClass()));
    assertTrue(ConstantReconnectionPolicy.class.isAssignableFrom(cluster
            .getConfiguration().getPolicies().getReconnectionPolicy().getClass()));
}

public void init() {
    LOG.info("Starting Cassandra connector initialization.");
    Cluster.Builder builder = Cluster.builder()
            .addContactPoints(hosts)
            .withReconnectionPolicy(new ConstantReconnectionPolicy(reconnectionDelayMs))
            .withRetryPolicy(DefaultRetryPolicy.INSTANCE)
            .withCompression(ProtocolOptions.Compression.LZ4)
            .withSocketoptions(new Socketoptions()
                    .setReceiveBufferSize(receiverBufferSize)
                    .setSendBufferSize(senderBufferSize))
            .withPort(port);
    if (poolingOptions != null) {
        int procs = Runtime.getRuntime().availableProcessors();
        poolingOptions
                .setConnectionsPerHost(Hostdistance.LOCAL,procs,procs * 2)
                .setConnectionsPerHost(Hostdistance.REMOTE,(procs / 2),procs * 2)
                .setPoolTimeoutMillis(poolTimeoutMillis)
                .setMaxRequestsPerConnection(Hostdistance.LOCAL,maxRequestsPerConnection)
                .setMaxRequestsPerConnection(Hostdistance.REMOTE,maxRequestsPerConnection);
        builder.withPoolingOptions(poolingOptions);
    }
    if (!Strings.isNullOrEmpty(username)) {
        builder.withCredentials(username,password);
    }
    cluster = builder.build();
    session = cluster.connect(keyspace);
}

public CassandraSupport() {
    cluster = Cluster.builder()
            .addContactPoint("127.0.0.1")
            .withReconnectionPolicy(new ConstantReconnectionPolicy(200))
            .build();

    createSchema();
    session = cluster.connect("ho");
}

@Test
public void testReconnectionPolicyParsing() throws Exception
{
    String retryPolicyStr = "ConstantReconnectionPolicy((long)10)";
    System.out.println(retryPolicyStr);
    assertTrue(Utils.parseReconnectionPolicy(retryPolicyStr) instanceof ConstantReconnectionPolicy);
    System.out.println("====================");
    retryPolicyStr = "ExponentialReconnectionPolicy((long)10,(Long)100)";
    System.out.println(retryPolicyStr);
    assertTrue(Utils.parseReconnectionPolicy(retryPolicyStr) instanceof ExponentialReconnectionPolicy);
    System.out.println("====================");

}

@Test
public void buildsPolicyWithDelayInMillis() throws Exception {
    final ConstantReconnectionPolicyFactory factory = new ConstantReconnectionPolicyFactory();
    factory.setDelay(Duration.seconds(5));

    final ConstantReconnectionPolicy policy = (ConstantReconnectionPolicy) factory.build();

    assertthat(policy.getConstantDelayMs()).isEqualTo(5000L);
}

public static synchronized Session createSession(String sessionKey,Set<InetAddress> host,String username,LoadBalancingPolicy loadBalancingPolicy) {

    instance = getInstance();
    Session session = instance.sessions.get(sessionKey);
      if (session == null) {

          Cluster.Builder cb = Cluster.builder()
                  .addContactPoints(host)
                  .withReconnectionPolicy(new ConstantReconnectionPolicy(10000)) ;

          if (loadBalancingPolicy != null ) {
              cb = cb.withLoadBalancingPolicy(loadBalancingPolicy);
          }

          if ( username != null && ! username.isEmpty()) {
              cb = cb.withCredentials(username,password);
          }

          Cluster cluster = cb.build();


          if (keyspace != null && !keyspace.isEmpty())
        session = cluster.connect(keyspace);
      else
        session = cluster.connect();

        instance.sessions.put(sessionKey,session);
    }
    return session;
  }

@Test
public void shouldCreateClusterWithConfig() throws Exception {

    CassandraServiceInfo info = new CassandraServiceInfo("local",Collections.singletonList("127.0.0.1"),9142);

    CassandraClusterConfig config = new CassandraClusterConfig();
    config.setCompression(ProtocolOptions.Compression.NONE);
    config.setPoolingOptions(new PoolingOptions().setPoolTimeoutMillis(1234));
    config.setQueryOptions(new QueryOptions());
    config.setProtocolVersion(ProtocolVersion.NEWEST_SUPPORTED);
    config.setLoadBalancingPolicy(new RoundRobinPolicy());
    config.setReconnectionPolicy(new ConstantReconnectionPolicy(1));
    config.setRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE);
    config.setSocketoptions(new Socketoptions());

    Cluster cluster = creator.create(info,config);

    Configuration configuration = cluster.getConfiguration();

    assertthat(configuration.getProtocolOptions().getCompression(),is(config.getCompression()));
    assertthat(configuration.getQueryOptions(),is(config.getQueryOptions()));
    assertthat(configuration.getSocketoptions(),is(config.getSocketoptions()));

    Policies policies = configuration.getPolicies();
    assertthat(policies.getLoadBalancingPolicy(),is(config.getLoadBalancingPolicy()));
    assertthat(policies.getReconnectionPolicy(),is(config.getReconnectionPolicy()));
    assertthat(policies.getRetryPolicy(),is(config.getRetryPolicy()));
}

@postconstruct
public void initialize() {
    String cassandraHosts = env.getProperty("ea.cassandra.hosts","localhost:9042");
    String cassandraClusterName = env.getProperty("ea.cassandra.cluster","ElasticActorsCluster");
    String cassandraKeyspaceName = env.getProperty("ea.cassandra.keyspace","\"ElasticActors\"");
    Integer cassandraPort = env.getProperty("ea.cassandra.port",Integer.class,9042);

    Set<String> hostSet = StringUtils.commaDelimitedListToSet(cassandraHosts);

    String[] contactPoints = new String[hostSet.size()];
    int i=0;
    for (String host : hostSet) {
        if(host.contains(":")) {
            contactPoints[i] = host.substring(0,host.indexOf(":"));
        } else {
            contactPoints[i] = host;
        }
        i+=1;
    }

    PoolingOptions poolingOptions = new PoolingOptions();
    poolingOptions.setHeartbeatIntervalSeconds(60);
    poolingOptions.setConnectionsPerHost(Hostdistance.LOCAL,2,env.getProperty("ea.cassandra.maxActive",Runtime.getRuntime().availableProcessors() * 3));
    poolingOptions.setPoolTimeoutMillis(2000);

    Cluster cassandraCluster =
            Cluster.builder().withClusterName(cassandraClusterName)
                    .addContactPoints(contactPoints)
                    .withPort(cassandraPort)
            .withLoadBalancingPolicy(new RoundRobinPolicy())
            .withRetryPolicy(new LoggingRetryPolicy(DefaultRetryPolicy.INSTANCE))
            .withPoolingOptions(poolingOptions)
            .withReconnectionPolicy(new ConstantReconnectionPolicy(env.getProperty("ea.cassandra.retryDownedHostsDelayInSeconds",1) * 1000))
            .withQueryOptions(new QueryOptions().setConsistencyLevel(ConsistencyLevel.QUORUM)).build();

    this.cassandraSession = cassandraCluster.connect(cassandraKeyspaceName);


}

private KeyspaceMetadata validate() {
    Cluster.Builder builder = Cluster.builder();

    System.out.printf("Trying connect to host \"%s\"" + Main.LINE_SEParaTOR,host);
    DatabaseValidator.validateHost(host,builder);
    System.out.println("Success!");

    //FIXME: check username and password
    builder.withPort(port)
           .withoutJMXReporting()
           .withoutMetrics()
           .withCredentials(username,password)
           .withReconnectionPolicy(new ConstantReconnectionPolicy(2000));

    if (secure)
        builder.withSSL();

    System.out.printf("Trying connect to port \"%s\" " + Main.LINE_SEParaTOR,port);
    Cluster cluster = DatabaseValidator.validateDatabasePort(host,port,username,password,builder);
    System.out.println("Success!");

    if (Main.VERBOSE) {
        QueryLogger queryLogger = QueryLogger.builder()
                                             .withConstantThreshold(1)
                                             .withMaxQueryStringLength(QueryLogger.DEFAULT_MAX_QUERY_STRING_LENGTH)
                                             .build();
        cluster.register(queryLogger);
    }

    Socketoptions socketoptions = cluster.getConfiguration().getSocketoptions();
    socketoptions.setConnectTimeoutMillis(10000);
    socketoptions.setReadTimeoutMillis(15000);
    socketoptions.setKeepAlive(true);

    System.out.printf("Trying connect to keyspace \"%s\"" + Main.LINE_SEParaTOR,keyspaceName);
    session = DatabaseValidator.validateKeyspace(cluster,keyspaceName);
    System.out.println("Success!");

    KeyspaceMetadata keyspace = cluster.getMetadata().getKeyspace(keyspaceName);

    if (!Strings.isNullOrEmpty(tableName)) {
        System.out.printf("Trying validate table \"%s\"" + Main.LINE_SEParaTOR,tableName);
        DatabaseValidator.validateTableName(tableName,keyspace);
        System.out.println("Success!");
    }
    return keyspace;
}

@Override
public ReconnectionPolicy build() {
    return new ConstantReconnectionPolicy(delay.toMilliseconds());
}