Spring/spring-data-cassandra
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATACASS-637 - General code maintenance.

Browse Source
This commit is contained in:
John Blum
2019-03-11 13:38:01 -07:00
parent e295e615bd
commit d967447b6e
24 changed files with 849 additions and 677 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractCassandraConfiguration.java
View File

@@ -33,6 +33,7 @@ import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
import org.springframework.data.cassandra.core.mapping.SimpleTupleTypeFactory;
import org.springframework.data.cassandra.core.mapping.SimpleUserTypeResolver;
import org.springframework.data.cassandra.core.mapping.Table;
import org.springframework.data.cassandra.core.mapping.UserTypeResolver;
import org.springframework.data.convert.CustomConversions;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.lang.Nullable;
@@ -146,8 +147,10 @@ public abstract class AbstractCassandraConfiguration extends AbstractClusterConf
Cluster cluster = getRequiredCluster();
CassandraMappingContext mappingContext = new CassandraMappingContext(
new SimpleUserTypeResolver(cluster, getKeyspaceName()), new SimpleTupleTypeFactory(cluster));
UserTypeResolver userTypeResolver = new SimpleUserTypeResolver(cluster, getKeyspaceName());
CassandraMappingContext mappingContext =
new CassandraMappingContext(userTypeResolver, new SimpleTupleTypeFactory(cluster));
Optional.ofNullable(this.beanClassLoader).ifPresent(mappingContext::setBeanClassLoader);

7
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractClusterConfiguration.java
View File

@@ -60,9 +60,12 @@ public abstract class AbstractClusterConfiguration {
protected Cluster getRequiredCluster() {
CassandraClusterFactoryBean factoryBean = cluster();
Assert.state(factoryBean.getObject() != null, "Cluster factory not initialized");
return factoryBean.getObject();
Cluster cluster = factoryBean.getObject();
Assert.state(cluster != null, "Cluster not initialized");
return cluster;
}
/**

374
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CassandraClusterFactoryBean.java
View File

@@ -41,15 +41,27 @@ import org.springframework.data.cassandra.core.cql.keyspace.AlterKeyspaceSpecifi
import org.springframework.data.cassandra.core.cql.keyspace.CreateKeyspaceSpecification;
import org.springframework.data.cassandra.core.cql.keyspace.DropKeyspaceSpecification;
import org.springframework.data.cassandra.core.cql.keyspace.KeyspaceActionSpecification;
import org.springframework.lang.NonNull;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import com.datastax.driver.core.*;
import com.datastax.driver.core.AuthProvider;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Cluster.Builder;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.LatencyTracker;
import com.datastax.driver.core.NettyOptions;
import com.datastax.driver.core.PoolingOptions;
import com.datastax.driver.core.ProtocolOptions.Compression;
import com.datastax.driver.core.ProtocolVersion;
import com.datastax.driver.core.QueryOptions;
import com.datastax.driver.core.SSLOptions;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SocketOptions;
import com.datastax.driver.core.TimestampGenerator;
import com.datastax.driver.core.policies.AddressTranslator;
import com.datastax.driver.core.policies.LoadBalancingPolicy;
import com.datastax.driver.core.policies.ReconnectionPolicy;
@@ -71,20 +83,20 @@ import com.datastax.driver.core.policies.SpeculativeExecutionPolicy;
* <strong>XML configuration</strong>
*
* <pre class="code">
<cql:cluster contact-points="…"
port="${build.cassandra.native_transport_port}" compression="SNAPPY" netty-options-ref="nettyOptions">
<cql:local-pooling-options
min-simultaneous-requests="26" max-simultaneous-requests="101"
core-connections="3" max-connections="9"/>
<cql:remote-pooling-options
min-simultaneous-requests="25" max-simultaneous-requests="100"
core-connections="1" max-connections="2"/>
<cql:socket-options connect-timeout-millis="5000"
keep-alive="true" reuse-address="true" so-linger="60" tcp-no-delay="true"
receive-buffer-size="65536" send-buffer-size="65536"/>
<cql:keyspace name="${cassandra.keyspace}" action="CREATE_DROP"
durable-writes="true"/>
</cass:cluster>
<cql:cluster contact-points="…"
port="${build.cassandra.native_transport_port}" compression="SNAPPY" netty-options-ref="nettyOptions">
<cql:local-pooling-options
min-simultaneous-requests="26" max-simultaneous-requests="101"
core-connections="3" max-connections="9"/>
<cql:remote-pooling-options
min-simultaneous-requests="25" max-simultaneous-requests="100"
core-connections="1" max-connections="2"/>
<cql:socket-options connect-timeout-millis="5000"
keep-alive="true" reuse-address="true" so-linger="60" tcp-no-delay="true"
receive-buffer-size="65536" send-buffer-size="65536"/>
<cql:keyspace name="${cassandra.keyspace}" action="CREATE_DROP"
durable-writes="true"/>
</cass:cluster>
* </pre>
*
* @author Alex Shvid
@@ -132,16 +144,6 @@ public class CassandraClusterFactoryBean
private @Nullable CompressionType compressionType;
private @Nullable Host.StateListener hostStateListener;
private @Nullable LatencyTracker latencyTracker;
private List<CreateKeyspaceSpecification> keyspaceCreations = new ArrayList<>();
private List<AlterKeyspaceSpecification> keyspaceAlterations = new ArrayList<>();
private List<DropKeyspaceSpecification> keyspaceDrops = new ArrayList<>();
private Set<KeyspaceActionSpecification> keyspaceSpecifications = new HashSet<>();
private List<KeyspaceActions> keyspaceActions = new ArrayList<>();
private List<String> startupScripts = new ArrayList<>();
private List<String> shutdownScripts = new ArrayList<>();
private @Nullable LoadBalancingPolicy loadBalancingPolicy;
private NettyOptions nettyOptions = NettyOptions.DEFAULT_INSTANCE;
private @Nullable PoolingOptions poolingOptions;
@@ -154,6 +156,15 @@ public class CassandraClusterFactoryBean
private @Nullable SSLOptions sslOptions;
private @Nullable TimestampGenerator timestampGenerator;
private List<AlterKeyspaceSpecification> keyspaceAlterations = new ArrayList<>();
private List<CreateKeyspaceSpecification> keyspaceCreations = new ArrayList<>();
private List<DropKeyspaceSpecification> keyspaceDrops = new ArrayList<>();
private List<KeyspaceActions> keyspaceActions = new ArrayList<>();
private List<String> startupScripts = new ArrayList<>();
private List<String> shutdownScripts = new ArrayList<>();
private Set<KeyspaceActionSpecification> keyspaceSpecifications = new HashSet<>();
private @Nullable String beanName;
private @Nullable String clusterName;
private String contactPoints = DEFAULT_CONTACT_POINTS;
@@ -166,106 +177,181 @@ public class CassandraClusterFactoryBean
*/
@Override
public void afterPropertiesSet() throws Exception {
this.cluster = initializeCluster(withRegistrations(buildCluster()));
}
Assert.hasText(contactPoints, "At least one server is required");
private @NonNull Cluster buildCluster() {
Cluster.Builder clusterBuilder = newClusterBuilder();
Assert.hasText(this.contactPoints, "At least one server is required");
clusterBuilder.addContactPoints(StringUtils.commaDelimitedListToStringArray(contactPoints)).withPort(port);
clusterBuilder.withMaxSchemaAgreementWaitSeconds(maxSchemaAgreementWaitSeconds);
Builder clusterBuilder = newClusterBuilder()
.addContactPoints(StringUtils.commaDelimitedListToStringArray(this.contactPoints))
.withMaxSchemaAgreementWaitSeconds(this.maxSchemaAgreementWaitSeconds)
.withPort(this.port);
Optional.ofNullable(compressionType).map(CassandraClusterFactoryBean::convertCompressionType)
.ifPresent(clusterBuilder::withCompression);
Optional.ofNullable(this.addressTranslator).ifPresent(clusterBuilder::withAddressTranslator);
Optional.ofNullable(this.loadBalancingPolicy).ifPresent(clusterBuilder::withLoadBalancingPolicy);
Optional.ofNullable(this.nettyOptions).ifPresent(clusterBuilder::withNettyOptions);
Optional.ofNullable(this.poolingOptions).ifPresent(clusterBuilder::withPoolingOptions);
Optional.ofNullable(this.protocolVersion).ifPresent(clusterBuilder::withProtocolVersion);
Optional.ofNullable(this.queryOptions).ifPresent(clusterBuilder::withQueryOptions);
Optional.ofNullable(this.reconnectionPolicy).ifPresent(clusterBuilder::withReconnectionPolicy);
Optional.ofNullable(this.retryPolicy).ifPresent(clusterBuilder::withRetryPolicy);
Optional.ofNullable(this.socketOptions).ifPresent(clusterBuilder::withSocketOptions);
Optional.ofNullable(this.speculativeExecutionPolicy).ifPresent(clusterBuilder::withSpeculativeExecutionPolicy);
Optional.ofNullable(this.timestampGenerator).ifPresent(clusterBuilder::withTimestampGenerator);
Optional.ofNullable(addressTranslator).ifPresent(clusterBuilder::withAddressTranslator);
Optional.ofNullable(loadBalancingPolicy).ifPresent(clusterBuilder::withLoadBalancingPolicy);
clusterBuilder.withNettyOptions(nettyOptions);
Optional.ofNullable(poolingOptions).ifPresent(clusterBuilder::withPoolingOptions);
Optional.ofNullable(protocolVersion).ifPresent(clusterBuilder::withProtocolVersion);
Optional.ofNullable(queryOptions).ifPresent(clusterBuilder::withQueryOptions);
Optional.ofNullable(reconnectionPolicy).ifPresent(clusterBuilder::withReconnectionPolicy);
Optional.ofNullable(retryPolicy).ifPresent(clusterBuilder::withRetryPolicy);
Optional.ofNullable(socketOptions).ifPresent(clusterBuilder::withSocketOptions);
Optional.ofNullable(speculativeExecutionPolicy).ifPresent(clusterBuilder::withSpeculativeExecutionPolicy);
Optional.ofNullable(timestampGenerator).ifPresent(clusterBuilder::withTimestampGenerator);
Optional.ofNullable(this.authProvider)
.map(clusterBuilder::withAuthProvider)
.orElseGet(() -> StringUtils.hasText(this.username)
? clusterBuilder.withCredentials(this.username, this.password)
: clusterBuilder);
if (authProvider != null) {
clusterBuilder.withAuthProvider(authProvider);
} else if (username != null) {
clusterBuilder.withCredentials(username, password);
}
Optional.ofNullable(this.compressionType)
.map(CassandraClusterFactoryBean::convertCompressionType)
.ifPresent(clusterBuilder::withCompression);
if (!jmxReportingEnabled) {
if (!this.jmxReportingEnabled) {
clusterBuilder.withoutJMXReporting();
}
if (!metricsEnabled) {
if (!this.metricsEnabled) {
clusterBuilder.withoutMetrics();
}
if (sslEnabled) {
if (sslOptions != null) {
clusterBuilder.withSSL(sslOptions);
} else {
clusterBuilder.withSSL();
}
if (this.sslEnabled) {
Optional.ofNullable(this.sslOptions)
.map(clusterBuilder::withSSL)
.orElseGet(clusterBuilder::withSSL);
}
Optional.ofNullable(resolveClusterName()).filter(StringUtils::hasText).ifPresent(clusterBuilder::withClusterName);
Optional.ofNullable(resolveClusterName())
.filter(StringUtils::hasText)
.ifPresent(clusterBuilder::withClusterName);
if (clusterBuilderConfigurer != null) {
clusterBuilderConfigurer.configure(clusterBuilder);
if (this.clusterBuilderConfigurer != null) {
this.clusterBuilderConfigurer.configure(clusterBuilder);
}
cluster = clusterBuilder.build();
return clusterBuilder.build();
}
Optional.ofNullable(hostStateListener).ifPresent(cluster::register);
Optional.ofNullable(latencyTracker).ifPresent(cluster::register);
private static Compression convertCompressionType(CompressionType type) {
generateSpecificationsFromFactoryBeans();
switch (type) {
case NONE:
return Compression.NONE;
case SNAPPY:
return Compression.SNAPPY;
case LZ4:
return Compression.LZ4;
}
List<KeyspaceActionSpecification> startup = new ArrayList<>(keyspaceCreations.size() + keyspaceAlterations.size());
startup.addAll(keyspaceCreations);
startup.addAll(keyspaceAlterations);
executeSpecsAndScripts(startup, startupScripts, cluster);
throw new IllegalArgumentException(String.format("Unknown compression type [%s]", type));
}
/*
* (non-Javadoc)
* @see com.datastax.driver.core.Cluster#builder()
*/
Cluster.Builder newClusterBuilder() {
@NonNull Cluster.Builder newClusterBuilder() {
return Cluster.builder();
}
/* (non-Javadoc) */
@Nullable
private String resolveClusterName() {
return StringUtils.hasText(clusterName) ? clusterName : beanName;
private @Nullable String resolveClusterName() {
return StringUtils.hasText(this.clusterName) ? this.clusterName : this.beanName;
}
/*
* (non-Javadoc)
* @see org.springframework.beans.factory.DisposableBean#destroy()
*/
@Override
public void destroy() {
private @NonNull Cluster withRegistrations(@NonNull Cluster cluster) {
if (cluster != null) {
Optional.ofNullable(this.hostStateListener).ifPresent(cluster::register);
Optional.ofNullable(this.latencyTracker).ifPresent(cluster::register);
executeSpecsAndScripts(keyspaceDrops, shutdownScripts, cluster);
cluster.close();
return cluster;
}
private @NonNull Cluster initializeCluster(@NonNull Cluster cluster) {
generateSpecificationsFromFactoryBeans();
List<KeyspaceActionSpecification> startupSpecifications =
new ArrayList<>(this.keyspaceCreations.size() + this.keyspaceAlterations.size());
startupSpecifications.addAll(this.keyspaceCreations);
startupSpecifications.addAll(this.keyspaceAlterations);
executeSpecsAndScripts(startupSpecifications, this.startupScripts, cluster);
return cluster;
}
private void executeSpecsAndScripts(List<? extends KeyspaceActionSpecification> keyspaceActionSpecifications,
List<String> scripts, Cluster cluster) {
if (!CollectionUtils.isEmpty(keyspaceActionSpecifications) || !CollectionUtils.isEmpty(scripts)) {
try (Session session = cluster.connect()) {
CqlTemplate template = new CqlTemplate(session);
keyspaceActionSpecifications
.forEach(keyspaceActionSpecification -> template.execute(toCql(keyspaceActionSpecification)));
scripts.forEach(template::execute);
}
}
}
/**
* Evaluates the contents of all the KeyspaceSpecificationFactoryBeans
* and generates the proper KeyspaceSpecification from them.
*/
private void generateSpecificationsFromFactoryBeans() {
generateSpecifications(this.keyspaceSpecifications);
this.keyspaceActions.forEach(actions -> generateSpecifications(actions.getActions()));
}
private void generateSpecifications(Collection<KeyspaceActionSpecification> specifications) {
specifications.forEach(keyspaceActionSpecification -> {
if (keyspaceActionSpecification instanceof AlterKeyspaceSpecification) {
this.keyspaceAlterations.add((AlterKeyspaceSpecification) keyspaceActionSpecification);
}
else if (keyspaceActionSpecification instanceof CreateKeyspaceSpecification) {
this.keyspaceCreations.add((CreateKeyspaceSpecification) keyspaceActionSpecification);
}
else if (keyspaceActionSpecification instanceof DropKeyspaceSpecification) {
this.keyspaceDrops.add((DropKeyspaceSpecification) keyspaceActionSpecification);
}
});
}
private String toCql(KeyspaceActionSpecification specification) {
if (specification instanceof AlterKeyspaceSpecification) {
return new AlterKeyspaceCqlGenerator((AlterKeyspaceSpecification) specification).toCql();
}
else if (specification instanceof CreateKeyspaceSpecification) {
return new CreateKeyspaceCqlGenerator((CreateKeyspaceSpecification) specification).toCql();
}
else if (specification instanceof DropKeyspaceSpecification) {
return new DropKeyspaceCqlGenerator((DropKeyspaceSpecification) specification).toCql();
}
throw new IllegalArgumentException("Unsupported specification type: "
+ ClassUtils.getQualifiedName(specification.getClass()));
}
/*
* (non-Javadoc)
* @see org.springframework.beans.factory.FactoryBean#getObject()
*/
@Override
public Cluster getObject() {
return cluster;
return this.cluster;
}
/*
@@ -274,7 +360,7 @@ public class CassandraClusterFactoryBean
*/
@Override
public Class<? extends Cluster> getObjectType() {
return (cluster != null ? cluster.getClass() : Cluster.class);
return this.cluster != null ? this.cluster.getClass() : Cluster.class;
}
/*
@@ -286,81 +372,26 @@ public class CassandraClusterFactoryBean
return true;
}
/*
* (non-Javadoc)
* @see org.springframework.beans.factory.DisposableBean#destroy()
*/
@Override
public void destroy() {
if (this.cluster != null) {
executeSpecsAndScripts(this.keyspaceDrops, this.shutdownScripts, this.cluster);
this.cluster.close();
}
}
/*
* (non-Javadoc)
* @see org.springframework.dao.support.PersistenceExceptionTranslator#translateExceptionIfPossible(java.lang.RuntimeException)
*/
@Override
public DataAccessException translateExceptionIfPossible(RuntimeException ex) {
return exceptionTranslator.translateExceptionIfPossible(ex);
}
/**
* Examines the contents of all the KeyspaceSpecificationFactoryBeans and generates the proper KeyspaceSpecification
* from them.
*/
private void generateSpecificationsFromFactoryBeans() {
generateSpecifications(keyspaceSpecifications);
keyspaceActions.forEach(actions -> generateSpecifications(actions.getActions()));
}
private void generateSpecifications(Collection<KeyspaceActionSpecification> specifications) {
specifications.forEach(keyspaceActionSpecification -> {
if (keyspaceActionSpecification instanceof CreateKeyspaceSpecification) {
keyspaceCreations.add((CreateKeyspaceSpecification) keyspaceActionSpecification);
}
if (keyspaceActionSpecification instanceof DropKeyspaceSpecification) {
keyspaceDrops.add((DropKeyspaceSpecification) keyspaceActionSpecification);
}
if (keyspaceActionSpecification instanceof AlterKeyspaceSpecification) {
keyspaceAlterations.add((AlterKeyspaceSpecification) keyspaceActionSpecification);
}
});
}
private void executeSpecsAndScripts(List<? extends KeyspaceActionSpecification> keyspaceActionSpecifications,
List<String> scripts, Cluster cluster) {
if (!CollectionUtils.isEmpty(keyspaceActionSpecifications) || !CollectionUtils.isEmpty(scripts)) {
Session session = cluster.connect();
try {
CqlTemplate template = new CqlTemplate(session);
keyspaceActionSpecifications
.forEach(keyspaceActionSpecification -> template.execute(toCql(keyspaceActionSpecification)));
scripts.forEach(template::execute);
} finally {
if (session != null) {
session.close();
}
}
}
}
private String toCql(KeyspaceActionSpecification specification) {
if (specification instanceof CreateKeyspaceSpecification) {
return new CreateKeyspaceCqlGenerator((CreateKeyspaceSpecification) specification).toCql();
}
if (specification instanceof DropKeyspaceSpecification) {
return new DropKeyspaceCqlGenerator((DropKeyspaceSpecification) specification).toCql();
}
if (specification instanceof AlterKeyspaceSpecification) {
return new AlterKeyspaceCqlGenerator((AlterKeyspaceSpecification) specification).toCql();
}
throw new IllegalArgumentException(
"Unsupported specification type: " + ClassUtils.getQualifiedName(specification.getClass()));
public DataAccessException translateExceptionIfPossible(RuntimeException cause) {
return exceptionTranslator.translateExceptionIfPossible(cause);
}
/*
@@ -369,7 +400,7 @@ public class CassandraClusterFactoryBean
* @since 1.5
*/
@Override
public void setBeanName(String beanName) {
public void setBeanName(@Nullable String beanName) {
this.beanName = beanName;
}
@@ -495,7 +526,7 @@ public class CassandraClusterFactoryBean
* @return the {@link List} of {@link KeyspaceActions}.
*/
public List<KeyspaceActions> getKeyspaceActions() {
return Collections.unmodifiableList(keyspaceActions);
return Collections.unmodifiableList(this.keyspaceActions);
}
/**
@@ -560,7 +591,7 @@ public class CassandraClusterFactoryBean
* @return the startup scripts
*/
public List<String> getStartupScripts() {
return Collections.unmodifiableList(startupScripts);
return Collections.unmodifiableList(this.startupScripts);
}
/**
@@ -578,7 +609,7 @@ public class CassandraClusterFactoryBean
* @return the shutdown scripts
*/
public List<String> getShutdownScripts() {
return Collections.unmodifiableList(shutdownScripts);
return Collections.unmodifiableList(this.shutdownScripts);
}
/**
@@ -592,7 +623,7 @@ public class CassandraClusterFactoryBean
* @return the {@link KeyspaceActionSpecification} associated with this factory.
*/
public Set<KeyspaceActionSpecification> getKeyspaceSpecifications() {
return Collections.unmodifiableSet(keyspaceSpecifications);
return Collections.unmodifiableSet(this.keyspaceSpecifications);
}
/**
@@ -646,7 +677,7 @@ public class CassandraClusterFactoryBean
}
/**
* @param latencyTracker The latencyTracker to set.
* @param latencyTracker {@link LatencyTracker} to set.
*/
public void setLatencyTracker(LatencyTracker latencyTracker) {
this.latencyTracker = latencyTracker;
@@ -667,13 +698,14 @@ public class CassandraClusterFactoryBean
/**
* Sets the {@link ClusterBuilderConfigurer} used to apply additional configuration logic to the
* {@link com.datastax.driver.core.Cluster.Builder}. {@link ClusterBuilderConfigurer} is invoked after all provided
* options are configured. The factory will {@link Builder#build()} the {@link Cluster} after applying
* {@link ClusterBuilderConfigurer}.
* {@link com.datastax.driver.core.Cluster.Builder} object.
*
* {@link ClusterBuilderConfigurer} is invoked after all provided options are configured. The factory will
* {@link Builder#build()} the {@link Cluster} after applying {@link ClusterBuilderConfigurer}.
*
* @param clusterBuilderConfigurer {@link ClusterBuilderConfigurer} used to configure the
* {@link com.datastax.driver.core.Cluster.Builder}.
* @see org.springframework.data.cql.config.ClusterBuilderConfigurer
* {@link com.datastax.driver.core.Cluster.Builder}.
* @see org.springframework.data.cassandra.config.ClusterBuilderConfigurer
*/
public void setClusterBuilderConfigurer(@Nullable ClusterBuilderConfigurer clusterBuilderConfigurer) {
this.clusterBuilderConfigurer = clusterBuilderConfigurer;
@@ -725,18 +757,4 @@ public class CassandraClusterFactoryBean
public void setTimestampGenerator(@Nullable TimestampGenerator timestampGenerator) {
this.timestampGenerator = timestampGenerator;
}
private static Compression convertCompressionType(CompressionType type) {
switch (type) {
case NONE:
return Compression.NONE;
case SNAPPY:
return Compression.SNAPPY;
case LZ4:
return Compression.LZ4;
}
throw new IllegalArgumentException(String.format("Unknown compression type [%s]", type));
}
}

10
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CassandraCqlSessionFactoryBean.java
View File

@@ -86,7 +86,10 @@ public class CassandraCqlSessionFactoryBean
/* (non-Javadoc) */
Session connect(@Nullable String keyspaceName) {
return (StringUtils.hasText(keyspaceName) ? getCluster().connect(keyspaceName) : getCluster().connect());
return StringUtils.hasText(keyspaceName)
? getCluster().connect(keyspaceName)
: getCluster().connect();
}
/*
@@ -104,7 +107,7 @@ public class CassandraCqlSessionFactoryBean
*/
@Override
public Class<? extends Session> getObjectType() {
return (this.session != null ? this.session.getClass() : Session.class);
return this.session != null ? this.session.getClass() : Session.class;
}
/*
@@ -121,6 +124,7 @@ public class CassandraCqlSessionFactoryBean
*/
@Override
public void destroy() throws Exception {
executeScripts(getShutdownScripts());
getSession().close();
}
@@ -253,7 +257,7 @@ public class CassandraCqlSessionFactoryBean
* Sets CQL scripts to be executed immediately before the session is shutdown.
*/
public void setShutdownScripts(@Nullable List<String> scripts) {
this.shutdownScripts = (scripts != null ? new ArrayList<>(scripts) : Collections.emptyList());
this.shutdownScripts = scripts != null ? new ArrayList<>(scripts) : Collections.emptyList();
}
/**

9
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CassandraSessionFactoryBean.java
View File

@@ -133,6 +133,7 @@ public class CassandraSessionFactoryBean extends CassandraCqlSessionFactoryBean
public void setSchemaAction(SchemaAction schemaAction) {
Assert.notNull(schemaAction, "SchemaAction must not be null");
this.schemaAction = schemaAction;
}
@@ -158,13 +159,13 @@ public class CassandraSessionFactoryBean extends CassandraCqlSessionFactoryBean
private void performSchemaActions(boolean drop, boolean dropUnused, boolean ifNotExists) {
CassandraPersistentEntitySchemaCreator schemaCreator = new CassandraPersistentEntitySchemaCreator(
getMappingContext(), getCassandraAdminOperations());
CassandraPersistentEntitySchemaCreator schemaCreator =
new CassandraPersistentEntitySchemaCreator(getMappingContext(), getCassandraAdminOperations());
if (drop) {
CassandraPersistentEntitySchemaDropper schemaDropper = new CassandraPersistentEntitySchemaDropper(
getMappingContext(), getCassandraAdminOperations());
CassandraPersistentEntitySchemaDropper schemaDropper =
new CassandraPersistentEntitySchemaDropper(getMappingContext(), getCassandraAdminOperations());
schemaDropper.dropTables(dropUnused);
schemaDropper.dropUserTypes(dropUnused);

29
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CassandraClusterFactoryBeanUnitTests.java
View File

@@ -15,18 +15,35 @@
*/
package org.springframework.data.cassandra.config;
import static org.assertj.core.api.Assertions.*;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isA;
import static org.mockito.Mockito.*;
import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import org.junit.Test;
import org.springframework.data.cassandra.support.IntegrationTestNettyOptions;
import org.springframework.test.util.ReflectionTestUtils;
import com.datastax.driver.core.*;
import com.datastax.driver.core.AuthProvider;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Configuration;
import com.datastax.driver.core.PlainTextAuthProvider;
import com.datastax.driver.core.PoolingOptions;
import com.datastax.driver.core.ProtocolOptions;
import com.datastax.driver.core.ProtocolOptions.Compression;
import com.datastax.driver.core.ProtocolVersion;
import com.datastax.driver.core.QueryOptions;
import com.datastax.driver.core.RemoteEndpointAwareJdkSSLOptions;
import com.datastax.driver.core.SSLOptions;
import com.datastax.driver.core.SocketOptions;
import com.datastax.driver.core.TimestampGenerator;
import com.datastax.driver.core.policies.AddressTranslator;
import com.datastax.driver.core.policies.ExponentialReconnectionPolicy;
import com.datastax.driver.core.policies.LoadBalancingPolicy;
@@ -47,6 +64,7 @@ public class CassandraClusterFactoryBeanUnitTests {
public void shouldInitializeWithoutAnyOptions() throws Exception {
CassandraClusterFactoryBean bean = new CassandraClusterFactoryBean();
bean.afterPropertiesSet();
assertThat(bean.getObject()).isNotNull();
@@ -59,6 +77,7 @@ public class CassandraClusterFactoryBeanUnitTests {
public void shouldShutdownClusterInstance() throws Exception {
CassandraClusterFactoryBean bean = new CassandraClusterFactoryBean();
bean.afterPropertiesSet();
bean.destroy();
@@ -273,6 +292,8 @@ public class CassandraClusterFactoryBeanUnitTests {
Cluster mockCluster = mock(Cluster.class);
when(mockClusterBuilder.addContactPoints(anyString())).thenReturn(mockClusterBuilder);
when(mockClusterBuilder.withMaxSchemaAgreementWaitSeconds(anyInt())).thenReturn(mockClusterBuilder);
when(mockClusterBuilder.withPort(anyInt())).thenReturn(mockClusterBuilder);
when(mockClusterBuilder.build()).thenReturn(mockCluster);
CassandraClusterFactoryBean bean = spy(new CassandraClusterFactoryBean());
@@ -294,6 +315,8 @@ public class CassandraClusterFactoryBeanUnitTests {
Cluster mockCluster = mock(Cluster.class);
when(mockClusterBuilder.addContactPoints(anyString())).thenReturn(mockClusterBuilder);
when(mockClusterBuilder.withMaxSchemaAgreementWaitSeconds(anyInt())).thenReturn(mockClusterBuilder);
when(mockClusterBuilder.withPort(anyInt())).thenReturn(mockClusterBuilder);
when(mockClusterBuilder.build()).thenReturn(mockCluster);
CassandraClusterFactoryBean bean = spy(new CassandraClusterFactoryBean());

32
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CassandraNamespaceIntegrationTests.java
View File

@@ -15,10 +15,11 @@
*/
package org.springframework.data.cassandra.config;
import static org.assertj.core.api.Assertions.*;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.data.cassandra.core.CassandraTemplate;
@@ -27,7 +28,7 @@ import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
import org.springframework.data.cassandra.core.mapping.SimpleUserTypeResolver;
import org.springframework.data.cassandra.repository.support.AbstractSpringDataEmbeddedCassandraIntegrationTest;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.util.ReflectionTestUtils;
import com.datastax.driver.core.Cluster;
@@ -43,24 +44,29 @@ import com.datastax.driver.core.SocketOptions;
*
* @author Mark Paluch
*/
@RunWith(SpringJUnit4ClassRunner.class)
@RunWith(SpringRunner.class)
@ContextConfiguration
@SuppressWarnings("unused")
public class CassandraNamespaceIntegrationTests extends AbstractSpringDataEmbeddedCassandraIntegrationTest {
@Autowired ApplicationContext applicationContext;
@Autowired
private ApplicationContext applicationContext;
@Test // DATACASS-271
public void clusterShouldHaveCompressionSet() {
Cluster cluster = applicationContext.getBean(Cluster.class);
Cluster cluster = this.applicationContext.getBean(Cluster.class);
Configuration configuration = cluster.getConfiguration();
assertThat(configuration.getProtocolOptions().getCompression()).isEqualTo(Compression.SNAPPY);
}
@Test // DATACASS-271
public void clusterShouldHavePoolingOptionsConfigured() {
Cluster cluster = applicationContext.getBean(Cluster.class);
Cluster cluster = this.applicationContext.getBean(Cluster.class);
PoolingOptions poolingOptions = cluster.getConfiguration().getPoolingOptions();
assertThat(poolingOptions.getMaxRequestsPerConnection(HostDistance.LOCAL)).isEqualTo(101);
@@ -76,7 +82,8 @@ public class CassandraNamespaceIntegrationTests extends AbstractSpringDataEmbedd
@Test // DATACASS-271
public void clusterShouldHaveSocketOptionsConfigured() {
Cluster cluster = applicationContext.getBean(Cluster.class);
Cluster cluster = this.applicationContext.getBean(Cluster.class);
SocketOptions socketOptions = cluster.getConfiguration().getSocketOptions();
assertThat(socketOptions.getConnectTimeoutMillis()).isEqualTo(5000);
@@ -91,10 +98,10 @@ public class CassandraNamespaceIntegrationTests extends AbstractSpringDataEmbedd
@Test // DATACASS-172
public void mappingContextShouldHaveUserTypeResolverConfigured() {
CassandraMappingContext mappingContext = applicationContext.getBean(CassandraMappingContext.class);
CassandraMappingContext mappingContext = this.applicationContext.getBean(CassandraMappingContext.class);
SimpleUserTypeResolver userTypeResolver = (SimpleUserTypeResolver) ReflectionTestUtils.getField(mappingContext,
"userTypeResolver");
SimpleUserTypeResolver userTypeResolver =
(SimpleUserTypeResolver) ReflectionTestUtils.getField(mappingContext, "userTypeResolver");
assertThat(userTypeResolver).isNotNull();
}
@@ -102,8 +109,9 @@ public class CassandraNamespaceIntegrationTests extends AbstractSpringDataEmbedd
@Test // DATACASS-417
public void mappingContextShouldCassandraTemplateConfigured() {
CassandraTemplate cassandraTemplate = applicationContext.getBean(CassandraTemplate.class);
CqlTemplate cqlTemplate = applicationContext.getBean(CqlTemplate.class);
CassandraTemplate cassandraTemplate = this.applicationContext.getBean(CassandraTemplate.class);
CqlTemplate cqlTemplate = this.applicationContext.getBean(CqlTemplate.class);
assertThat(cassandraTemplate.getCqlOperations()).isSameAs(cqlTemplate);
}

132
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/SchemaActionIntegrationTests.java
View File

@@ -16,15 +16,16 @@
package org.springframework.data.cassandra.config;
import static org.assertj.core.api.Assertions.*;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import org.junit.Rule;
import org.junit.Before;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
@@ -32,8 +33,7 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.cassandra.core.cql.SessionCallback;
import org.springframework.data.cassandra.domain.Person;
import org.springframework.data.cassandra.test.util.AbstractEmbeddedCassandraIntegrationTest;
import org.springframework.data.cassandra.test.util.KeyspaceRule;
import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.KeyspaceMetadata;
@@ -42,48 +42,44 @@ import com.datastax.driver.core.Session;
import com.datastax.driver.core.TableMetadata;
/**
* The SchemaActionIntegrationTests class is a test suite of test cases testing the contract and behavior of various
* {@link SchemaAction}s on startup of a Spring configured, Cassandra application client.
* Integration test testing various {@link SchemaAction SchemaActions} on startup of a Spring configured,
* Apache Cassandra application client.
*
* @author John Blum
* @author Mark Paluch
* @see org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest
*/
public class SchemaActionIntegrationTests extends AbstractEmbeddedCassandraIntegrationTest {
public class SchemaActionIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest {
protected static final String KEYSPACE_NAME = SchemaActionIntegrationTests.class.getSimpleName().toLowerCase();
protected static final String CREATE_PERSON_TABLE_CQL =
"CREATE TABLE IF NOT EXISTS person (id int, firstName text, lastName text, PRIMARY KEY(id));";
protected static final String PERSON_TABLE_DEFINITION_CQL = String
.format("CREATE TABLE %s.person (id int, firstName text, lastName text, PRIMARY KEY(id));", KEYSPACE_NAME);
@Rule public ExpectedException exception = ExpectedException.none();
@Rule public KeyspaceRule KEYSPACE_RULE = new KeyspaceRule(cassandraEnvironment, KEYSPACE_NAME);
protected static final String DROP_ADDRESS_TYPE_CQL = "DROP TYPE IF EXISTS address";
protected static final String DROP_PERSON_TABLE_CQL = "DROP TABLE IF EXISTS person";
protected ConfigurableApplicationContext newApplicationContext(Class<?>... annotatedClasses) {
AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext(annotatedClasses);
AnnotationConfigApplicationContext applicationContext =
new AnnotationConfigApplicationContext(annotatedClasses);
applicationContext.registerShutdownHook();
return applicationContext;
}
@SuppressWarnings("all")
protected <T> T doInSessionWithConfiguration(Class<?> annotatedClass, SessionCallback<T> sessionCallback) {
ConfigurableApplicationContext applicationContext = null;
try {
applicationContext = newApplicationContext(annotatedClass);
try (ConfigurableApplicationContext applicationContext = newApplicationContext(annotatedClass)) {
return sessionCallback.doInSession(applicationContext.getBean(Session.class));
} finally {
if (applicationContext != null) {
applicationContext.close();
}
}
}
protected void assertHasTableWithColumns(Session session, String tableName, String... columns) {
@SuppressWarnings("all")
protected void assertTableWithColumnsExists(Session session, String tableName, String... columns) {
Metadata clusterMetadata = session.getCluster().getMetadata();
KeyspaceMetadata keyspaceMetadata = clusterMetadata.getKeyspace(KEYSPACE_NAME);
KeyspaceMetadata keyspaceMetadata = clusterMetadata.getKeyspace(getKeyspace());
assertThat(keyspaceMetadata).isNotNull();
@@ -99,11 +95,24 @@ public class SchemaActionIntegrationTests extends AbstractEmbeddedCassandraInteg
assertThat(tableMetadata.getColumns()).hasSize(columns.length);
}
@Before
public void setup() {
Session session = getSession();
session.execute(DROP_PERSON_TABLE_CQL);
session.execute(DROP_ADDRESS_TYPE_CQL);
}
@Test
public void createWithNoExistingTableCreatesTableFromEntity() {
doInSessionWithConfiguration(CreateWithNoExistingTableConfiguration.class, (SessionCallback<Void>) session -> {
assertHasTableWithColumns(session, "person", "firstName", "lastName", "nickname", "birthDate", "numberOfChildren",
"cool", "createdDate", "zoneId", "mainAddress", "alternativeAddresses");
doInSessionWithConfiguration(CreateWithNoExistingTableConfiguration.class, session -> {
assertTableWithColumnsExists(session, "person", "firstName", "lastName", "nickname",
"birthDate", "numberOfChildren", "cool", "createdDate", "zoneId", "mainAddress",
"alternativeAddresses");
return null;
});
}
@@ -112,43 +121,54 @@ public class SchemaActionIntegrationTests extends AbstractEmbeddedCassandraInteg
public void createWithExistingTableThrowsErrorWhenCreatingTableFromEntity() {
try {
doInSessionWithConfiguration(CreateWithExistingTableConfiguration.class, s -> {
fail(String.format("%s should have failed!", CreateWithExistingTableConfiguration.class.getSimpleName()));
doInSessionWithConfiguration(CreateWithExistingTableConfiguration.class, session -> {
fail(String.format("%s should have failed", CreateWithExistingTableConfiguration.class.getSimpleName()));
return null;
});
fail("Missing BeanCreationException");
} catch (BeanCreationException e) {
assertThat(e).hasMessageContaining(String.format("Table %s.person already exists", KEYSPACE_NAME));
fail("Expected BeanCreationException");
} catch (BeanCreationException cause) {
assertThat(cause).hasMessageContaining(String.format("Table %s.person already exists", getKeyspace()));
}
}
@Test
public void createIfNotExistsWithNoExistingTableCreatesTableFromEntity() {
doInSessionWithConfiguration(CreateIfNotExistsWithNoExistingTableConfiguration.class,
(SessionCallback<Void>) session -> {
assertHasTableWithColumns(session, "person", "firstName", "lastName", "nickname", "birthDate",
"numberOfChildren", "cool", "createdDate", "zoneId", "mainAddress", "alternativeAddresses");
return null;
});
doInSessionWithConfiguration(CreateIfNotExistsWithNoExistingTableConfiguration.class, session -> {
assertTableWithColumnsExists(session, "person", "firstName", "lastName", "nickname",
"birthDate", "numberOfChildren", "cool", "createdDate", "zoneId", "mainAddress",
"alternativeAddresses");
return null;
});
}
@Test
public void createIfNotExistsWithExistingTableUsesExistingTable() {
doInSessionWithConfiguration(CreateIfNotExistsWithExistingTableConfiguration.class,
(SessionCallback<Void>) session -> {
assertHasTableWithColumns(session, "person", "id", "firstName", "lastName");
return null;
});
doInSessionWithConfiguration(CreateIfNotExistsWithExistingTableConfiguration.class, session -> {
assertTableWithColumnsExists(session, "person", "id", "firstName", "lastName");
return null;
});
}
@Test
public void recreateTableFromEntityDropsExistingTable() {
doInSessionWithConfiguration(RecreateSchemaActionWithExistingTableConfiguration.class,
(SessionCallback<Void>) session -> {
assertHasTableWithColumns(session, "person", "firstName", "lastName", "nickname", "birthDate",
"numberOfChildren", "cool", "createdDate", "zoneId", "mainAddress", "alternativeAddresses");
return null;
});
doInSessionWithConfiguration(RecreateSchemaActionWithExistingTableConfiguration.class, session -> {
assertTableWithColumnsExists(session, "person", "firstName", "lastName", "nickname",
"birthDate", "numberOfChildren", "cool", "createdDate", "zoneId", "mainAddress",
"alternativeAddresses");
return null;
});
}
@Configuration
@@ -170,7 +190,7 @@ public class SchemaActionIntegrationTests extends AbstractEmbeddedCassandraInteg
@Override
protected List<String> getStartupScripts() {
return Collections.singletonList(PERSON_TABLE_DEFINITION_CQL);
return Collections.singletonList(CREATE_PERSON_TABLE_CQL);
}
}
@@ -193,7 +213,7 @@ public class SchemaActionIntegrationTests extends AbstractEmbeddedCassandraInteg
@Override
protected List<String> getStartupScripts() {
return Collections.singletonList(PERSON_TABLE_DEFINITION_CQL);
return Collections.singletonList(CREATE_PERSON_TABLE_CQL);
}
}
@@ -207,7 +227,7 @@ public class SchemaActionIntegrationTests extends AbstractEmbeddedCassandraInteg
@Override
protected List<String> getStartupScripts() {
return Collections.singletonList(PERSON_TABLE_DEFINITION_CQL);
return Collections.singletonList(CREATE_PERSON_TABLE_CQL);
}
}
@@ -217,7 +237,9 @@ public class SchemaActionIntegrationTests extends AbstractEmbeddedCassandraInteg
@Bean
@Override
public CassandraClusterFactoryBean cluster() {
return new CassandraClusterFactoryBean() {
@Override
public void afterPropertiesSet() throws Exception {
// avoid Cassandra Cluster creation; use embedded
@@ -231,13 +253,13 @@ public class SchemaActionIntegrationTests extends AbstractEmbeddedCassandraInteg
}
@Override
protected Set<Class<?>> getInitialEntitySet() throws ClassNotFoundException {
protected Set<Class<?>> getInitialEntitySet() {
return Collections.singleton(Person.class);
}
@Override
protected String getKeyspaceName() {
return KEYSPACE_NAME;
return keyspaceRule.getKeyspaceName();
}
}
}

4
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/AlterKeyspaceCqlGeneratorUnitTests.java
View File

@@ -25,7 +25,7 @@ import org.springframework.data.cassandra.core.cql.keyspace.AlterKeyspaceSpecifi
import org.springframework.data.cassandra.core.cql.keyspace.DefaultOption;
import org.springframework.data.cassandra.core.cql.keyspace.KeyspaceOption;
import org.springframework.data.cassandra.core.cql.keyspace.Option;
import org.springframework.data.cassandra.support.RandomKeySpaceName;
import org.springframework.data.cassandra.support.RandomKeyspaceName;
/**
* Unit tests for {@link AlterKeyspaceCqlGenerator}.
@@ -62,7 +62,7 @@ public class AlterKeyspaceCqlGeneratorUnitTests {
public static class CompleteTest extends AlterKeyspaceTest {
public String name = RandomKeySpaceName.create();
public String name = RandomKeyspaceName.create();
public Boolean durableWrites = true;
public Map<Option, Object> replicationMap = new HashMap<>();

8
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/CreateKeyspaceCqlGeneratorUnitTests.java
View File

@@ -26,7 +26,7 @@ import org.springframework.data.cassandra.core.cql.keyspace.DefaultOption;
import org.springframework.data.cassandra.core.cql.keyspace.KeyspaceAttributes;
import org.springframework.data.cassandra.core.cql.keyspace.KeyspaceOption;
import org.springframework.data.cassandra.core.cql.keyspace.Option;
import org.springframework.data.cassandra.support.RandomKeySpaceName;
import org.springframework.data.cassandra.support.RandomKeyspaceName;
/**
* Unit tests for {@link CreateKeyspaceCqlGenerator}.
@@ -72,7 +72,7 @@ public class CreateKeyspaceCqlGeneratorUnitTests {
public static class BasicTest extends CreateKeyspaceTest {
public String name = RandomKeySpaceName.create();
public String name = RandomKeyspaceName.create();
public Boolean durableWrites = true;
public Map<Option, Object> replicationMap = KeyspaceAttributes.newSimpleReplication();
@@ -97,7 +97,7 @@ public class CreateKeyspaceCqlGeneratorUnitTests {
public static class NoOptionsBasicTest extends CreateKeyspaceTest {
public String name = RandomKeySpaceName.create();
public String name = RandomKeyspaceName.create();
public Boolean durableWrites = true;
public Map<Option, Object> replicationMap = KeyspaceAttributes.newSimpleReplication();
@@ -121,7 +121,7 @@ public class CreateKeyspaceCqlGeneratorUnitTests {
public static class NetworkTopologyTest extends CreateKeyspaceTest {
public String name = RandomKeySpaceName.create();
public String name = RandomKeyspaceName.create();
public Boolean durableWrites = false;
public Map<Option, Object> replicationMap = new HashMap<>();

4
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/DropKeyspaceCqlGeneratorUnitTests.java
View File

@@ -19,7 +19,7 @@ import static org.assertj.core.api.Assertions.*;
import org.junit.Test;
import org.springframework.data.cassandra.core.cql.keyspace.DropKeyspaceSpecification;
import org.springframework.data.cassandra.support.RandomKeySpaceName;
import org.springframework.data.cassandra.support.RandomKeyspaceName;
/**
* Unit tests for {@link DropKeyspaceCqlGenerator}.
@@ -45,7 +45,7 @@ public class DropKeyspaceCqlGeneratorUnitTests {
public static class BasicTest extends DropTableTest {
public String name = RandomKeySpaceName.create();
public String name = RandomKeyspaceName.create();
@Override
public DropKeyspaceSpecification specification() {

4
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/cdi/CassandraOperationsProducer.java
View File

@@ -37,7 +37,7 @@ import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity
import org.springframework.data.cassandra.core.mapping.SimpleUserTypeResolver;
import org.springframework.data.cassandra.domain.User;
import org.springframework.data.cassandra.support.CassandraConnectionProperties;
import org.springframework.data.cassandra.support.RandomKeySpaceName;
import org.springframework.data.cassandra.support.RandomKeyspaceName;
import com.datastax.driver.core.Cluster;
import com.google.common.collect.Sets;
@@ -48,7 +48,7 @@ import com.google.common.util.concurrent.Service;
*/
class CassandraOperationsProducer {
public static final String KEYSPACE_NAME = RandomKeySpaceName.create();
public static final String KEYSPACE_NAME = RandomKeyspaceName.create();
@Produces
@Singleton

14
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/AbstractSpringDataEmbeddedCassandraIntegrationTest.java
View File

@@ -30,17 +30,19 @@ import org.springframework.data.cassandra.test.util.AbstractEmbeddedCassandraInt
public abstract class AbstractSpringDataEmbeddedCassandraIntegrationTest
extends AbstractEmbeddedCassandraIntegrationTest {
@Autowired private CassandraOperations template;
@Autowired @SuppressWarnings("unused")
private CassandraOperations template;
/**
* Truncate table for all known {@link org.springframework.data.mapping.PersistentEntity entities}.
* Truncate tables for all known {@link org.springframework.data.mapping.PersistentEntity entities}.
*/
public void deleteAllEntities() {
Stream<? extends CassandraPersistentEntity<?>> stream = template.getConverter().getMappingContext()
.getTableEntities()
.stream();
Stream<? extends CassandraPersistentEntity<?>> stream =
this.template.getConverter().getMappingContext().getTableEntities().stream();
stream.map(CassandraPersistentEntity::getType).filter(type -> !type.isInterface()).forEach(template::truncate);
stream.map(CassandraPersistentEntity::getType)
.filter(type -> !type.isInterface())
.forEach(this.template::truncate);
}
}

4
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/IntegrationTestConfig.java
View File

@@ -27,7 +27,7 @@ import org.springframework.data.cassandra.core.cql.keyspace.CreateKeyspaceSpecif
import org.springframework.data.cassandra.core.cql.keyspace.DropKeyspaceSpecification;
import org.springframework.data.cassandra.support.CassandraConnectionProperties;
import org.springframework.data.cassandra.support.IntegrationTestNettyOptions;
import org.springframework.data.cassandra.support.RandomKeySpaceName;
import org.springframework.data.cassandra.support.RandomKeyspaceName;
import com.datastax.driver.core.NettyOptions;
import com.datastax.driver.core.QueryOptions;
@@ -45,7 +45,7 @@ public class IntegrationTestConfig extends AbstractReactiveCassandraConfiguratio
public static final CassandraConnectionProperties PROPS = new CassandraConnectionProperties();
public static final int PORT = PROPS.getCassandraPort();
public String keyspaceName = RandomKeySpaceName.create();
public String keyspaceName = RandomKeyspaceName.create();
@Override
protected int getPort() {

56
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/support/CassandraConnectionProperties.java
View File

@@ -30,7 +30,7 @@ import org.springframework.util.Assert;
@SuppressWarnings("serial")
public class CassandraConnectionProperties extends Properties {
protected String resourceName = null;
protected String resourceName;
/**
* Create a new {@link CassandraConnectionProperties} using properties from
@@ -41,36 +41,45 @@ public class CassandraConnectionProperties extends Properties {
}
protected CassandraConnectionProperties(String resourceName) {
this.resourceName = resourceName;
loadProperties();
}
private void loadProperties() {
loadProperties(resourceName);
loadProperties(this.resourceName);
putAll(System.getProperties());
}
private void loadProperties(String resourceName) {
InputStream in = null;
try {
in = getClass().getResourceAsStream(resourceName);
if (in == null) {
return;
if (in != null) {
load(in);
}
load(in);
} catch (Exception x) {
throw new RuntimeException(x);
} catch (Exception cause) {
throw new RuntimeException(cause);
} finally {
if (in != null) {
try {
in.close();
} catch (Exception e) {
// gulp
}
} catch (Exception ignore) { }
}
}
}
/**
* @return the Cassandra hostname
*/
public String getCassandraHost() {
return getProperty("build.cassandra.host");
}
/**
* @return the Cassandra port (native).
*/
@@ -99,23 +108,16 @@ public class CassandraConnectionProperties extends Properties {
return getInt("build.cassandra.ssl_storage_port");
}
/**
* @return the Cassandra hostname
*/
public String getCassandraHost() {
return getProperty("build.cassandra.host");
}
/**
* @return the Cassandra type (Embedded or External)
*/
public CassandraType getCassandraType() {
String property = getProperty("build.cassandra.mode");
if (property != null && property.equalsIgnoreCase(CassandraType.EXTERNAL.name())) {
return CassandraType.EXTERNAL;
}
return CassandraType.EMBEDDED;
String cassandraType = getProperty("build.cassandra.mode");
return CassandraType.EXTERNAL.name().equalsIgnoreCase(cassandraType)
? CassandraType.EXTERNAL
: CassandraType.EMBEDDED;
}
/**
@@ -155,14 +157,16 @@ public class CassandraConnectionProperties extends Properties {
String propertyValue = getProperty(propertyName);
try {
return converter.convert(propertyValue);
} catch (Exception e) {
throw new IllegalArgumentException(String.format("%1$s: cannot parse value [%2$s] of property [%3$s] as a [%4$s]",
resourceName, propertyValue, propertyName, type.getSimpleName()), e);
} catch (Exception cause) {
String message = "%1$s: cannot parse value [%2$s] of property [%3$s] as a [%4$s]";
throw new IllegalArgumentException(String.format(message, this.resourceName, propertyValue, propertyName,
type.getSimpleName()), cause);
}
}
public enum CassandraType {
EMBEDDED, EXTERNAL
}
}

12
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/support/RandomKeySpaceName.java → spring-data-cassandra/src/test/java/org/springframework/data/cassandra/support/RandomKeyspaceName.java
View File

@@ -22,16 +22,16 @@ import java.util.UUID;
*
* @author Matthew T. Adams
*/
public class RandomKeySpaceName {
public abstract class RandomKeyspaceName {
private RandomKeySpaceName() {
}
private RandomKeyspaceName() { }
/**
* Creates a random key space name starting with {@code ks} based on a random {@link UUID}.
* Creates a random {@link String keyspace name} starting with {@code ks} based on a random {@link UUID}.
*
* @return
* @return a random {@link String keyspace name}.
* @see java.lang.String
* @see java.util.UUID
*/
public static String create() {
return "ks" + UUID.randomUUID().toString().replace("-", "");

57
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/util/AbstractEmbeddedCassandraIntegrationTest.java
View File

@@ -19,6 +19,7 @@ import java.util.UUID;
import org.junit.ClassRule;
import org.junit.Rule;
import org.springframework.data.cassandra.support.CqlDataSet;
import com.datastax.driver.core.Cluster;
@@ -36,49 +37,55 @@ import com.datastax.driver.core.Cluster;
public abstract class AbstractEmbeddedCassandraIntegrationTest {
/**
* Initiate a Cassandra environment in test class scope.
* Initiate a Cassandra environment in this test scope.
*/
@ClassRule public static final CassandraRule cassandraEnvironment = new CassandraRule("embedded-cassandra.yaml");
@ClassRule
public static final CassandraRule cassandraEnvironment =
new CassandraRule("embedded-cassandra.yaml");
/**
* Initiate a Cassandra environment in test scope.
*/
@Rule public final CassandraRule cassandraRule = cassandraEnvironment.testInstance().before(session -> {
AbstractEmbeddedCassandraIntegrationTest.this.cluster = session.getCluster();
return null;
});
/**
* The {@link Cluster} that's connected to Cassandra.
*/
protected Cluster cluster;
/**
* Creates a random UUID.
* Create and return a random {@link UUID} as a {@link String}.
*
* @return
* @return a random {@link UUID} as a {@link String}.
* @see java.util.UUID#randomUUID()
*/
public static String uuid() {
return UUID.randomUUID().toString();
}
/**
* Returns the {@link Cluster}.
* Initiate a Cassandra environment in test scope.
*/
@Rule
public final CassandraRule cassandraRule = cassandraEnvironment.testInstance().before(session -> {
AbstractEmbeddedCassandraIntegrationTest.this.cluster = session.getCluster();
return null;
});
/**
* The {@link Cluster} connected to Cassandra.
*/
protected Cluster cluster;
/**
* Returns the {@link Cluster} instance.
*
* @return
* @return an instance of {@link Cluster} connected to Cassandra.
*/
public Cluster getCluster() {
return cluster;
return this.cluster;
}
/**
* Executes a CQL script from a classpath resource in given {@code keyspace}.
* Executes a CQL script from a classpath resource in the given {@code keyspace}.
*
* @param cqlResourceName
* @param keyspace
* @param cqlResourceName {@link String resource name} of the CQL script to apply.
* @param keyspace {@link String name} of the Cassandra Keyspace in which to apply the CQL script.
*/
public void execute(String cqlResourceName, String keyspace) {
cassandraRule.execute(CqlDataSet.fromClassPath(cqlResourceName).executeIn(keyspace));
}
CqlDataSet cqlDataSet = CqlDataSet.fromClassPath(cqlResourceName).executeIn(keyspace);
this.cassandraRule.execute(cqlDataSet);
}
}

60
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/util/AbstractKeyspaceCreatingIntegrationTest.java
View File

@@ -16,6 +16,7 @@
package org.springframework.data.cassandra.test.util;
import org.junit.ClassRule;
import org.springframework.util.Assert;
import com.datastax.driver.core.Session;
@@ -33,72 +34,83 @@ import com.datastax.driver.core.Session;
* @author Matthew T. Adams
* @author David Webb
* @author Mark Paluch
* @author John Blum
*/
public abstract class AbstractKeyspaceCreatingIntegrationTest extends AbstractEmbeddedCassandraIntegrationTest {
/**
* Class rule to prepare a keyspace to give tests a keyspace context. The keyspace name is random and changes per
* test.
* Class rule to prepare a Cassandra Keyspace giving tests a Keyspace context.
* The Keyspace name is random and changes per test.
*/
@ClassRule public static final KeyspaceRule keyspaceRule = new KeyspaceRule(cassandraEnvironment);
@ClassRule
public static final KeyspaceRule keyspaceRule = new KeyspaceRule(cassandraEnvironment);
/**
* The session that's connected to the keyspace used in the current instance's test.
* The Session that's connected to the Cassandra Keyspace used in tests.
*/
protected Session session;
/**
* The name of the keyspace to use for this test instance.
* The name of the Cassanda Keyspace to use for this test.
*/
protected final String keyspace;
/**
* Create a new {@link AbstractKeyspaceCreatingIntegrationTest}.
* Constructs a new instance of {@link AbstractKeyspaceCreatingIntegrationTest}.
*/
public AbstractKeyspaceCreatingIntegrationTest() {
this(keyspaceRule.getKeyspaceName());
}
private AbstractKeyspaceCreatingIntegrationTest(final String keyspace) {
private AbstractKeyspaceCreatingIntegrationTest(String keyspace) {
Assert.hasText(keyspace, "Keyspace must not be empty");
this.keyspace = keyspace;
this.session = keyspaceRule.getSession();
cassandraRule.before(session -> {
this.cassandraRule.before(session -> {
if (!keyspace.equals(session.getLoggedKeyspace())) {
session.execute(String.format("USE %s;", keyspace));
session.execute(String.format(KeyspaceRule.USE_KEYSPACE_CQL, keyspace));
}
return null;
});
}
/**
* Returns the {@link Session}. The session is logged into the {@link #getKeyspace()}.
* Returns the configured {@link String name} of the Cassandra Keyspace used for tests.
*
* @return
*/
public Session getSession() {
return session;
}
/**
* Returns the keyspace name.
*
* @return
* @return the confiured {@link String name} of the Cassandra Keyspace used for tests.
*/
public String getKeyspace() {
return keyspace;
return this.keyspace;
}
/**
* Drop a Keyspace if it exists.
* Returns the configured {@link Session}.
*
* @param keyspace
* The {@link Session} is logged into the {@link #getKeyspace()}.
*
* @return the configured {@link Session}.
* @see com.datastax.driver.core.Session
*/
public Session getSession() {
return this.session;
}
@SuppressWarnings("unused")
protected void dropKeyspace() {
dropKeyspace(getKeyspace());
}
/**
* Drops the given Keyspace by {@link String name} if it exists.
*
* @param keyspace {@link String name} of the Keyspace to drop.
*/
public void dropKeyspace(String keyspace) {
session.execute("DROP KEYSPACE IF EXISTS " + keyspace);
this.session.execute(String.format(KeyspaceRule.DROP_KEYSPACE_IF_EXISTS_CQL, keyspace));
}
}

502
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/util/CassandraRule.java
View File

@@ -15,21 +15,22 @@
*/
package org.springframework.data.cassandra.test.util;
import static org.springframework.data.cassandra.test.util.CassandraRule.InvocationMode.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import org.junit.rules.ExternalResource;
import org.springframework.data.cassandra.core.cql.SessionCallback;
import org.springframework.data.cassandra.support.CassandraConnectionProperties;
import org.springframework.data.cassandra.support.CqlDataSet;
import org.springframework.data.cassandra.support.IntegrationTestNettyOptions;
import org.springframework.util.Assert;
import org.springframework.util.SocketUtils;
import org.springframework.util.StringUtils;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.QueryOptions;
@@ -37,45 +38,47 @@ import com.datastax.driver.core.Session;
import com.datastax.driver.core.SocketOptions;
/**
* Rule to provide a Cassandra context for integration tests. This rule can use/spin up either an embedded Cassandra
* instance or use an external instance. Typical usage:
* JUnit Rule used to provide a Cassandra context for integration tests.
*
* This rule can use/spin up either an embedded Cassandra instance or use an external instance.
*
* Typical usage:
*
* <pre>
* {
* public class MyIntegrationTest {
* &#064;Rule public CassandraRule rule = new CassandraRule(CONFIG). //
* before(new ClassPathCQLDataSet("CreateIndexCqlGeneratorIntegrationTests-BasicTest.cql", "keyspace"));
* }
* public class MyIntegrationTest {
* &#064;Rule public CassandraRule rule = new CassandraRule(CONFIG)
* .before(new ClassPathCQLDataSet("CreateIndexCqlGeneratorIntegrationTests-BasicTest.cql", "keyspace"));
* }
* </pre>
*
* @author Mark Paluch
* @author John Blum
* @since 1.5
*/
public class CassandraRule extends ExternalResource {
private static ResourceHolder resourceHolder;
private final long startupTimeout;
@SuppressWarnings("all")
private final CassandraConnectionProperties properties = new CassandraConnectionProperties();
private final String configurationFileName;
private final long startUpTimeout;
private List<SessionCallback<Void>> before = new ArrayList<>();
private Map<SessionCallback<?>, InvocationMode> invocationModeMap = new HashMap<>();
private List<SessionCallback<Void>> after = new ArrayList<>();
private Session session;
private Cluster cluster;
private CassandraRule parent;
private Cluster cluster;
private Integer cassandraPort;
private final List<SessionCallback<Void>> after = new ArrayList<>();
private final List<SessionCallback<Void>> before = new ArrayList<>();
private final Map<SessionCallback<?>, InvocationMode> invocationModeMap = new HashMap<>();
private Session session;
private final String configurationFilename;
/**
* Create a new {@link CassandraRule} and allows the use of a config file.
*
@@ -86,31 +89,78 @@ public class CassandraRule extends ExternalResource {
}
/**
* Create a new {@link CassandraRule}, allows the use of a config file and to provide a startup timeout.
* Constructs a new instance of {@link CassandraRule} initialized with the given YAML configuration resource,
* thereby allowing the use of a configuration file and to provide a startup timeout.
*
* @param yamlConfigurationResource name of the configuration resource, must not be {@literal null} and not empty
* @param startUpTimeout the startup timeout
* @param yamlConfigurationResource {@link String name} of the configuration resource;
* must not be {@literal null} or empty.
* @param startupTimeout long value indicating the startup timeout in milliseconds.
*/
public CassandraRule(String yamlConfigurationResource, long startUpTimeout) {
public CassandraRule(String yamlConfigurationResource, long startupTimeout) {
Assert.hasText(yamlConfigurationResource, "Configuration file name must not be empty!");
Assert.hasText(yamlConfigurationResource, "YAML configuration resource must not be empty");
this.configurationFileName = yamlConfigurationResource;
this.startUpTimeout = startUpTimeout;
this.configurationFilename = yamlConfigurationResource;
this.startupTimeout = startupTimeout;
}
/**
* Create a new {@link CassandraRule} using a parent {@link CassandraRule} to preserve cluster/connection facilities.
* Constructs a new instance of {@link CassandraRule} using the provided (parent) {@link CassandraRule}
* to preserve cluster/connection context.
*
* @param parent the parent instance
* @param parent the {@link CassandraRule parent} instance.
*/
private CassandraRule(CassandraRule parent) {
this.configurationFileName = null;
this.startUpTimeout = -1;
this.configurationFilename = null;
this.startupTimeout = -1;
this.parent = parent;
}
/**
* Creates a {@link CassandraRule} to be used in an "owning" scope.
*
* The derived {@link CassandraRule} shares the connection of {@literal this} instance
* and starts with a fresh before/after configuration.
*
* @return a derived {@link CassandraRule} sharing the connection of {@literal this} instance.
* @see #CassandraRule(CassandraRule)
*/
public CassandraRule testInstance() {
return new CassandraRule(this);
}
/**
* Returns the {@link Cluster}.
*
* @return the Cluster
*/
public Cluster getCluster() {
return this.cluster;
}
/**
* Returns the Cassandra port.
*
* @return the Cassandra port
*/
public int getPort() {
Assert.state(this.cassandraPort != null, "Cassandra port was not initialized");
return this.cassandraPort;
}
/**
* Returns the {@link Session}. The session state can be initialized and pointing to a keyspace other than
* {@code system}.
*
* @return the Session
*/
public Session getSession() {
return this.session;
}
/**
* Add a {@link CqlDataSet} to execute before each test run.
*
@@ -118,7 +168,7 @@ public class CassandraRule extends ExternalResource {
* @return the rule
*/
public CassandraRule before(CqlDataSet cqlDataSet) {
return before(each(), cqlDataSet);
return before(InvocationMode.EACH, cqlDataSet);
}
/**
@@ -128,9 +178,9 @@ public class CassandraRule extends ExternalResource {
* @param cqlDataSet must not be {@literal null}
* @return the rule
*/
public CassandraRule before(InvocationMode invocationMode, final CqlDataSet cqlDataSet) {
public CassandraRule before(InvocationMode invocationMode, CqlDataSet cqlDataSet) {
Assert.notNull(cqlDataSet, "CQLDataSet must not be null");
Assert.notNull(cqlDataSet, "CqlDataSet must not be null");
SessionCallback<Void> sessionCallback = session -> {
load(session, cqlDataSet);
@@ -138,6 +188,7 @@ public class CassandraRule extends ExternalResource {
};
before(invocationMode, sessionCallback);
return this;
}
@@ -147,10 +198,8 @@ public class CassandraRule extends ExternalResource {
* @param sessionCallback must not be {@literal null}
* @return the rule
*/
public CassandraRule before(final SessionCallback<?> sessionCallback) {
Assert.notNull(sessionCallback, "SessionCallback must not be null");
return before(each(), sessionCallback);
public CassandraRule before(SessionCallback<?> sessionCallback) {
return before(InvocationMode.EACH, sessionCallback);
}
/**
@@ -161,12 +210,13 @@ public class CassandraRule extends ExternalResource {
* @return the rule
*/
@SuppressWarnings("unchecked")
public CassandraRule before(InvocationMode invocationMode, final SessionCallback<?> sessionCallback) {
public CassandraRule before(InvocationMode invocationMode, SessionCallback<?> sessionCallback) {
Assert.notNull(sessionCallback, "SessionCallback must not be null");
before.add((SessionCallback<Void>) sessionCallback);
invocationModeMap.put(sessionCallback, invocationMode);
this.before.add((SessionCallback<Void>) sessionCallback);
this.invocationModeMap.put(sessionCallback, invocationMode);
return this;
}
@@ -176,11 +226,12 @@ public class CassandraRule extends ExternalResource {
* @param cqlDataSet must not be {@literal null}
* @return the rule
*/
public CassandraRule after(final CqlDataSet cqlDataSet) {
@SuppressWarnings("unused")
public CassandraRule after(CqlDataSet cqlDataSet) {
Assert.notNull(cqlDataSet, "CQLDataSet must not be null");
Assert.notNull(cqlDataSet, "CqlDataSet must not be null");
after.add(session -> {
this.after.add(session -> {
load(CassandraRule.this.session, cqlDataSet);
return null;
});
@@ -189,29 +240,125 @@ public class CassandraRule extends ExternalResource {
}
/**
* Execute a {@link CqlDataSet}.
*
* @param cqlDataSet the CQL data set, must not be {@literal null}.
*/
public void execute(CqlDataSet cqlDataSet) {
Assert.notNull(cqlDataSet, "CQLDataSet must not be null");
load(session, cqlDataSet);
}
/**
* Execute the {@code before} sequence.
*
* @throws Exception
* Execute {@code before} sequence.
*/
@Override
public void before() throws Exception {
protected void before() throws Exception {
startCassandraIfNeeded();
setupConnection();
initializeConnection();
executeBeforeHooks();
}
private void startCassandraIfNeeded() throws Exception {
if (isStartNeeded()) {
configureRemoteJmxPort();
runEmbeddedCassandra();
}
}
private boolean isStartNeeded() {
return isParent() && isEmbedded();
}
private void configureRemoteJmxPort() {
if (!System.getProperties().containsKey("com.sun.management.jmxremote.port")) {
System.setProperty("com.sun.management.jmxremote.port",
String.valueOf(SocketUtils.findAvailableTcpPort(1024)));
}
}
private void runEmbeddedCassandra() throws Exception {
if (this.configurationFilename != null) {
EmbeddedCassandraServerHelper.startEmbeddedCassandra(this.configurationFilename, this.startupTimeout);
}
}
private synchronized void initializeConnection() {
if (isParent()) {
this.cassandraPort = resolvePort();
if (resourceHolder == null) {
this.cluster = buildCluster(this.cassandraPort);
if (isClusterReuseEnabled()) {
resourceHolder = new ResourceHolder(this.cluster);
}
} else {
this.cluster = resourceHolder.cluster;
}
} else {
this.cassandraPort = this.parent.cassandraPort;
this.cluster = this.parent.cluster;
}
this.session = resolveSession();
}
private Cluster buildCluster(int port) {
QueryOptions queryOptions = new QueryOptions().setRefreshSchemaIntervalMillis(0);
SocketOptions socketOptions = new SocketOptions()
.setConnectTimeoutMillis((int) TimeUnit.SECONDS.toMillis(15))
.setReadTimeoutMillis((int) TimeUnit.SECONDS.toMillis(15));
String host = resolveHost();
return new Cluster.Builder()
.addContactPoints(host)
.withPort(port)
.withMaxSchemaAgreementWaitSeconds(3)
.withNettyOptions(IntegrationTestNettyOptions.INSTANCE)
.withQueryOptions(queryOptions)
.withSocketOptions(socketOptions)
.build();
}
private String resolveHost() {
return isEmbedded()
? EmbeddedCassandraServerHelper.getHost()
: this.properties.getCassandraHost();
}
private int resolvePort() {
return isEmbedded()
? EmbeddedCassandraServerHelper.getNativeTransportPort()
: this.properties.getCassandraPort();
}
private Session resolveSession() {
return isNotParent() ? this.parent.getSession()
: resourceHolder != null ? resourceHolder.session
: this.cluster.connect();
}
private void executeBeforeHooks() {
this.before.forEach(sessionCallback -> {
InvocationMode invocationMode = this.invocationModeMap.get(sessionCallback);
if (!InvocationMode.NEVER.equals(invocationMode)) {
if (InvocationMode.ONCE.equals(invocationMode)) {
this.invocationModeMap.put(sessionCallback, InvocationMode.NEVER);
}
sessionCallback.doInSession(this.session);
}
});
}
/**
* Execute the {@code after} sequence.
*/
@@ -220,161 +367,62 @@ public class CassandraRule extends ExternalResource {
super.after();
executeAfterHooks();
cleanupConnection();
}
/**
* Returns the {@link Cluster}.
*
* @return the Cluster
*/
public Cluster getCluster() {
return cluster;
}
/**
* Returns the {@link Session}. The session state can be initialized and pointing to a keyspace other than
* {@code system}.
*
* @return the Session
*/
public Session getSession() {
return session;
}
/**
* Returns the Cassandra port.
*
* @return the Cassandra port
*/
public int getPort() {
Assert.state(cassandraPort != null, "Cassandra port is not initialized");
return cassandraPort;
}
/**
* Creates a {@link CassandraRule} to be used in a own scope. The derived {@link CassandraRule} shares the connection
* of this instance and starts with a fresh before/after configuration.
*
* @return a derived {@link CassandraRule} sharing the connection of this instance
*/
public CassandraRule testInstance() {
return new CassandraRule(this);
}
private void startCassandraIfNeeded() throws Exception {
if (parent == null && properties.getCassandraType() == CassandraConnectionProperties.CassandraType.EMBEDDED) {
/* start an embedded Cassandra instance*/
if (!System.getProperties().containsKey("com.sun.management.jmxremote.port")) {
System.setProperty("com.sun.management.jmxremote.port", "" + SocketUtils.findAvailableTcpPort(1024));
}
if (configurationFileName != null) {
EmbeddedCassandraServerHelper.startEmbeddedCassandra(configurationFileName, startUpTimeout);
}
}
}
private void executeBeforeHooks() {
for (SessionCallback<Void> sessionCallback : before) {
InvocationMode invocationMode = invocationModeMap.get(sessionCallback);
if (invocationMode == never()) {
continue;
}
if (invocationMode == firstTest()) {
invocationModeMap.put(sessionCallback, never());
}
sessionCallback.doInSession(session);
}
releaseConnection();
}
private void executeAfterHooks() {
for (SessionCallback<Void> sessionCallback : after) {
sessionCallback.doInSession(session);
}
this.after.forEach(sessionCallback -> sessionCallback.doInSession(this.session));
}
private void setupConnection() {
if (parent == null) {
String hostIp;
int port;
if (properties.getCassandraType() == CassandraConnectionProperties.CassandraType.EMBEDDED) {
hostIp = EmbeddedCassandraServerHelper.getHost();
port = EmbeddedCassandraServerHelper.getNativeTransportPort();
} else {
hostIp = properties.getCassandraHost();
port = properties.getCassandraPort();
}
cassandraPort = port;
QueryOptions queryOptions = new QueryOptions();
queryOptions.setRefreshSchemaIntervalMillis(0);
SocketOptions socketOptions = new SocketOptions();
socketOptions.setConnectTimeoutMillis((int) TimeUnit.SECONDS.toMillis(15));
socketOptions.setReadTimeoutMillis((int) TimeUnit.SECONDS.toMillis(15));
if (resourceHolder == null) {
cluster = new Cluster.Builder().addContactPoints(hostIp) //
.withPort(port) //
.withQueryOptions(queryOptions) //
.withMaxSchemaAgreementWaitSeconds(3) //
.withSocketOptions(socketOptions) //
.withNettyOptions(IntegrationTestNettyOptions.INSTANCE) //
.build();
if (properties.getBoolean("build.cassandra.reuse-cluster")) {
resourceHolder = new ResourceHolder(cluster, cluster.connect());
}
} else {
cluster = resourceHolder.cluster;
}
} else {
cluster = parent.cluster;
cassandraPort = parent.cassandraPort;
}
if (parent != null) {
session = parent.getSession();
} else if (resourceHolder == null) {
session = cluster.connect();
} else {
session = resourceHolder.session;
}
}
private void cleanupConnection() {
private synchronized void releaseConnection() {
if (resourceHolder == null) {
if (parent == null) {
session.close();
cluster.closeAsync();
cluster = null;
if (isParent()) {
this.session.close();
this.cluster.closeAsync();
this.cluster = null;
} else {
session.closeAsync();
this.session.closeAsync();
}
}
session = null;
this.session = null;
}
private void load(Session session, final CqlDataSet cqlDataSet) {
private boolean isClusterReuseEnabled() {
return this.properties.getBoolean("build.cassandra.reuse-cluster");
}
if (cqlDataSet.getKeyspaceName() != null && !cqlDataSet.getKeyspaceName().equals(session.getLoggedKeyspace())) {
session.execute(String.format("USE %s;", cqlDataSet.getKeyspaceName()));
}
private boolean isEmbedded() {
return CassandraConnectionProperties.CassandraType.EMBEDDED.equals(this.properties.getCassandraType());
}
private boolean isNotParent() {
return !isParent();
}
private boolean isParent() {
return this.parent == null;
}
/**
* Execute a {@link CqlDataSet}.
*
* @param cqlDataSet the CQL data set, must not be {@literal null}.
*/
public void execute(CqlDataSet cqlDataSet) {
Assert.notNull(cqlDataSet, "CqlDataSet must not be null");
load(this.session, cqlDataSet);
}
private void load(Session session, CqlDataSet cqlDataSet) {
Optional.of(cqlDataSet.getKeyspaceName())
.filter(StringUtils::hasText)
.filter(keyspaceName -> !keyspaceName.equals(session.getLoggedKeyspace()))
.ifPresent(keyspaceName -> session.execute(String.format(KeyspaceRule.USE_KEYSPACE_CQL, keyspaceName)));
cqlDataSet.getCqlStatements().forEach(session::execute);
}
@@ -382,42 +430,17 @@ public class CassandraRule extends ExternalResource {
/**
* Invocation mode for before calls.
*/
public static class InvocationMode {
public enum InvocationMode {
private static final InvocationMode once = new InvocationMode();
private static final InvocationMode each = new InvocationMode();
private static final InvocationMode never = new InvocationMode();
/** {@link InvocationMode} to invoke an action before each test. */
EACH,
/**
* Invocation mode to invoke an action once at before the first test.
*
* @return the {@code on first test} invocation mode
*/
public static InvocationMode firstTest() {
return once;
}
/** {@link InvocationMode} to never invoke an action. */
NEVER,
/**
* Invocation mode to invoke an action on each run.
*
* @return the {@code on each test} invocation mode
*/
public static InvocationMode each() {
return each;
}
/** {@link InvocationMode} to invoke an action once before the first test. */
ONCE
/**
* Invocation mode to never invoke an action.
*
* @return the {@code never} invocation mode
*/
static InvocationMode never() {
return never;
}
private InvocationMode() {
}
}
private static class ResourceHolder {
@@ -425,7 +448,12 @@ public class CassandraRule extends ExternalResource {
private Cluster cluster;
private Session session;
public ResourceHolder(final Cluster cluster, final Session session) {
private ResourceHolder(Cluster cluster) {
this(cluster, cluster.connect());
}
private ResourceHolder(Cluster cluster, Session session) {
this.cluster = cluster;
this.session = session;

48
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/util/EmbeddedCassandraServerHelper.java
View File

@@ -15,7 +15,7 @@
*/
package org.springframework.data.cassandra.test.util;
import static java.util.concurrent.TimeUnit.*;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import java.io.File;
import java.io.FileOutputStream;
@@ -73,15 +73,6 @@ class EmbeddedCassandraServerHelper {
return DatabaseDescriptor.getRpcAddress().getHostName();
}
/**
* Get embedded cassandra RPC port.
*
* @return the cassandra RPC port
*/
public static int getRpcPort() {
return DatabaseDescriptor.getRpcPort();
}
/**
* Get embedded cassandra native transport port.
*
@@ -91,6 +82,15 @@ class EmbeddedCassandraServerHelper {
return DatabaseDescriptor.getNativeTransportPort();
}
/**
* Get embedded cassandra RPC port.
*
* @return the cassandra RPC port
*/
public static int getRpcPort() {
return DatabaseDescriptor.getRpcPort();
}
/**
* Start an embedded Cassandra instance.
*
@@ -113,16 +113,17 @@ class EmbeddedCassandraServerHelper {
public static void startEmbeddedCassandra(String yamlResource, String tmpDir, long timeout) throws Exception {
if (cassandraRef.get() != null) {
/* nothing to do Cassandra is already started */
/* Nothing to do; Cassandra is already started */
return;
}
if (!sync.compareAndSet(null, new Object())) {
/* A different Thread was faster, so nothing to do for us here */
/* A different Thread was faster, so nothing to do this time */
return;
}
File yamlFile = new File(tmpDir, new File(yamlResource).getName());
prepareCassandraDirectory(yamlResource, tmpDir, yamlFile);
startEmbeddedCassandra(yamlFile, timeout);
}
@@ -160,8 +161,10 @@ class EmbeddedCassandraServerHelper {
cleanupAndRecreateDirectories();
final CassandraDaemon cassandraDaemon = new CassandraDaemon();
CassandraDaemon cassandraDaemon = new CassandraDaemon();
ExecutorService executor = Executors.newSingleThreadExecutor();
Future<?> future = executor.submit(() -> {
cassandraDaemon.activate();
cassandraRef.compareAndSet(null, cassandraDaemon);
@@ -169,16 +172,17 @@ class EmbeddedCassandraServerHelper {
try {
future.get(timeout, MILLISECONDS);
} catch (ExecutionException e) {
} catch (ExecutionException cause) {
log.error("Cassandra daemon did not start after " + timeout + " ms. Consider increasing the timeout");
throw new IllegalStateException("Cassandra daemon did not start within timeout", e);
} catch (InterruptedException e) {
log.error("Interrupted waiting for Cassandra daemon to start:", e);
throw new IllegalStateException("Cassandra daemon did not start within timeout", cause);
} catch (InterruptedException cause) {
log.error("Interrupted waiting for Cassandra daemon to start:", cause);
Thread.currentThread().interrupt();
throw new IllegalStateException(e);
throw new IllegalStateException(cause);
} finally {
executor.shutdown();
}
@@ -192,13 +196,14 @@ class EmbeddedCassandraServerHelper {
createCassandraDirectories();
CommitLog commitLog = CommitLog.instance;
commitLog.getCurrentPosition(); // wait for commit log allocator instantiation to avoid hanging on a race condition
commitLog.resetUnsafe(true); // cleanup screws w/ CommitLog, this brings it back to safe state
}
private static void cleanup() throws IOException {
// clean up commitlog and data locations
// clean up commit log and data locations
rmdirs(DatabaseDescriptor.getCommitLogLocation());
rmdirs(DatabaseDescriptor.getAllDataFileLocations());
}
@@ -247,6 +252,9 @@ class EmbeddedCassandraServerHelper {
}
private static void rmdirs(File... fileOrDirectories) throws IOException {
Arrays.stream(fileOrDirectories).filter(File::exists).forEach(FileUtils::deleteRecursive);
Arrays.stream(fileOrDirectories)
.filter(File::exists)
.forEach(FileUtils::deleteRecursive);
}
}

119
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/util/KeyspaceRule.java
View File

@@ -16,7 +16,8 @@
package org.springframework.data.cassandra.test.util;
import org.junit.rules.ExternalResource;
import org.springframework.data.cassandra.support.RandomKeySpaceName;
import org.springframework.data.cassandra.support.RandomKeyspaceName;
import org.springframework.util.Assert;
import com.datastax.driver.core.Cluster;
@@ -38,78 +39,104 @@ import com.datastax.driver.core.Session;
*/
public class KeyspaceRule extends ExternalResource {
static final String CREATE_KEYSPACE_CQL =
"CREATE KEYSPACE %s WITH durable_writes = false AND replication = {'class': 'SimpleStrategy', 'replication_factor' : 1};";
static final String DROP_KEYSPACE_CQL = "DROP KEYSPACE %s;";
static final String DROP_KEYSPACE_IF_EXISTS_CQL = String.format(DROP_KEYSPACE_CQL, "IF EXISTS %s");
static final String USE_KEYSPACE_CQL = "USE %s;";
private final CassandraRule cassandraRule;
private Session session;
private final String keyspaceName;
/**
* Create a {@link KeyspaceRule} initialized with a {@link CassandraRule} for creating a keyspace using a random name.
* Constructs a new instance of {@link KeyspaceRule} initialized with a {@link CassandraRule}
* to create a Cassandra Keyspace using a random name.
*
* @param cassandraRule
* @param cassandraRule {@link CassandraRule} used to setup the Cassandra environment.
* @throws IllegalArgumentException if {@link CassandraRule} is {@literal null}.
* @see org.springframework.data.cassandra.support.RandomKeyspaceName
* @see org.springframework.data.cassandra.test.util.CassandraRule
*/
public KeyspaceRule(CassandraRule cassandraRule) {
this(cassandraRule, RandomKeySpaceName.create());
this(cassandraRule, RandomKeyspaceName.create());
}
/**
* Create a {@link KeyspaceRule} initialized with a {@link CassandraRule} for creating a keyspace using the given
* {@code keyspaceName}.
* Constructs a new instance of {@link KeyspaceRule} initialized with a {@link CassandraRule}
* to create a Cassandra Keyspace with the given {@code keyspaceName}.
*
* @param cassandraRule
* @param keyspaceName
* @param cassandraRule {@link CassandraRule} used to setup the Cassandra environment.
* @param keyspaceName {@link String name} of the Cassandra Keyspace to use in tests.
* @throws IllegalArgumentException if {@link CassandraRule} is {@literal null}
* or the Keyspace name is not specified.
* @see org.springframework.data.cassandra.test.util.CassandraRule
* @see org.springframework.data.cassandra.test.util.CassandraRule
*/
public KeyspaceRule(CassandraRule cassandraRule, String keyspaceName) {
Assert.notNull(cassandraRule, "CassandraRule must not be null!");
Assert.hasText(keyspaceName, "KeyspaceName must not be empty!");
Assert.notNull(cassandraRule, "CassandraRule must not be null");
Assert.hasText(keyspaceName, "KeyspaceName must not be empty");
this.keyspaceName = keyspaceName;
this.cassandraRule = cassandraRule;
this.keyspaceName = keyspaceName;
this.cassandraRule.before(session -> {
KeyspaceRule.this.session = this.cassandraRule.getSession();
return null;
});
}
/**
* Returns the {@link String name} of the Cassandra Keyspace.
*
* @return the {@link String name} of the Cassandra keyspace.
*/
public String getKeyspaceName() {
return this.keyspaceName;
}
/**
* Returns the {@link Session}.
*
* The {@link Session} state can be initialized and pointing to a Keyspace other than {@code system}.
*
* @return the current Cassandr {@link Session}.
* @see com.datastax.driver.core.Session
*/
public Session getSession() {
return this.session;
}
private Session resolveSession() {
Session session = getSession();
this.session = session != null ? session : this.cassandraRule.getSession();
Assert.state(this.session != null, "Session was not initialized");
return this.session;
}
@Override
protected void before() throws Throwable {
protected void before() {
// Support initialized and initializing CassandraRule.
if (cassandraRule.getCluster() != null) {
this.session = cassandraRule.getSession();
} else {
cassandraRule.before(session -> {
KeyspaceRule.this.session = cassandraRule.getSession();
return null;
});
}
Session session = resolveSession();
Assert.state(session != null, "Session was not initialized");
session.execute(String.format("CREATE KEYSPACE %s WITH durable_writes = false AND "
+ "replication = {'class': 'SimpleStrategy', 'replication_factor' : 1};", keyspaceName));
session.execute(String.format("USE %s;", keyspaceName));
session.execute(String.format(CREATE_KEYSPACE_CQL, this.keyspaceName));
session.execute(String.format(USE_KEYSPACE_CQL, this.keyspaceName));
}
@Override
protected void after() {
session.execute("USE system;");
session.execute(String.format("DROP KEYSPACE %s;", keyspaceName));
}
Session session = getSession();
/**
* Returns the {@link Session}. The session state can be initialized and pointing to a keyspace other than
* {@code system}.
*
* @return
*/
public Session getSession() {
return session;
}
/**
* Returns the keyspace name.
*
* @return
*/
public String getKeyspaceName() {
return keyspaceName;
session.execute(String.format(USE_KEYSPACE_CQL, "system"));
session.execute(String.format(DROP_KEYSPACE_CQL, this.keyspaceName));
}
}

2
spring-data-cassandra/src/test/resources/config/spring-data-cassandra-basic.xml
View File

@@ -10,7 +10,7 @@
">
<bean name="randomKeyspaceName" class="java.lang.String">
<constructor-arg value="#{T(org.springframework.data.cassandra.support.RandomKeySpaceName).create()}"/>
<constructor-arg value="#{T(org.springframework.data.cassandra.support.RandomKeyspaceName).create()}"/>
</bean>
<bean id="nettyOptions" class="org.springframework.data.cassandra.support.IntegrationTestNettyOptions"/>

17
spring-data-cassandra/src/test/resources/org/springframework/data/cassandra/repository/forcequote/compositeprimarykey/ForceQuotedCompositePrimaryKeyRepositoryXmlConfigIntegrationTests-context.xml
View File

@@ -1,12 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:cass="http://www.springframework.org/schema/data/cassandra"
xsi:schemaLocation="
http://www.springframework.org/schema/data/cassandra http://www.springframework.org/schema/data/cassandra/spring-cassandra.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
">
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:cass="http://www.springframework.org/schema/data/cassandra"
xsi:schemaLocation="
http://www.springframework.org/schema/data/cassandra http://www.springframework.org/schema/data/cassandra/spring-cassandra.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
">
<import resource="classpath:/config/spring-data-cassandra-basic.xml" />
<import resource="classpath:/config/spring-data-cassandra-basic.xml"/>
<cass:mapping
entity-base-packages="org.springframework.data.cassandra.repository.forcequote.compositeprimarykey">
@@ -36,6 +37,6 @@
schema-action="RECREATE_DROP_UNUSED">
</cass:session>
<cass:repositories
base-package="org.springframework.data.cassandra.repository.forcequote.compositeprimarykey"/>
<cass:repositories base-package="org.springframework.data.cassandra.repository.forcequote.compositeprimarykey"/>
</beans>

15
spring-data-cassandra/src/test/resources/org/springframework/data/cassandra/repository/forcequote/config/ForceQuotedRepositoryXmlConfigIntegrationTests-context.xml
View File

@@ -1,12 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:cass="http://www.springframework.org/schema/data/cassandra"
xsi:schemaLocation="
xmlns:cass="http://www.springframework.org/schema/data/cassandra"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="
http://www.springframework.org/schema/data/cassandra http://www.springframework.org/schema/data/cassandra/spring-cassandra.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
">
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
">
<import resource="classpath:/config/spring-data-cassandra-basic.xml" />
<import resource="classpath:/config/spring-data-cassandra-basic.xml"/>
<cass:mapping
entity-base-packages="org.springframework.data.cassandra.repository.forcequote.config">
@@ -46,6 +47,6 @@
schema-action="RECREATE_DROP_UNUSED">
</cass:session>
<cass:repositories
base-package="org.springframework.data.cassandra.repository.forcequote.config"/>
<cass:repositories base-package="org.springframework.data.cassandra.repository.forcequote.config"/>
</beans>