Java 类com.datastax.driver.core.policies.ConstantSpeculativeExecutionPolicy 实例源码

@Test
public void buildsPolicyWithConfiguredValues() throws Exception {
    final ConstantSpeculativeExecutionPolicyFactory factory = new ConstantSpeculativeExecutionPolicyFactory();
    factory.setDelay(Duration.seconds(5));
    factory.setMaxSpeculativeExecutions(2);

    final SpeculativeExecutionPolicy policy = factory.build();

    assertThat(policy).isExactlyInstanceOf(ConstantSpeculativeExecutionPolicy.class);
}

/**
 * Creates a {@link SessionManager} instance. Sub-class my override this
 * method to customized its own {@link SessionManager}.
 * 
 * @return
 */
protected SessionManager createSessionManager() {
    SessionManager sm = new SessionManager();

    // sm.setRetryPolicy(new
    // LoggingRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE));
    sm.setRetryPolicy(new LoggingRetryPolicy(DefaultRetryPolicy.INSTANCE));
    sm.setSpeculativeExecutionPolicy(new ConstantSpeculativeExecutionPolicy(10000, 3));

    sm.init();
    return sm;
}

public CassandraSessionManaged build(Environment environment, String localDc) {
    PoolingOptions poolingOptions = new PoolingOptions();
    poolingOptions.setConnectionsPerHost(HostDistance.LOCAL,  3, 5)
        .setConnectionsPerHost(HostDistance.REMOTE, 1, 2);
    final DCAwareRoundRobinPolicy.Builder builder = DCAwareRoundRobinPolicy.builder();
    if (localDc != null) {
        builder.withLocalDc(localDc);
    }
    QueryOptions queryOptions = new QueryOptions();
    queryOptions.setConsistencyLevel(ConsistencyLevel.LOCAL_ONE).setSerialConsistencyLevel(ConsistencyLevel.LOCAL_SERIAL);
    final Cluster cluster = Cluster
        .builder()
        .withRetryPolicy(DefaultRetryPolicy.INSTANCE)
        .withReconnectionPolicy(new ExponentialReconnectionPolicy(10L, 1000L))
        .withQueryOptions(queryOptions)
        .withLoadBalancingPolicy(new TokenAwarePolicy(builder.build()))
        .addContactPoints(getContactPoints().stream().toArray(String[]::new))
        .withPort(getPort())
        .withSpeculativeExecutionPolicy(new ConstantSpeculativeExecutionPolicy(1000, 2))
        .withPoolingOptions(poolingOptions)
        .build();

    cluster.getConfiguration().getCodecRegistry()
        .register(InstantCodec.instance);

    Session session = cluster.connect(getKeySpace());

    CassandraSessionManaged cassandraSessionManaged = new CassandraSessionManaged(cluster, session);
    environment.lifecycle().manage(cassandraSessionManaged);

    return cassandraSessionManaged;
}

@Override
public SpeculativeExecutionPolicy build() {
    return new ConstantSpeculativeExecutionPolicy(delay.toMilliseconds(), maxSpeculativeExecutions);
}