@Inject
public CQLSession(Configuration configuration)
{
    inet = InetSocketAddress.createUnresolved(configuration.getCassandraHost(), configuration.getCassandraPort());
    wlp = new WhiteListPolicy(new RoundRobinPolicy(), Collections.singletonList(inet));
    this.nettyOptions = new NettyOptions();
    this.queryOptions = new QueryOptions().setConsistencyLevel(ConsistencyLevel.ONE);
    this.reconnectionPolicy = new ExponentialReconnectionPolicy(1000,
                                                                configuration.getHealthCheckFrequencyMillis());
}
 

@VisibleForTesting
CQLSession(InetSocketAddress target, NettyOptions options)
{
    inet = target;
    wlp = new WhiteListPolicy(new RoundRobinPolicy(), Collections.singletonList(inet));
    this.nettyOptions = options;
    this.queryOptions = new QueryOptions().setConsistencyLevel(ConsistencyLevel.ONE);
    reconnectionPolicy = new ExponentialReconnectionPolicy(100, 1000);
}
 

/**
 * Currently we connect just once and then reuse the connection.
 * We do not bother with closing the connection.
 *
 * It is normal to use one Session per DB. The Session is thread safe.
 */
private void connect() {

    if (cluster == null) {

        log.info("Connecting to Cassandra server on " + this.dbHost + " at port " + this.dbPort);

        // allow fetching as much data as present in the DB
        QueryOptions queryOptions = new QueryOptions();
        queryOptions.setFetchSize(Integer.MAX_VALUE);
        queryOptions.setConsistencyLevel(ConsistencyLevel.ONE);

        cluster = Cluster.builder()
                         .addContactPoint(this.dbHost)
                         .withPort(this.dbPort)
                         .withLoadBalancingPolicy(new TokenAwarePolicy(new RoundRobinPolicy()))
                         .withReconnectionPolicy(new ExponentialReconnectionPolicy(500, 30000))
                         .withQueryOptions(queryOptions)
                         .withCredentials(this.dbUser, this.dbPassword)
                         .build();

    }

    if (session == null) {

        log.info("Connecting to Cassandra DB with name " + this.dbName);
        session = cluster.connect(dbName);
    }
}
 

@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 buildsPolicyWithDelayAndMaxInMillis() throws Exception {
    final ExponentialReconnectionPolicyFactory factory = new ExponentialReconnectionPolicyFactory();
    factory.setBaseDelay(Duration.seconds(4));
    factory.setMaxDelay(Duration.seconds(7));

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

    assertThat(policy.getBaseDelayMs()).isEqualTo(4000L);
    assertThat(policy.getMaxDelayMs()).isEqualTo(7000L);
}
 

@Override
public ReconnectionPolicy build() {
    return new ExponentialReconnectionPolicy(baseDelay.toMilliseconds(), maxDelay.toMilliseconds());
}
 

@Inject
public CassandraSessionImpl(@Named("cassandra.keyspace") String keyspace, @Named("cassandra.hostname") String hostname,
        @Named("cassandra.port") int port, @Named("cassandra.compression") String compression,
        @Named("cassandra.username") String username, @Named("cassandra.password") String password,
        @Named("cassandra.ssl") boolean ssl,
        @Named("cassandra.pool.core-connections-per-host") Integer coreConnectionsPerHost,
        @Named("cassandra.pool.max-connections-per-host") Integer maxConnectionsPerHost,
        @Named("cassandra.pool.max-requests-per-connection") Integer maxRequestsPerConnection) {

    checkNotNull(keyspace, "keyspace argument");
    checkNotNull(hostname, "hostname argument");
    checkArgument(port > 0 && port < 65535, "not a valid port number: %d", port);
    checkNotNull(compression, "compression argument");

    LOG.info("Setting up session with {}:{} using compression {}", hostname, port, compression.toUpperCase());

    final PoolingOptions poolingOptions = new PoolingOptions();
    if (coreConnectionsPerHost != null) {
        LOG.debug("Using {} core connections per host.", coreConnectionsPerHost);
        poolingOptions.setCoreConnectionsPerHost(HostDistance.LOCAL, coreConnectionsPerHost)
                .setCoreConnectionsPerHost(HostDistance.REMOTE, coreConnectionsPerHost);
    }
    if (maxConnectionsPerHost != null) {
        LOG.debug("Using {} max connections per host.", maxConnectionsPerHost);
        poolingOptions.setMaxConnectionsPerHost(HostDistance.LOCAL, maxConnectionsPerHost)
                .setMaxConnectionsPerHost(HostDistance.REMOTE, maxConnectionsPerHost);
    }
    if (maxRequestsPerConnection != null) {
        LOG.debug("Using {} max requests per connection.", maxRequestsPerConnection);
        poolingOptions.setMaxRequestsPerConnection(HostDistance.LOCAL, maxRequestsPerConnection)
                .setMaxRequestsPerConnection(HostDistance.REMOTE, maxRequestsPerConnection);
    }

    Builder builder = Cluster
            .builder()
            .withPort(port)
            .addContactPoints(hostname.split(","))
            .withReconnectionPolicy(new ExponentialReconnectionPolicy(1000, 2 * 60 * 1000))
            .withPoolingOptions(poolingOptions)
            .withCompression(Compression.valueOf(compression.toUpperCase()));

    if (username != null && password != null) {
        LOG.info("Using username: {} and password: XXXXXXXX", username);
        builder.withCredentials(username, password);
    }

    if (ssl) {
        LOG.info("Enabling SSL.");
        builder.withSSL();
    }

    m_session = builder.build().connect(keyspace);
}