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

DATACASS-656 - Migrate CqlSessionFactoryBean to CqlSession.

Browse Source 
Original pull request: #167.
This commit is contained in:
Mark Paluch
2019-11-27 14:40:11 +01:00
parent 2dd83238af
commit 7f743e3c4e
5 changed files with 94 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
pom.xml
View File

@@ -122,6 +122,12 @@
<version>${cassandra-driver.version}</version>
</dependency>
<dependency>
<groupId>com.datastax.oss</groupId>
<artifactId>java-driver-core</artifactId>
<version>4.2.0</version>
</dependency>
<!-- CDI -->
<dependency>
<groupId>javax.enterprise</groupId>

9
spring-data-cassandra/pom.xml
View File

@@ -16,7 +16,9 @@
<name>Spring Data for Apache Cassandra Core</name>
<description>Cassandra support for Spring Data</description>
<url>https://github.com/spring-projects/spring-data-cassandra/tree/master/spring-data-cassandra</url>
<url>
https://github.com/spring-projects/spring-data-cassandra/tree/master/spring-data-cassandra
</url>
<properties>
<validation>1.1.0.Final</validation>
@@ -77,6 +79,11 @@
<artifactId>cassandra-driver-core</artifactId>
</dependency>
<dependency>
<groupId>com.datastax.oss</groupId>
<artifactId>java-driver-core</artifactId>
</dependency>
<!-- Reactor -->
<dependency>
<groupId>io.projectreactor</groupId>

13
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/SessionFactory.java
View File

@@ -16,6 +16,7 @@
package org.springframework.data.cassandra;
import com.datastax.driver.core.Session;
import com.datastax.oss.driver.api.core.CqlSession;
/**
* A factory for Apache Cassandra sessions.
@@ -43,4 +44,16 @@ public interface SessionFactory {
*/
Session getSession();
/**
* Attempts to establish a {@link Session} with the connection infrastructure that this {@link SessionFactory} object
* represents.
*
* @return a {@link Session} to Apache Cassandra.
*/
default CqlSession getCqlSession() {
// TODO
return null;
}
}

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

@@ -101,6 +101,31 @@ public abstract class AbstractCassandraConfiguration extends AbstractClusterConf
return session;
}
/**
* Creates a {@link CassandraCqlSessionFactoryBean} that provides a Cassandra
* {@link com.datastax.oss.driver.api.core.CqlSession}. The lifecycle of {@link CassandraSessionFactoryBean}
* initializes the {@link #getSchemaAction() schema} in the {@link #getKeyspaceName() configured keyspace}.
*
* @return the {@link CassandraSessionFactoryBean}.
* @see #cluster()
* @see #cassandraConverter()
* @see #getKeyspaceName()
* @see #getSchemaAction()
* @see #getStartupScripts()
* @see #getShutdownScripts()
*/
@Bean
public CqlSessionFactoryBean cassandraSession() {
CqlSessionFactoryBean session = new CqlSessionFactoryBean();
session.setContactPoints(getContactPoints());
session.setPort(getPort());
session.setKeyspaceName(getKeyspaceName());
return session;
}
/**
* Creates a {@link DefaultSessionFactory} using the configured {@link #session()} to be used with
* {@link org.springframework.data.cassandra.core.CassandraTemplate}.

87
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CqlSessionFactoryBean.java
View File

@@ -16,6 +16,7 @@
package org.springframework.data.cassandra.config;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -32,12 +33,9 @@ import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.cassandra.core.CassandraAdminOperations;
import org.springframework.data.cassandra.core.CassandraAdminTemplate;
import org.springframework.data.cassandra.core.CassandraPersistentEntitySchemaCreator;
import org.springframework.data.cassandra.core.CassandraPersistentEntitySchemaDropper;
import org.springframework.data.cassandra.core.convert.CassandraConverter;
import org.springframework.data.cassandra.core.cql.CqlOperations;
import org.springframework.data.cassandra.core.cql.CqlTemplate;
import org.springframework.data.cassandra.core.cql.generator.AlterKeyspaceCqlGenerator;
import org.springframework.data.cassandra.core.cql.generator.CreateKeyspaceCqlGenerator;
import org.springframework.data.cassandra.core.cql.generator.DropKeyspaceCqlGenerator;
@@ -52,8 +50,8 @@ import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Session;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.CqlSessionBuilder;
/**
* Factory for creating and configuring a Cassandra {@link Session}, which is a thread-safe singleton. As such, it is
@@ -65,7 +63,7 @@ import com.datastax.driver.core.Session;
* @author Mark Paluch
* @since 3.0
*/
public class CqlSessionFactoryBean implements FactoryBean<Session>, InitializingBean, DisposableBean {
public class CqlSessionFactoryBean implements FactoryBean<CqlSession>, InitializingBean, DisposableBean {
public static final int DEFAULT_PORT = 9042;
public static final String DEFAULT_CONTACT_POINTS = "localhost";
@@ -76,8 +74,8 @@ public class CqlSessionFactoryBean implements FactoryBean<Session>, Initializing
private static final boolean DEFAULT_DROP_TABLES = false;
private static final boolean DEFAULT_DROP_UNUSED_TABLES = false;
private @Nullable Cluster cluster;
private @Nullable Session session;
private @Nullable CqlSession systemSession;
private @Nullable CqlSession session;
private String contactPoints = DEFAULT_CONTACT_POINTS;
private int port = DEFAULT_PORT;
@@ -171,7 +169,7 @@ public class CqlSessionFactoryBean implements FactoryBean<Session>, Initializing
*/
public boolean isConnected() {
Session session = getObject();
CqlSession session = getObject();
return !(session == null || session.isClosed());
}
@@ -183,9 +181,9 @@ public class CqlSessionFactoryBean implements FactoryBean<Session>, Initializing
* @throws IllegalStateException if the Cassandra {@link Session} was not properly initialized.
* @see Session
*/
protected Session getSession() {
protected CqlSession getSession() {
Session session = getObject();
CqlSession session = getObject();
Assert.state(session != null, "Session was not properly initialized");
@@ -392,12 +390,16 @@ public class CqlSessionFactoryBean implements FactoryBean<Session>, Initializing
@Override
public void afterPropertiesSet() {
Cluster cluster = buildCluster();
this.cluster = cluster;
initializeCluster(cluster);
CqlSessionBuilder sessionBuilder = buildBuilder();
this.systemSession = sessionBuilder.build();
this.session = StringUtils.hasText(getKeyspaceName()) ? this.cluster.connect(getKeyspaceName())
: this.cluster.connect();
initializeCluster(this.systemSession);
if (StringUtils.hasText(getKeyspaceName())) {
sessionBuilder.withKeyspace(getKeyspaceName());
}
this.session = sessionBuilder.build();
executeScripts(getStartupScripts().stream(), this.session);
performSchemaAction();
@@ -412,31 +414,30 @@ public class CqlSessionFactoryBean implements FactoryBean<Session>, Initializing
if (session != null) {
executeScripts(getShutdownScripts().stream(), this.session);
getSession().close();
}
if (cluster != null) {
executeSpecsAndScripts(keyspaceDrops, keyspaceShutdownScripts, this.cluster);
cluster.close();
cluster = null;
executeSpecsAndScripts(keyspaceDrops, keyspaceShutdownScripts, this.systemSession);
systemSession.close();
session.close();
}
}
private Cluster buildCluster() {
private CqlSessionBuilder buildBuilder() {
Assert.hasText(this.contactPoints, "At least one server is required");
Cluster.Builder clusterBuilder = Cluster.builder()
.addContactPoints(StringUtils.commaDelimitedListToStringArray(this.contactPoints)).withPort(this.port);
CqlSessionBuilder builder = CqlSession.builder();
StringUtils.commaDelimitedListToSet(this.contactPoints).stream().forEach(host -> {
builder.addContactPoint(InetSocketAddress.createUnresolved(host, this.port));
});
if (StringUtils.hasText(this.username)) {
clusterBuilder = clusterBuilder.withCredentials(this.username, this.password);
builder.withAuthCredentials(this.username, this.password);
}
return clusterBuilder.build();
return builder;
}
private void initializeCluster(Cluster cluster) {
private void initializeCluster(CqlSession session) {
generateSpecificationsFromFactoryDeclarations();
@@ -446,20 +447,17 @@ public class CqlSessionFactoryBean implements FactoryBean<Session>, Initializing
startupSpecifications.addAll(this.keyspaceCreations);
startupSpecifications.addAll(this.keyspaceAlterations);
executeSpecsAndScripts(startupSpecifications, this.keyspaceStartupScripts, cluster);
executeSpecsAndScripts(startupSpecifications, this.keyspaceStartupScripts, session);
}
private void executeSpecsAndScripts(List<? extends KeyspaceActionSpecification> keyspaceActionSpecifications,
List<String> scripts, Cluster cluster) {
List<String> scripts, CqlSession session) {
if (!CollectionUtils.isEmpty(keyspaceActionSpecifications) || !CollectionUtils.isEmpty(scripts)) {
try (Session session = cluster.connect()) {
Stream<String> keyspaceActions = keyspaceActionSpecifications.stream().map(this::toCql);
Stream<String> keyspaceActions = keyspaceActionSpecifications.stream().map(this::toCql);
executeScripts(Stream.concat(keyspaceActions, scripts.stream()), session);
}
executeScripts(Stream.concat(keyspaceActions, scripts.stream()), session);
}
}
@@ -502,8 +500,9 @@ public class CqlSessionFactoryBean implements FactoryBean<Session>, Initializing
* statement.
*/
protected void createTables(boolean drop, boolean dropUnused, boolean ifNotExists) {
CassandraAdminTemplate adminTemplate = new CassandraAdminTemplate(this.session, converter);
performSchemaActions(drop, dropUnused, ifNotExists, adminTemplate);
// TODO
/*CassandraAdminTemplate adminTemplate = new CassandraAdminTemplate(this.session, converter);
performSchemaActions(drop, dropUnused, ifNotExists, adminTemplate);*/
}
private void performSchemaActions(boolean drop, boolean dropUnused, boolean ifNotExists,
@@ -531,7 +530,7 @@ public class CqlSessionFactoryBean implements FactoryBean<Session>, Initializing
* @see org.springframework.beans.factory.FactoryBean#getObject()
*/
@Override
public Session getObject() {
public CqlSession getObject() {
return this.session;
}
@@ -540,8 +539,8 @@ public class CqlSessionFactoryBean implements FactoryBean<Session>, Initializing
* @see org.springframework.beans.factory.FactoryBean#getObjectType()
*/
@Override
public Class<? extends Session> getObjectType() {
return this.session != null ? this.session.getClass() : Session.class;
public Class<? extends CqlSession> getObjectType() {
return CqlSession.class;
}
/*
@@ -556,13 +555,11 @@ public class CqlSessionFactoryBean implements FactoryBean<Session>, Initializing
/**
* Executes the given Cassandra CQL scripts. The {@link Session} must be connected when this method is called.
*/
private void executeScripts(Stream<String> scripts, Session session) {
CqlOperations template = new CqlTemplate(session);
private void executeScripts(Stream<String> scripts, CqlSession session) {
scripts.forEach(script -> {
logger.info("executing raw CQL [{}]", script);
template.execute(script);
session.execute(script);
});
}