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

Merge branch 'DATACASS-71'

Browse Source
This commit is contained in:
David Webb
2014-01-21 20:38:14 +00:00
parent ad40ac1c08 7f640c20de
commit 611b827504
27 changed files with 1706 additions and 663 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-cassandra/pom.xml
View File

@@ -41,6 +41,11 @@
<groupId>org.springframework</groupId>
<artifactId>spring-tx</artifactId>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>spring-data-commons</artifactId>
<version>${springdata.commons}</version>
</dependency>
<dependency>
<groupId>com.datastax.cassandra</groupId>
<artifactId>cassandra-driver-core</artifactId>

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

@@ -16,8 +16,10 @@
package org.springframework.cassandra.config;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,7 +31,7 @@ import org.springframework.cassandra.core.cql.generator.CreateKeyspaceCqlGenerat
import org.springframework.cassandra.core.cql.generator.DropKeyspaceCqlGenerator;
import org.springframework.cassandra.core.keyspace.CreateKeyspaceSpecification;
import org.springframework.cassandra.core.keyspace.DropKeyspaceSpecification;
import org.springframework.cassandra.core.keyspace.KeyspaceNameSpecification;
import org.springframework.cassandra.core.keyspace.KeyspaceActionSpecification;
import org.springframework.cassandra.support.CassandraExceptionTranslator;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.PersistenceExceptionTranslator;
@@ -37,7 +39,6 @@ import org.springframework.util.StringUtils;
import com.datastax.driver.core.AuthProvider;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.HostDistance;
import com.datastax.driver.core.PoolingOptions;
import com.datastax.driver.core.ProtocolOptions.Compression;
import com.datastax.driver.core.Session;
@@ -51,32 +52,39 @@ import com.datastax.driver.core.policies.RetryPolicy;
*
* @author Alex Shvid
* @author Matthew T. Adams
* @author David Webb
*/
public class CassandraClusterFactoryBean implements FactoryBean<Cluster>, InitializingBean, DisposableBean,
PersistenceExceptionTranslator {
public static final String DEFAULT_CONTACT_POINTS = "localhost";
public static final boolean DEFAULT_METRICS_ENABLED = true;
public static final boolean DEFAULT_DEFERRED_INITIALIZATION = false;
public static final boolean DEFAULT_JMX_REPORTING_ENABLED = true;
public static final int DEFAULT_PORT = 9042;
protected static final Logger log = LoggerFactory.getLogger(CassandraClusterFactoryBean.class);
private Cluster cluster;
/**
* Comma-delimited string of servers.
/*
* Attributes needed for cluster builder
*/
private String contactPoints = DEFAULT_CONTACT_POINTS;
private int port = CassandraClusterFactoryBean.DEFAULT_PORT;
private CompressionType compressionType;
private PoolingOptionsConfig localPoolingOptions;
private PoolingOptionsConfig remotePoolingOptions;
private SocketOptionsConfig socketOptions;
private PoolingOptions poolingOptions;
private SocketOptions socketOptions;
private AuthProvider authProvider;
private String username;
private String password;
private LoadBalancingPolicy loadBalancingPolicy;
private ReconnectionPolicy reconnectionPolicy;
private RetryPolicy retryPolicy;
private boolean metricsEnabled = DEFAULT_METRICS_ENABLED;
private boolean deferredInitialization = DEFAULT_DEFERRED_INITIALIZATION;
private boolean jmxReportingEnabled = DEFAULT_JMX_REPORTING_ENABLED;
private Set<KeyspaceActionSpecification<?>> keyspaceSpecifications = new HashSet<KeyspaceActionSpecification<?>>();
private List<CreateKeyspaceSpecification> keyspaceCreations = new ArrayList<CreateKeyspaceSpecification>();
private List<DropKeyspaceSpecification> keyspaceDrops = new ArrayList<DropKeyspaceSpecification>();
private List<String> startupScripts = new ArrayList<String>();
@@ -119,20 +127,20 @@ public class CassandraClusterFactoryBean implements FactoryBean<Cluster>, Initia
builder.withCompression(convertCompressionType(compressionType));
}
if (localPoolingOptions != null) {
builder.withPoolingOptions(configPoolingOptions(HostDistance.LOCAL, localPoolingOptions));
}
if (remotePoolingOptions != null) {
builder.withPoolingOptions(configPoolingOptions(HostDistance.REMOTE, remotePoolingOptions));
if (poolingOptions != null) {
builder.withPoolingOptions(poolingOptions);
}
if (socketOptions != null) {
builder.withSocketOptions(configSocketOptions(socketOptions));
builder.withSocketOptions(socketOptions);
}
if (authProvider != null) {
builder.withAuthProvider(authProvider);
if (username != null) {
builder.withCredentials(username, password);
}
}
if (loadBalancingPolicy != null) {
@@ -147,14 +155,44 @@ public class CassandraClusterFactoryBean implements FactoryBean<Cluster>, Initia
builder.withRetryPolicy(retryPolicy);
}
if (deferredInitialization) {
builder.withDeferredInitialization();
}
if (!metricsEnabled) {
builder.withoutMetrics();
}
if (!jmxReportingEnabled) {
builder.withoutJMXReporting();
}
cluster = builder.build();
generateSpecificationsFromFactoryBeans();
executeSpecsAndScripts(keyspaceCreations, startupScripts);
}
/**
* Examines the contents of all the KeyspaceSpecificationFactoryBeans and generates the proper KeyspaceSpecification
* from them.
*/
private void generateSpecificationsFromFactoryBeans() {
for (KeyspaceActionSpecification<?> spec : keyspaceSpecifications) {
if (spec instanceof CreateKeyspaceSpecification) {
keyspaceCreations.add((CreateKeyspaceSpecification) spec);
}
if (spec instanceof DropKeyspaceSpecification) {
keyspaceDrops.add((DropKeyspaceSpecification) spec);
}
}
}
protected void executeSpecsAndScripts(@SuppressWarnings("rawtypes") List specs, List<String> scripts) {
Session system = null;
@@ -167,7 +205,7 @@ public class CassandraClusterFactoryBean implements FactoryBean<Cluster>, Initia
Iterator<?> i = specs.iterator();
while (i.hasNext()) {
KeyspaceNameSpecification<?> spec = (KeyspaceNameSpecification<?>) i.next();
KeyspaceActionSpecification<?> spec = (KeyspaceActionSpecification<?>) i.next();
String cql = (spec instanceof CreateKeyspaceSpecification) ? new CreateKeyspaceCqlGenerator(
(CreateKeyspaceSpecification) spec).toCql() : new DropKeyspaceCqlGenerator(
(DropKeyspaceSpecification) spec).toCql();
@@ -229,15 +267,11 @@ public class CassandraClusterFactoryBean implements FactoryBean<Cluster>, Initia
this.compressionType = compressionType;
}
public void setLocalPoolingOptions(PoolingOptionsConfig localPoolingOptions) {
this.localPoolingOptions = localPoolingOptions;
public void setPoolingOptions(PoolingOptions poolingOptions) {
this.poolingOptions = poolingOptions;
}
public void setRemotePoolingOptions(PoolingOptionsConfig remotePoolingOptions) {
this.remotePoolingOptions = remotePoolingOptions;
}
public void setSocketOptions(SocketOptionsConfig socketOptions) {
public void setSocketOptions(SocketOptions socketOptions) {
this.socketOptions = socketOptions;
}
@@ -295,52 +329,48 @@ public class CassandraClusterFactoryBean implements FactoryBean<Cluster>, Initia
throw new IllegalArgumentException("unknown compression type " + type);
}
private static PoolingOptions configPoolingOptions(HostDistance hostDistance, PoolingOptionsConfig config) {
PoolingOptions poolingOptions = new PoolingOptions();
if (config.getMinSimultaneousRequests() != null) {
poolingOptions
.setMinSimultaneousRequestsPerConnectionThreshold(hostDistance, config.getMinSimultaneousRequests());
}
if (config.getMaxSimultaneousRequests() != null) {
poolingOptions
.setMaxSimultaneousRequestsPerConnectionThreshold(hostDistance, config.getMaxSimultaneousRequests());
}
if (config.getCoreConnections() != null) {
poolingOptions.setCoreConnectionsPerHost(hostDistance, config.getCoreConnections());
}
if (config.getMaxConnections() != null) {
poolingOptions.setMaxConnectionsPerHost(hostDistance, config.getMaxConnections());
}
return poolingOptions;
/**
* @return Returns the keyspaceSpecifications.
*/
public Set<KeyspaceActionSpecification<?>> getKeyspaceSpecifications() {
return keyspaceSpecifications;
}
private static SocketOptions configSocketOptions(SocketOptionsConfig config) {
SocketOptions socketOptions = new SocketOptions();
/**
* If accumlating is true, we append to the list, otherwise we replace the list.
*
* @param keyspaceSpecifications The keyspaceSpecifications to set.
*/
public void setKeyspaceSpecifications(Set<KeyspaceActionSpecification<?>> keyspaceSpecifications) {
log.info("Setter Called");
this.keyspaceSpecifications = keyspaceSpecifications;
}
if (config.getConnectTimeoutMls() != null) {
socketOptions.setConnectTimeoutMillis(config.getConnectTimeoutMls());
}
if (config.getKeepAlive() != null) {
socketOptions.setKeepAlive(config.getKeepAlive());
}
if (config.getReuseAddress() != null) {
socketOptions.setReuseAddress(config.getReuseAddress());
}
if (config.getSoLinger() != null) {
socketOptions.setSoLinger(config.getSoLinger());
}
if (config.getTcpNoDelay() != null) {
socketOptions.setTcpNoDelay(config.getTcpNoDelay());
}
if (config.getReceiveBufferSize() != null) {
socketOptions.setReceiveBufferSize(config.getReceiveBufferSize());
}
if (config.getSendBufferSize() != null) {
socketOptions.setSendBufferSize(config.getSendBufferSize());
}
/**
* @param username The username to set.
*/
public void setUsername(String username) {
this.username = username;
}
return socketOptions;
/**
* @param password The password to set.
*/
public void setPassword(String password) {
this.password = password;
}
/**
* @param deferredInitialization The deferredInitialization to set.
*/
public void setDeferredInitialization(boolean deferredInitialization) {
this.deferredInitialization = deferredInitialization;
}
/**
* @param jmxReportingEnabled The jmxReportingEnabled to set.
*/
public void setJmxReportingEnabled(boolean jmxReportingEnabled) {
this.jmxReportingEnabled = jmxReportingEnabled;
}
}

25
spring-cassandra/src/main/java/org/springframework/cassandra/config/KeyspaceAction.java Normal file
View File

@@ -0,0 +1,25 @@
/*
* Copyright 2010-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.config;
/**
* Available actions for Keyspace Specifications
*
* @author David Webb
*/
public enum KeyspaceAction {
CREATE, CREATE_DROP, ALTER;
}

267
spring-cassandra/src/main/java/org/springframework/cassandra/config/KeyspaceActionSpecificationFactoryBean.java Normal file
View File

@@ -0,0 +1,267 @@
/*
* Copyright 2011-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.config;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.cassandra.core.keyspace.CreateKeyspaceSpecification;
import org.springframework.cassandra.core.keyspace.DefaultOption;
import org.springframework.cassandra.core.keyspace.DropKeyspaceSpecification;
import org.springframework.cassandra.core.keyspace.KeyspaceActionSpecification;
import org.springframework.cassandra.core.keyspace.KeyspaceOption;
import org.springframework.cassandra.core.keyspace.KeyspaceOption.ReplicationStrategy;
import org.springframework.cassandra.core.keyspace.Option;
import org.springframework.util.Assert;
/**
* A single keyspace XML Element can result in multiple actions. Example: {@literal CREATE_DROP}.
*
* This FactoryBean inspects the action required to satisfy the keyspace element, and then returns a Set of atomic
* {@link KeyspaceActionSpecification} required to satisfy the configuration action.
*
* @author David Webb
*
*/
public class KeyspaceActionSpecificationFactoryBean implements FactoryBean<Set<KeyspaceActionSpecification<?>>>,
InitializingBean, DisposableBean {
private final static Logger log = LoggerFactory.getLogger(KeyspaceActionSpecificationFactoryBean.class);
private KeyspaceAction action;
private String name;
private List<String> networkTopologyDataCenters = new LinkedList<String>();
private List<String> networkTopologyReplicationFactors = new LinkedList<String>();
private String replicationStrategy;
private long replicationFactor;
private boolean durableWrites = false;
private boolean ifNotExists = false;
private Set<KeyspaceActionSpecification<?>> specs = new HashSet<KeyspaceActionSpecification<?>>();
@Override
public void destroy() throws Exception {
action = null;
name = null;
networkTopologyDataCenters = null;
networkTopologyReplicationFactors = null;
replicationStrategy = null;
specs = null;
}
@Override
public void afterPropertiesSet() throws Exception {
Assert.hasText(name, "Keyspace Name is required for a Keyspace Action");
Assert.notNull(action, "Keyspace Action is required for a Keyspace Action");
switch (action) {
case CREATE_DROP:
specs.add(generateDropKeyspaceSpecification());
case CREATE:
// Assert.notNull(replicationStrategy, "Replication Strategy is required to create a Keyspace");
specs.add(generateCreateKeyspaceSpecification());
break;
case ALTER:
break;
}
}
/**
* Generate a {@link CreateKeyspaceSpecification} for the keyspace.
*
* @return The {@link CreateKeyspaceSpecification}
*/
private CreateKeyspaceSpecification generateCreateKeyspaceSpecification() {
CreateKeyspaceSpecification create = new CreateKeyspaceSpecification();
create.name(name).ifNotExists(ifNotExists).with(KeyspaceOption.DURABLE_WRITES, durableWrites);
Map<Option, Object> replicationStrategyMap = new HashMap<Option, Object>();
replicationStrategyMap.put(new DefaultOption("class", String.class, true, false, true), ReplicationStrategy
.valueOf(replicationStrategy).getValue());
/*
* Just set replication factor for SimpleStrategy
*/
if (replicationStrategy.equals(ReplicationStrategy.SIMPLE_STRATEGY.name())) {
replicationStrategyMap.put(new DefaultOption("replication_factor", Long.class, true, false, false),
replicationFactor);
}
if (replicationStrategy.equals(ReplicationStrategy.NETWORK_TOPOLOGY_STRATEGY.name())) {
int i = 0;
for (String datacenter : networkTopologyDataCenters) {
replicationStrategyMap.put(new DefaultOption(datacenter, Long.class, true, false, false),
networkTopologyReplicationFactors.get(i));
i++;
}
}
create.with(KeyspaceOption.REPLICATION, replicationStrategyMap);
return create;
}
/**
* Generate a {@link DropKeyspaceSpecification} for the keyspace.
*
* @return The {@link DropKeyspaceSpecification}
*/
private DropKeyspaceSpecification generateDropKeyspaceSpecification() {
DropKeyspaceSpecification drop = new DropKeyspaceSpecification();
drop.name(getName());
return drop;
}
@Override
public Set<KeyspaceActionSpecification<?>> getObject() throws Exception {
return specs;
}
@Override
public Class<?> getObjectType() {
return Set.class;
}
@Override
public boolean isSingleton() {
return false;
}
/**
* @return Returns the name.
*/
public String getName() {
return name;
}
/**
* @param name The name to set.
*/
public void setName(String name) {
this.name = name;
}
/**
* @return Returns the ifNotExists.
*/
public boolean isIfNotExists() {
return ifNotExists;
}
/**
* @param ifNotExists The ifNotExists to set.
*/
public void setIfNotExists(boolean ifNotExists) {
this.ifNotExists = ifNotExists;
}
/**
* @return Returns the action.
*/
public KeyspaceAction getAction() {
return action;
}
/**
* @param action The action to set.
*/
public void setAction(KeyspaceAction action) {
this.action = action;
}
/**
* @return Returns the durableWrites.
*/
public boolean isDurableWrites() {
return durableWrites;
}
/**
* @param durableWrites The durableWrites to set.
*/
public void setDurableWrites(boolean durableWrites) {
this.durableWrites = durableWrites;
}
/**
* @return Returns the replicationStrategy.
*/
public String getReplicationStrategy() {
return replicationStrategy;
}
/**
* @param replicationStrategy The replicationStrategy to set.
*/
public void setReplicationStrategy(String replicationStrategy) {
this.replicationStrategy = replicationStrategy;
}
/**
* @return Returns the networkTopologyDataCenters.
*/
public List<String> getNetworkTopologyDataCenters() {
return networkTopologyDataCenters;
}
/**
* @param networkTopologyDataCenters The networkTopologyDataCenters to set.
*/
public void setNetworkTopologyDataCenters(List<String> networkTopologyDataCenters) {
this.networkTopologyDataCenters = networkTopologyDataCenters;
}
/**
* @return Returns the networkTopologyReplicationFactors.
*/
public List<String> getNetworkTopologyReplicationFactors() {
return networkTopologyReplicationFactors;
}
/**
* @param networkTopologyReplicationFactors The networkTopologyReplicationFactors to set.
*/
public void setNetworkTopologyReplicationFactors(List<String> networkTopologyReplicationFactors) {
this.networkTopologyReplicationFactors = networkTopologyReplicationFactors;
}
/**
* @return Returns the replicationFactor.
*/
public long getReplicationFactor() {
return replicationFactor;
}
/**
* @param replicationFactor The replicationFactor to set.
*/
public void setReplicationFactor(long replicationFactor) {
this.replicationFactor = replicationFactor;
}
}

71
spring-cassandra/src/main/java/org/springframework/cassandra/config/MultiLevelListFlattenerFactoryBean.java Normal file
View File

@@ -0,0 +1,71 @@
/*
* Copyright 2011-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.config;
import java.util.ArrayList;
import java.util.List;
import org.springframework.beans.factory.FactoryBean;
/**
* Given List of Lists where all child Lists contain the same class, then a single level List of <T> is generated.
*
* @author David Webb
* @param <T>
*
*/
public class MultiLevelListFlattenerFactoryBean<T> implements FactoryBean<List<T>> {
private List<List<T>> multiLevelList;
@Override
public List<T> getObject() throws Exception {
List<T> list = new ArrayList<T>();
for (List<T> topList : multiLevelList) {
for (T t : topList) {
list.add(t);
}
}
return list;
}
@Override
public Class<?> getObjectType() {
return List.class;
}
@Override
public boolean isSingleton() {
return true;
}
/**
* @return Returns the multiLevelList.
*/
public List<List<T>> getMultiLevelList() {
return multiLevelList;
}
/**
* @param multiLevelList The multiLevelList to set.
*/
public void setMultiLevelList(List<List<T>> multiLevelList) {
this.multiLevelList = multiLevelList;
}
}

77
spring-cassandra/src/main/java/org/springframework/cassandra/config/MultiLevelSetFlattenerFactoryBean.java Normal file
View File

@@ -0,0 +1,77 @@
/*
* Copyright 2011-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.config;
import java.util.HashSet;
import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.FactoryBean;
/**
* Given Set of Sets where all child Sets contain the same class, then a single level Set of <T> is generated.
*
* @author David Webb
* @param <T>
*
*/
public class MultiLevelSetFlattenerFactoryBean<T> implements FactoryBean<Set<T>> {
private final static Logger log = LoggerFactory.getLogger(MultiLevelSetFlattenerFactoryBean.class);
private Set<Set<T>> multiLevelSet;
@Override
public Set<T> getObject() throws Exception {
Set<T> set = new HashSet<T>();
for (Set<T> topSet : multiLevelSet) {
for (T t : topSet) {
log.info(t.toString());
log.info("Set contains -> " + set.contains(t));
set.add(t);
}
}
return set;
}
@Override
public Class<?> getObjectType() {
return Set.class;
}
@Override
public boolean isSingleton() {
return true;
}
/**
* @return Returns the multiLevelSet.
*/
public Set<Set<T>> getMultiLevelSet() {
return multiLevelSet;
}
/**
* @param multiLevelSet The multiLevelSet to set.
*/
public void setMultiLevelSet(Set<Set<T>> multiLevelSet) {
this.multiLevelSet = multiLevelSet;
}
}

62
spring-cassandra/src/main/java/org/springframework/cassandra/config/PoolingOptionsConfig.java
View File

@@ -1,62 +0,0 @@
/*
* Copyright 2011-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.config;
/**
* Pooling options.
*
* @author Alex Shvid
* @author Matthew T. Adams
*/
public class PoolingOptionsConfig {
private Integer minSimultaneousRequests;
private Integer maxSimultaneousRequests;
private Integer coreConnections;
private Integer maxConnections;
public Integer getMinSimultaneousRequests() {
return minSimultaneousRequests;
}
public void setMinSimultaneousRequests(Integer minSimultaneousRequests) {
this.minSimultaneousRequests = minSimultaneousRequests;
}
public Integer getMaxSimultaneousRequests() {
return maxSimultaneousRequests;
}
public void setMaxSimultaneousRequests(Integer maxSimultaneousRequests) {
this.maxSimultaneousRequests = maxSimultaneousRequests;
}
public Integer getCoreConnections() {
return coreConnections;
}
public void setCoreConnections(Integer coreConnections) {
this.coreConnections = coreConnections;
}
public Integer getMaxConnections() {
return maxConnections;
}
public void setMaxConnections(Integer maxConnections) {
this.maxConnections = maxConnections;
}
}

224
spring-cassandra/src/main/java/org/springframework/cassandra/config/PoolingOptionsFactoryBean.java Normal file
View File

@@ -0,0 +1,224 @@
/*
* Copyright 2011-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.config;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import com.datastax.driver.core.HostDistance;
import com.datastax.driver.core.PoolingOptions;
/**
* Pooling Options Factory Bean.
*
* @author Matthew T. Adams
* @author David Webb
*/
public class PoolingOptionsFactoryBean implements FactoryBean<PoolingOptions>, InitializingBean, DisposableBean {
private Integer localMinSimultaneousRequests;
private Integer localMaxSimultaneousRequests;
private Integer localCoreConnections;
private Integer localMaxConnections;
private Integer remoteMinSimultaneousRequests;
private Integer remoteMaxSimultaneousRequests;
private Integer remoteCoreConnections;
private Integer remoteMaxConnections;
PoolingOptions poolingOptions;
@Override
public void destroy() throws Exception {
localMinSimultaneousRequests = null;
localMaxSimultaneousRequests = null;
localCoreConnections = null;
localMaxConnections = null;
remoteMinSimultaneousRequests = null;
remoteMaxSimultaneousRequests = null;
remoteCoreConnections = null;
remoteMaxConnections = null;
}
@Override
public void afterPropertiesSet() throws Exception {
poolingOptions = new PoolingOptions();
if (localMinSimultaneousRequests != null) {
poolingOptions.setMinSimultaneousRequestsPerConnectionThreshold(HostDistance.LOCAL, localMinSimultaneousRequests);
}
if (localMaxSimultaneousRequests != null) {
poolingOptions.setMaxSimultaneousRequestsPerConnectionThreshold(HostDistance.LOCAL, localMaxSimultaneousRequests);
}
if (localCoreConnections != null) {
poolingOptions.setCoreConnectionsPerHost(HostDistance.LOCAL, localCoreConnections);
}
if (localMaxConnections != null) {
poolingOptions.setMaxConnectionsPerHost(HostDistance.LOCAL, localMaxConnections);
}
if (remoteMinSimultaneousRequests != null) {
poolingOptions.setMinSimultaneousRequestsPerConnectionThreshold(HostDistance.REMOTE,
remoteMinSimultaneousRequests);
}
if (remoteMaxSimultaneousRequests != null) {
poolingOptions.setMaxSimultaneousRequestsPerConnectionThreshold(HostDistance.REMOTE,
remoteMaxSimultaneousRequests);
}
if (remoteCoreConnections != null) {
poolingOptions.setCoreConnectionsPerHost(HostDistance.REMOTE, remoteCoreConnections);
}
if (remoteMaxConnections != null) {
poolingOptions.setMaxConnectionsPerHost(HostDistance.REMOTE, remoteMaxConnections);
}
}
@Override
public PoolingOptions getObject() throws Exception {
return poolingOptions;
}
@Override
public Class<?> getObjectType() {
return PoolingOptions.class;
}
@Override
public boolean isSingleton() {
return true;
}
/**
* @return Returns the localMinSimultaneousRequests.
*/
public Integer getLocalMinSimultaneousRequests() {
return localMinSimultaneousRequests;
}
/**
* @param localMinSimultaneousRequests The localMinSimultaneousRequests to set.
*/
public void setLocalMinSimultaneousRequests(Integer localMinSimultaneousRequests) {
this.localMinSimultaneousRequests = localMinSimultaneousRequests;
}
/**
* @return Returns the localMaxSimultaneousRequests.
*/
public Integer getLocalMaxSimultaneousRequests() {
return localMaxSimultaneousRequests;
}
/**
* @param localMaxSimultaneousRequests The localMaxSimultaneousRequests to set.
*/
public void setLocalMaxSimultaneousRequests(Integer localMaxSimultaneousRequests) {
this.localMaxSimultaneousRequests = localMaxSimultaneousRequests;
}
/**
* @return Returns the localCoreConnections.
*/
public Integer getLocalCoreConnections() {
return localCoreConnections;
}
/**
* @param localCoreConnections The localCoreConnections to set.
*/
public void setLocalCoreConnections(Integer localCoreConnections) {
this.localCoreConnections = localCoreConnections;
}
/**
* @return Returns the localMaxConnections.
*/
public Integer getLocalMaxConnections() {
return localMaxConnections;
}
/**
* @param localMaxConnections The localMaxConnections to set.
*/
public void setLocalMaxConnections(Integer localMaxConnections) {
this.localMaxConnections = localMaxConnections;
}
/**
* @return Returns the remoteMinSimultaneousRequests.
*/
public Integer getRemoteMinSimultaneousRequests() {
return remoteMinSimultaneousRequests;
}
/**
* @param remoteMinSimultaneousRequests The remoteMinSimultaneousRequests to set.
*/
public void setRemoteMinSimultaneousRequests(Integer remoteMinSimultaneousRequests) {
this.remoteMinSimultaneousRequests = remoteMinSimultaneousRequests;
}
/**
* @return Returns the remoteMaxSimultaneousRequests.
*/
public Integer getRemoteMaxSimultaneousRequests() {
return remoteMaxSimultaneousRequests;
}
/**
* @param remoteMaxSimultaneousRequests The remoteMaxSimultaneousRequests to set.
*/
public void setRemoteMaxSimultaneousRequests(Integer remoteMaxSimultaneousRequests) {
this.remoteMaxSimultaneousRequests = remoteMaxSimultaneousRequests;
}
/**
* @return Returns the remoteCoreConnections.
*/
public Integer getRemoteCoreConnections() {
return remoteCoreConnections;
}
/**
* @param remoteCoreConnections The remoteCoreConnections to set.
*/
public void setRemoteCoreConnections(Integer remoteCoreConnections) {
this.remoteCoreConnections = remoteCoreConnections;
}
/**
* @return Returns the remoteMaxConnections.
*/
public Integer getRemoteMaxConnections() {
return remoteMaxConnections;
}
/**
* @param remoteMaxConnections The remoteMaxConnections to set.
*/
public void setRemoteMaxConnections(Integer remoteMaxConnections) {
this.remoteMaxConnections = remoteMaxConnections;
}
}

89
spring-cassandra/src/main/java/org/springframework/cassandra/config/SocketOptionsConfig.java
View File

@@ -1,89 +0,0 @@
/*
* Copyright 2011-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.config;
/**
* Socket options.
*
* @author Alex Shvid
* @author Matthew T. Adams
*/
public class SocketOptionsConfig {
private Integer connectTimeoutMls;
private Boolean keepAlive;
private Boolean reuseAddress;
private Integer soLinger;
private Boolean tcpNoDelay;
private Integer receiveBufferSize;
private Integer sendBufferSize;
public Integer getConnectTimeoutMls() {
return connectTimeoutMls;
}
public void setConnectTimeoutMls(Integer connectTimeoutMls) {
this.connectTimeoutMls = connectTimeoutMls;
}
public Boolean getKeepAlive() {
return keepAlive;
}
public void setKeepAlive(Boolean keepAlive) {
this.keepAlive = keepAlive;
}
public Boolean getReuseAddress() {
return reuseAddress;
}
public void setReuseAddress(Boolean reuseAddress) {
this.reuseAddress = reuseAddress;
}
public Integer getSoLinger() {
return soLinger;
}
public void setSoLinger(Integer soLinger) {
this.soLinger = soLinger;
}
public Boolean getTcpNoDelay() {
return tcpNoDelay;
}
public void setTcpNoDelay(Boolean tcpNoDelay) {
this.tcpNoDelay = tcpNoDelay;
}
public Integer getReceiveBufferSize() {
return receiveBufferSize;
}
public void setReceiveBufferSize(Integer receiveBufferSize) {
this.receiveBufferSize = receiveBufferSize;
}
public Integer getSendBufferSize() {
return sendBufferSize;
}
public void setSendBufferSize(Integer sendBufferSize) {
this.sendBufferSize = sendBufferSize;
}
}

185
spring-cassandra/src/main/java/org/springframework/cassandra/config/SocketOptionsFactoryBean.java Normal file
View File

@@ -0,0 +1,185 @@
/*
* Copyright 2011-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.config;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import com.datastax.driver.core.SocketOptions;
/**
* Socket Options Factory Bean.
*
* @author Matthew T. Adams
* @author David Webb
*/
public class SocketOptionsFactoryBean implements FactoryBean<SocketOptions>, InitializingBean, DisposableBean {
private Integer connectTimeoutMillis;
private Boolean keepAlive;
private Integer readTimeoutMillis;
private Boolean reuseAddress;
private Integer soLinger;
private Boolean tcpNoDelay;
private Integer receiveBufferSize;
private Integer sendBufferSize;
SocketOptions socketOptions;
@Override
public SocketOptions getObject() throws Exception {
return socketOptions;
}
@Override
public Class<?> getObjectType() {
return SocketOptions.class;
}
@Override
public void destroy() throws Exception {
connectTimeoutMillis = null;
keepAlive = null;
readTimeoutMillis = null;
reuseAddress = null;
soLinger = null;
tcpNoDelay = null;
receiveBufferSize = null;
sendBufferSize = null;
}
@Override
public void afterPropertiesSet() throws Exception {
socketOptions = new SocketOptions();
if (connectTimeoutMillis != null) {
socketOptions.setConnectTimeoutMillis(connectTimeoutMillis);
}
if (keepAlive != null) {
socketOptions.setKeepAlive(keepAlive);
}
if (readTimeoutMillis != null) {
socketOptions.setReadTimeoutMillis(readTimeoutMillis);
}
if (reuseAddress != null) {
socketOptions.setReuseAddress(reuseAddress);
}
if (soLinger != null) {
socketOptions.setSoLinger(soLinger);
}
if (tcpNoDelay != null) {
socketOptions.setTcpNoDelay(tcpNoDelay);
}
if (receiveBufferSize != null) {
socketOptions.setReceiveBufferSize(receiveBufferSize);
}
if (sendBufferSize != null) {
socketOptions.setSendBufferSize(sendBufferSize);
}
}
@Override
public boolean isSingleton() {
return true;
}
public Boolean getKeepAlive() {
return keepAlive;
}
public void setKeepAlive(Boolean keepAlive) {
this.keepAlive = keepAlive;
}
public Boolean getReuseAddress() {
return reuseAddress;
}
public void setReuseAddress(Boolean reuseAddress) {
this.reuseAddress = reuseAddress;
}
public Integer getSoLinger() {
return soLinger;
}
public void setSoLinger(Integer soLinger) {
this.soLinger = soLinger;
}
public Boolean getTcpNoDelay() {
return tcpNoDelay;
}
public void setTcpNoDelay(Boolean tcpNoDelay) {
this.tcpNoDelay = tcpNoDelay;
}
public Integer getReceiveBufferSize() {
return receiveBufferSize;
}
public void setReceiveBufferSize(Integer receiveBufferSize) {
this.receiveBufferSize = receiveBufferSize;
}
public Integer getSendBufferSize() {
return sendBufferSize;
}
public void setSendBufferSize(Integer sendBufferSize) {
this.sendBufferSize = sendBufferSize;
}
/**
* @return Returns the connectTimeoutMillis.
*/
public Integer getConnectTimeoutMillis() {
return connectTimeoutMillis;
}
/**
* @param connectTimeoutMillis The connectTimeoutMillis to set.
*/
public void setConnectTimeoutMillis(Integer connectTimeoutMillis) {
this.connectTimeoutMillis = connectTimeoutMillis;
}
/**
* @return Returns the readTimeoutMillis.
*/
public Integer getReadTimeoutMillis() {
return readTimeoutMillis;
}
/**
* @param readTimeoutMillis The readTimeoutMillis to set.
*/
public void setReadTimeoutMillis(Integer readTimeoutMillis) {
this.readTimeoutMillis = readTimeoutMillis;
}
}

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

@@ -6,10 +6,6 @@ import java.util.List;
import org.springframework.cassandra.config.CassandraClusterFactoryBean;
import org.springframework.cassandra.config.CassandraSessionFactoryBean;
import org.springframework.cassandra.config.CompressionType;
import org.springframework.cassandra.config.PoolingOptionsConfig;
import org.springframework.cassandra.config.SocketOptionsConfig;
import org.springframework.cassandra.core.CassandraOperations;
import org.springframework.cassandra.core.CassandraTemplate;
import org.springframework.cassandra.core.keyspace.CreateKeyspaceSpecification;
import org.springframework.cassandra.core.keyspace.DropKeyspaceSpecification;
import org.springframework.context.annotation.Bean;
@@ -17,6 +13,8 @@ import org.springframework.context.annotation.Configuration;
import com.datastax.driver.core.AuthProvider;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.PoolingOptions;
import com.datastax.driver.core.SocketOptions;
import com.datastax.driver.core.policies.LoadBalancingPolicy;
import com.datastax.driver.core.policies.ReconnectionPolicy;
import com.datastax.driver.core.policies.RetryPolicy;
@@ -42,11 +40,10 @@ public abstract class AbstractCassandraConfiguration {
bean.setKeyspaceCreations(getKeyspaceCreations());
bean.setKeyspaceDrops(getKeyspaceDrops());
bean.setLoadBalancingPolicy(getLoadBalancingPolicy());
bean.setLocalPoolingOptions(getLocalPoolingOptions());
bean.setMetricsEnabled(getMetricsEnabled());
bean.setPort(getPort());
bean.setReconnectionPolicy(getReconnectionPolicy());
bean.setRemotePoolingOptions(getRemotePoolingOptions());
bean.setPoolingOptions(getPoolingOptions());
bean.setRetryPolicy(getRetryPolicy());
bean.setShutdownScripts(getShutdownScripts());
bean.setSocketOptions(getSocketOptions());
@@ -71,7 +68,7 @@ public abstract class AbstractCassandraConfiguration {
return Collections.emptyList();
}
protected SocketOptionsConfig getSocketOptions() {
protected SocketOptions getSocketOptions() {
return null;
}
@@ -87,7 +84,7 @@ public abstract class AbstractCassandraConfiguration {
return null;
}
protected PoolingOptionsConfig getRemotePoolingOptions() {
protected PoolingOptions getPoolingOptions() {
return null;
}
@@ -99,10 +96,6 @@ public abstract class AbstractCassandraConfiguration {
return CassandraClusterFactoryBean.DEFAULT_METRICS_ENABLED;
}
protected PoolingOptionsConfig getLocalPoolingOptions() {
return null;
}
protected LoadBalancingPolicy getLoadBalancingPolicy() {
return null;
}

371
spring-cassandra/src/main/java/org/springframework/cassandra/config/xml/CassandraClusterParser.java
View File

@@ -15,44 +15,47 @@
*/
package org.springframework.cassandra.config.xml;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import static org.springframework.data.config.ParsingUtils.getSourceBeanDefinition;
import java.util.ArrayList;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.beans.factory.support.ManagedSet;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.cassandra.config.CassandraClusterFactoryBean;
import org.springframework.cassandra.config.CompressionType;
import org.springframework.cassandra.config.KeyspaceActionSpecificationFactoryBean;
import org.springframework.cassandra.config.KeyspaceAttributes;
import org.springframework.cassandra.config.PoolingOptionsConfig;
import org.springframework.cassandra.config.SocketOptionsConfig;
import org.springframework.cassandra.core.keyspace.CreateKeyspaceSpecification;
import org.springframework.cassandra.core.keyspace.DefaultOption;
import org.springframework.cassandra.core.keyspace.DropKeyspaceSpecification;
import org.springframework.cassandra.core.keyspace.KeyspaceOption;
import org.springframework.cassandra.core.keyspace.Option;
import org.springframework.cassandra.config.MultiLevelSetFlattenerFactoryBean;
import org.springframework.cassandra.config.PoolingOptionsFactoryBean;
import org.springframework.cassandra.config.SocketOptionsFactoryBean;
import org.springframework.cassandra.core.keyspace.KeyspaceActionSpecification;
import org.springframework.cassandra.core.keyspace.KeyspaceOption.ReplicationStrategy;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
import org.w3c.dom.NodeList;
import com.datastax.driver.core.HostDistance;
import com.datastax.driver.core.PoolingOptions;
import com.datastax.driver.core.SocketOptions;
/**
* Parser for &lt;cluster;gt; definitions.
* Parses the {@literal <cluster>} element of the XML Configuration.
*
* @author Alex Shvid
* @author Matthew T. Adams
* @author David Webb
*/
public class CassandraClusterParser extends AbstractSimpleBeanDefinitionParser {
public class CassandraClusterParser extends AbstractBeanDefinitionParser {
@Override
protected Class<?> getBeanClass(Element element) {
return CassandraClusterFactoryBean.class;
}
private final static Logger log = LoggerFactory.getLogger(CassandraClusterParser.class);
@Override
protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
@@ -63,7 +66,34 @@ public class CassandraClusterParser extends AbstractSimpleBeanDefinitionParser {
}
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(CassandraClusterFactoryBean.class);
builder.getRawBeanDefinition().setSource(parserContext.extractSource(element));
builder.getRawBeanDefinition().setDestroyMethodName("destroy");
if (parserContext.isNested()) {
// Inner bean definition must receive same scope as containing bean.
builder.setScope(parserContext.getContainingBeanDefinition().getScope());
}
if (parserContext.isDefaultLazyInit()) {
// Default-lazy-init applies to custom bean definitions as well.
builder.setLazyInit(true);
}
doParse(element, parserContext, builder);
return builder.getBeanDefinition();
}
/**
* Parses the attributes on the top level element, then parses all children.
*
* @param element The Element being parsed
* @param context The Parser Context
* @param builder The parent {@link BeanDefinitionBuilder}
*/
protected void doParse(Element element, ParserContext context, BeanDefinitionBuilder builder) {
String contactPoints = element.getAttribute("contactPoints");
if (StringUtils.hasText(contactPoints)) {
@@ -77,41 +107,98 @@ public class CassandraClusterParser extends AbstractSimpleBeanDefinitionParser {
String compression = element.getAttribute("compression");
if (StringUtils.hasText(compression)) {
builder.addPropertyValue("compressionType", CompressionType.valueOf(compression));
builder.addPropertyValue("compressionType", compression);
}
parseChildElements(builder, element);
String authProvider = element.getAttribute("auth-info-provider-ref");
if (StringUtils.hasText(authProvider)) {
builder.addPropertyReference("authProvider", authProvider);
}
String username = element.getAttribute("username");
if (StringUtils.hasText(username)) {
builder.addPropertyValue("username", username);
}
String password = element.getAttribute("password");
if (StringUtils.hasText(password)) {
builder.addPropertyValue("password", password);
}
String deferredInitialization = element.getAttribute("deferredInitialization");
if (StringUtils.hasText(deferredInitialization)) {
builder.addPropertyValue("deferredInitialization", deferredInitialization);
}
String metricsEnabled = element.getAttribute("metricsEnabled");
if (StringUtils.hasText(metricsEnabled)) {
builder.addPropertyValue("metricsEnabled", metricsEnabled);
}
String jmxReportingEnabled = element.getAttribute("jmxReportingEnabled");
if (StringUtils.hasText(jmxReportingEnabled)) {
builder.addPropertyValue("jmxReportingEnabled", jmxReportingEnabled);
}
String loadBalancingPolicy = element.getAttribute("load-balancing-policy-ref");
if (StringUtils.hasText(loadBalancingPolicy)) {
builder.addPropertyReference("loadBalancingPolicy", loadBalancingPolicy);
}
String reconnectionPolicy = element.getAttribute("reconnection-policy-ref");
if (StringUtils.hasText(reconnectionPolicy)) {
builder.addPropertyReference("reconnectionPolicy", reconnectionPolicy);
}
String retryPolicy = element.getAttribute("retry-policy-ref");
if (StringUtils.hasText(retryPolicy)) {
builder.addPropertyReference("retryPolicy", retryPolicy);
}
parseChildElements(element, context, builder);
}
protected void parseChildElements(BeanDefinitionBuilder builder, Element element) {
/**
* Parse the Child Elemement of {@link BeanNames.CASSANDRA_CLUSTER}
*
* @param element The Element being parsed
* @param context The Parser Context
* @param builder The parent {@link BeanDefinitionBuilder}
*/
protected void parseChildElements(Element element, ParserContext context, BeanDefinitionBuilder builder) {
List<CreateKeyspaceSpecification> creates = new ArrayList<CreateKeyspaceSpecification>();
List<DropKeyspaceSpecification> drops = new ArrayList<DropKeyspaceSpecification>();
ManagedSet<BeanDefinition> keyspaceActionSpecificationBeanDefinitions = new ManagedSet<BeanDefinition>();
List<String> startupScripts = new ArrayList<String>();
List<String> shutdownScripts = new ArrayList<String>();
List<Element> elements = DomUtils.getChildElements(element);
BeanDefinition keyspaceActionSpecificationBeanDefinition = null;
// parse nested elements
/*
* PoolingOptionsBuilder has two potential parsing cycles so it is defined
* before the child elements are iterated over, then converted to a BeanDefinition
* just in time.
*/
BeanDefinitionBuilder poolingOptionsBuilder = null;
/*
* Parse each of the child elements
*/
for (Element subElement : elements) {
String name = subElement.getLocalName();
if ("local-pooling-options".equals(name)) {
builder.addPropertyValue("localPoolingOptions", parsePoolingOptions(subElement));
poolingOptionsBuilder = parsePoolingOptions(subElement, poolingOptionsBuilder, HostDistance.LOCAL);
} else if ("remote-pooling-options".equals(name)) {
builder.addPropertyValue("remotePoolingOptions", parsePoolingOptions(subElement));
poolingOptionsBuilder = parsePoolingOptions(subElement, poolingOptionsBuilder, HostDistance.REMOTE);
} else if ("socket-options".equals(name)) {
builder.addPropertyValue("socketOptions", parseSocketOptions(subElement));
builder.addPropertyValue("socketOptions", getSocketOptionsBeanDefinition(subElement, context));
} else if ("keyspace".equals(name)) {
KeyspaceSpecifications specifications = parseKeyspace(subElement);
keyspaceActionSpecificationBeanDefinition = getKeyspaceSpecificationBeanDefinition(subElement, context);
keyspaceActionSpecificationBeanDefinitions.add(keyspaceActionSpecificationBeanDefinition);
if (specifications.create != null) {
creates.add(specifications.create);
}
if (specifications.drop != null) {
drops.add(specifications.drop);
}
} else if ("startup-cql".equals(name)) {
startupScripts.add(parseScript(subElement));
} else if ("shutdown-cql".equals(name)) {
@@ -119,123 +206,169 @@ public class CassandraClusterParser extends AbstractSimpleBeanDefinitionParser {
}
}
builder.addPropertyValue("keyspaceCreations", creates);
builder.addPropertyValue("keyspaceDrops", drops);
/*
* If the PoolingOptionsBuilder was initilized during parsing, process it now.
*/
if (poolingOptionsBuilder != null) {
builder.addPropertyValue("poolingOptions", getSourceBeanDefinition(poolingOptionsBuilder, context, element));
}
builder.addPropertyValue("keyspaceSpecifications",
getKeyspaceSetFlattenerBeanDefinition(element, context, keyspaceActionSpecificationBeanDefinitions));
builder.addPropertyValue("startupScripts", startupScripts);
builder.addPropertyValue("shutdownScripts", startupScripts);
}
protected KeyspaceSpecifications parseKeyspace(Element element) {
/**
* Create the Single Factory Bean that will flatten all List<List<KeyspaceActionSpecificationFactoryBean>>
*
* @param element The Element being parsed
* @param context The Parser Context
* @param keyspaceActionSpecificationBeanDefinitions The List of Definitions to flatten
* @return A single level List of KeyspaceActionSpecifications
*/
private Object getKeyspaceSetFlattenerBeanDefinition(Element element, ParserContext context,
ManagedSet<BeanDefinition> keyspaceActionSpecificationBeanDefinitions) {
CreateKeyspaceSpecification create = null;
DropKeyspaceSpecification drop = null;
BeanDefinitionBuilder flat = BeanDefinitionBuilder.genericBeanDefinition(MultiLevelSetFlattenerFactoryBean.class);
flat.addPropertyValue("multiLevelSet", keyspaceActionSpecificationBeanDefinitions);
return getSourceBeanDefinition(flat, context, element);
String name = element.getAttribute("name");
if (name == null || name.trim().length() == 0) {
name = BeanNames.CASSANDRA_KEYSPACE;
}
boolean durableWrites = Boolean.valueOf(element.getAttribute("durable-writes"));
String action = element.getAttribute("action");
if (action == null || action.trim().length() == 0) {
throw new IllegalArgumentException("attribute action must be given");
}
if (action.startsWith("CREATE")) {
create = CreateKeyspaceSpecification.createKeyspace().name(name)
.with(KeyspaceOption.DURABLE_WRITES, durableWrites);
NodeList nodes = element.getElementsByTagName("replication");
create = parseReplication((Element) (nodes.getLength() == 1 ? nodes.item(0) : null), create);
}
if (action.equals("CREATE-DROP")) {
drop = DropKeyspaceSpecification.dropKeyspace().name(create.getName());
}
return new KeyspaceSpecifications(create, drop);
}
protected CreateKeyspaceSpecification parseReplication(Element element, CreateKeyspaceSpecification create) {
/**
* Parses the keyspace replication options and adds them to the supplied {@link BeanDefinitionBuilder}.
*
* @param element The Element being parsed
* @param builder The {@link BeanDefinitionBuilder} to add the replication to
*/
protected void parseReplication(Element element, BeanDefinitionBuilder builder) {
ManagedList<String> networkTopologyDataCenters = new ManagedList<String>();
ManagedList<String> networkTopologyReplicationFactors = new ManagedList<String>();
String strategyClass = null;
String replicationFactor = null;
if (element != null) {
strategyClass = element.getAttribute("class");
}
if (strategyClass == null || (strategyClass = strategyClass.trim()).length() == 0) {
strategyClass = KeyspaceAttributes.DEFAULT_REPLICATION_STRATEGY;
}
Long replicationFactor = null;
if (element != null) {
String s = element.getAttribute("replication-factor");
replicationFactor = (s == null || s.trim().length() == 0) ? null : Long.parseLong(s);
}
if (replicationFactor == null) {
replicationFactor = KeyspaceAttributes.DEFAULT_REPLICATION_FACTOR;
}
Map<Option, Object> replicationMap = new HashMap<Option, Object>();
replicationMap.put(new DefaultOption("class", String.class, false, false, true), strategyClass);
replicationMap.put(new DefaultOption("replication_factor", Long.class, true, false, false), replicationFactor);
if (element != null) {
NodeList dataCenters = element.getElementsByTagName("data-center");
int length = dataCenters.getLength();
for (int i = 0; i < length; i++) {
Element dataCenter = (Element) dataCenters.item(i);
replicationMap.put(new DefaultOption(dataCenter.getAttribute("name"), Long.class, false, false, true),
dataCenter.getAttribute("replication-factor"));
if (!StringUtils.hasText(strategyClass)) {
strategyClass = KeyspaceAttributes.DEFAULT_REPLICATION_STRATEGY;
}
replicationFactor = element.getAttribute("replication-factor");
if (!StringUtils.hasText(replicationFactor)) {
replicationFactor = KeyspaceAttributes.DEFAULT_REPLICATION_FACTOR + "";
}
/*
* DataCenters only apply to NetworkTolopogyStrategy
*/
List<Element> dcElements = DomUtils.getChildElementsByTagName(element, "data-center");
for (Element dataCenter : dcElements) {
networkTopologyDataCenters.add(dataCenter.getAttribute("name"));
networkTopologyReplicationFactors.add(dataCenter.getAttribute("replication-factor"));
}
} else {
strategyClass = ReplicationStrategy.SIMPLE_STRATEGY.name();
replicationFactor = KeyspaceAttributes.DEFAULT_REPLICATION_FACTOR + "";
}
return create.with(KeyspaceOption.REPLICATION, replicationMap);
builder.addPropertyValue("replicationStrategy", strategyClass);
builder.addPropertyValue("replicationFactor", replicationFactor);
builder.addPropertyValue("networkTopologyDataCenters", networkTopologyDataCenters);
builder.addPropertyValue("networkTopologyReplicationFactors", networkTopologyReplicationFactors);
}
/**
* Parse CQL Script Elements
*
* @param element The Element being parsed
* @return
*/
protected String parseScript(Element element) {
return element.getTextContent();
}
protected BeanDefinition parsePoolingOptions(Element element) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PoolingOptionsConfig.class);
/**
* Returns a {@link BeanDefinition} for a {@link PoolingOptions} object.
*
* @param element The Element being parsed
* @param builder The {@link BeanDefinition} to use for building if one already exists
* @param hostDistance The scope of the PoolingOptions to apply
* @return The {@link BeanDefinitionBuilder}
*/
protected BeanDefinitionBuilder parsePoolingOptions(Element element, BeanDefinitionBuilder builder,
HostDistance hostDistance) {
ParsingUtils.setPropertyValue(builder, element, "min-simultaneous-requests", "minSimultaneousRequests");
ParsingUtils.setPropertyValue(builder, element, "max-simultaneous-requests", "maxSimultaneousRequests");
ParsingUtils.setPropertyValue(builder, element, "core-connections", "coreConnections");
ParsingUtils.setPropertyValue(builder, element, "max-connections", "maxConnections");
if (builder == null) {
builder = BeanDefinitionBuilder.genericBeanDefinition(PoolingOptionsFactoryBean.class);
}
return builder.getBeanDefinition();
if (hostDistance.equals(HostDistance.LOCAL)) {
ParsingUtils.setPropertyValue(builder, element, "min-simultaneous-requests", "localMinSimultaneousRequests");
ParsingUtils.setPropertyValue(builder, element, "max-simultaneous-requests", "localMaxSimultaneousRequests");
ParsingUtils.setPropertyValue(builder, element, "core-connections", "localCoreConnections");
ParsingUtils.setPropertyValue(builder, element, "max-connections", "localMaxConnections");
}
if (hostDistance.equals(HostDistance.REMOTE)) {
ParsingUtils.setPropertyValue(builder, element, "min-simultaneous-requests", "remoteMinSimultaneousRequests");
ParsingUtils.setPropertyValue(builder, element, "max-simultaneous-requests", "remoteMaxSimultaneousRequests");
ParsingUtils.setPropertyValue(builder, element, "core-connections", "remoteCoreConnections");
ParsingUtils.setPropertyValue(builder, element, "max-connections", "remoteMaxConnections");
}
return builder;
}
protected BeanDefinition parseSocketOptions(Element element) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(SocketOptionsConfig.class);
/**
* Returns a {@link BeanDefinition} for a {@link SocketOptions} object.
*
* @param element The Element being parsed
* @param context The ParserContext
* @return The {@link BeanDefinition}
*/
protected BeanDefinition getSocketOptionsBeanDefinition(Element element, ParserContext context) {
ParsingUtils.setPropertyValue(builder, element, "connect-timeout-mls", "connectTimeoutMls");
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(SocketOptionsFactoryBean.class);
ParsingUtils.setPropertyValue(builder, element, "connect-timeout-mls", "connectTimeoutMillis");
ParsingUtils.setPropertyValue(builder, element, "keep-alive", "keepAlive");
ParsingUtils.setPropertyValue(builder, element, "read-timeout-mls", "readTimeoutMillis");
ParsingUtils.setPropertyValue(builder, element, "reuse-address", "reuseAddress");
ParsingUtils.setPropertyValue(builder, element, "so-linger", "soLinger");
ParsingUtils.setPropertyValue(builder, element, "tcp-no-delay", "tcpNoDelay");
ParsingUtils.setPropertyValue(builder, element, "receive-buffer-size", "receiveBufferSize");
ParsingUtils.setPropertyValue(builder, element, "send-buffer-size", "sendBufferSize");
return builder.getBeanDefinition();
return getSourceBeanDefinition(builder, context, element);
}
protected static class KeyspaceSpecifications {
/**
* Returns a {@link BeanDefinition} for a {@link KeyspaceActionSpecification} object.
*
* @param element The Element being parsed
* @param context The Parser Context
* @return The {@link BeanDefinition} or {@literal null} if action is not given.
*/
private BeanDefinition getKeyspaceSpecificationBeanDefinition(Element element, ParserContext context) {
public KeyspaceSpecifications(CreateKeyspaceSpecification create, DropKeyspaceSpecification drop) {
this.create = create;
this.drop = drop;
}
String action = element.getAttribute("action");
public CreateKeyspaceSpecification create;
public DropKeyspaceSpecification drop;
// TODO: public AlterKeyspaceSpecification alter;
Assert.notNull(action, "Keyspace Action must not be null!");
BeanDefinitionBuilder keyspaceBuilder = BeanDefinitionBuilder
.genericBeanDefinition(KeyspaceActionSpecificationFactoryBean.class);
ParsingUtils.setPropertyValue(keyspaceBuilder, element, "name", "name");
ParsingUtils.setPropertyValue(keyspaceBuilder, element, "action", "action");
ParsingUtils.setPropertyValue(keyspaceBuilder, element, "durableWrites", "durableWrites");
Element replicationElement = DomUtils.getChildElementByTagName(element, "replication");
parseReplication(replicationElement, keyspaceBuilder);
return getSourceBeanDefinition(keyspaceBuilder, context, element);
}
}

10
spring-cassandra/src/main/java/org/springframework/cassandra/core/cql/generator/KeyspaceNameCqlGenerator.java
View File

@@ -15,20 +15,20 @@
*/
package org.springframework.cassandra.core.cql.generator;
import org.springframework.cassandra.core.keyspace.KeyspaceNameSpecification;
import org.springframework.cassandra.core.keyspace.KeyspaceActionSpecification;
import org.springframework.util.Assert;
public abstract class KeyspaceNameCqlGenerator<T extends KeyspaceNameSpecification<T>> {
public abstract class KeyspaceNameCqlGenerator<T extends KeyspaceActionSpecification<T>> {
public abstract StringBuilder toCql(StringBuilder cql);
private KeyspaceNameSpecification<T> specification;
private KeyspaceActionSpecification<T> specification;
public KeyspaceNameCqlGenerator(KeyspaceNameSpecification<T> specification) {
public KeyspaceNameCqlGenerator(KeyspaceActionSpecification<T> specification) {
setSpecification(specification);
}
protected void setSpecification(KeyspaceNameSpecification<T> specification) {
protected void setSpecification(KeyspaceActionSpecification<T> specification) {
Assert.notNull(specification);
this.specification = specification;
}

56
spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/DropKeyspaceSpecification.java
View File

@@ -1,28 +1,28 @@
package org.springframework.cassandra.core.keyspace;
public class DropKeyspaceSpecification extends KeyspaceNameSpecification<DropKeyspaceSpecification> {
private boolean ifExists;
public DropKeyspaceSpecification ifExists() {
return ifExists(true);
}
public DropKeyspaceSpecification ifExists(boolean ifExists) {
this.ifExists = ifExists;
return this;
}
public boolean getIfExists() {
return ifExists;
}
/**
* Entry point into the {@link DropKeyspaceSpecification}'s fluent API to drop a keyspace. Convenient if imported
* statically.
*/
public static DropKeyspaceSpecification dropKeyspace() {
return new DropKeyspaceSpecification();
}
}
package org.springframework.cassandra.core.keyspace;
public class DropKeyspaceSpecification extends KeyspaceActionSpecification<DropKeyspaceSpecification> {
private boolean ifExists;
public DropKeyspaceSpecification ifExists() {
return ifExists(true);
}
public DropKeyspaceSpecification ifExists(boolean ifExists) {
this.ifExists = ifExists;
return this;
}
public boolean getIfExists() {
return ifExists;
}
/**
* Entry point into the {@link DropKeyspaceSpecification}'s fluent API to drop a keyspace. Convenient if imported
* statically.
*/
public static DropKeyspaceSpecification dropKeyspace() {
return new DropKeyspaceSpecification();
}
}

76
spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/KeyspaceActionSpecification.java Normal file
View File

@@ -0,0 +1,76 @@
package org.springframework.cassandra.core.keyspace;
import static org.springframework.cassandra.core.cql.CqlStringUtils.checkIdentifier;
import static org.springframework.cassandra.core.cql.CqlStringUtils.identifize;
/**
* Abstract builder class to support the construction of keyspace specifications.
*
* @author John McPeek
* @author David Webb
* @param <T> The subtype of the {@link KeyspaceActionSpecification}
*/
public abstract class KeyspaceActionSpecification<T extends KeyspaceActionSpecification<T>> {
/**
* The name of the table.
*/
private String name;
/**
* Sets the keyspace name.
*
* @return this
*/
@SuppressWarnings("unchecked")
public T name(String name) {
checkIdentifier(name);
this.name = name;
return (T) this;
}
public String getName() {
return name;
}
public String getNameAsIdentifier() {
return identifize(name);
}
/**
* For debugging KeyspaceActionSprcifications
*/
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append("Keyspace Action Specification {name: " + name + ", class: " + this.getClass() + "}");
return sb.toString();
}
/**
* Determine the KeyspaceActionSpecifications are the same if they have the same "name" and same class.
*
* @param that The object to compare this to.
* @return Are this and that the same?
*/
@Override
public boolean equals(Object that) {
if (this == that) {
return true;
}
if (that == null) {
return false;
}
if (!(that instanceof KeyspaceActionSpecification)) {
return false;
}
KeyspaceActionSpecification<?> thatSpec = (KeyspaceActionSpecification<?>) that;
return this.name.equals(thatSpec.name) && this.getClass().equals(that.getClass());
}
@Override
public int hashCode() {
return this.name.hashCode() ^ this.getClass().hashCode();
}
}

39
spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/KeyspaceNameSpecification.java
View File

@@ -1,39 +0,0 @@
package org.springframework.cassandra.core.keyspace;
import static org.springframework.cassandra.core.cql.CqlStringUtils.checkIdentifier;
import static org.springframework.cassandra.core.cql.CqlStringUtils.identifize;
/**
* Abstract builder class to support the construction of keyspace specifications.
*
* @author John McPeek
* @param <T> The subtype of the {@link KeyspaceNameSpecification}
*/
public abstract class KeyspaceNameSpecification<T extends KeyspaceNameSpecification<T>> {
/**
* The name of the table.
*/
private String name;
/**
* Sets the keyspace name.
*
* @return this
*/
@SuppressWarnings( "unchecked" )
public T name(String name) {
checkIdentifier(name);
this.name = name;
return (T) this;
}
public String getName() {
return name;
}
public String getNameAsIdentifier() {
return identifize(name);
}
}

158
spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/KeyspaceOption.java
View File

@@ -1,79 +1,79 @@
package org.springframework.cassandra.core.keyspace;
import java.util.Map;
public enum KeyspaceOption implements Option {
REPLICATION("replication", Map.class, true, false, false),
DURABLE_WRITES("durable_writes", Boolean.class, false, false, false);
private Option delegate;
private KeyspaceOption(String name, Class<?> type, boolean requiresValue, boolean escapesValue, boolean quotesValue) {
this.delegate = new DefaultOption(name, type, requiresValue, escapesValue, quotesValue);
}
public Class<?> getType() {
return delegate.getType();
}
public boolean takesValue() {
return delegate.takesValue();
}
public String getName() {
return delegate.getName();
}
public boolean escapesValue() {
return delegate.escapesValue();
}
public boolean quotesValue() {
return delegate.quotesValue();
}
public boolean requiresValue() {
return delegate.requiresValue();
}
public void checkValue(Object value) {
delegate.checkValue(value);
}
public boolean isCoerceable(Object value) {
return delegate.isCoerceable(value);
}
public String toString() {
return delegate.toString();
}
public String toString(Object value) {
return delegate.toString(value);
}
/**
* Known Replication Strategy options.
*
* @author John McPeek
*
*/
public enum ReplicationStrategy {
SIMPLE_STRATEGY("SimpleStrategy"), NETWORK_TOPOLOGY_STRATEGY("NetworkTopologyStrategy");
private String value;
private ReplicationStrategy(String value) {
this.value = value;
}
public String getValue() {
return value;
}
public String toString() {
return getValue();
}
}
}
package org.springframework.cassandra.core.keyspace;
import java.util.Map;
public enum KeyspaceOption implements Option {
REPLICATION("replication", Map.class, true, false, false),
DURABLE_WRITES("durable_writes", Boolean.class, false, false, false);
private Option delegate;
private KeyspaceOption(String name, Class<?> type, boolean requiresValue, boolean escapesValue, boolean quotesValue) {
this.delegate = new DefaultOption(name, type, requiresValue, escapesValue, quotesValue);
}
public Class<?> getType() {
return delegate.getType();
}
public boolean takesValue() {
return delegate.takesValue();
}
public String getName() {
return delegate.getName();
}
public boolean escapesValue() {
return delegate.escapesValue();
}
public boolean quotesValue() {
return delegate.quotesValue();
}
public boolean requiresValue() {
return delegate.requiresValue();
}
public void checkValue(Object value) {
delegate.checkValue(value);
}
public boolean isCoerceable(Object value) {
return delegate.isCoerceable(value);
}
public String toString() {
return delegate.toString();
}
public String toString(Object value) {
return delegate.toString(value);
}
/**
* Known Replication Strategy options.
*
* @author John McPeek
*
*/
public enum ReplicationStrategy {
SIMPLE_STRATEGY("SimpleStrategy"), NETWORK_TOPOLOGY_STRATEGY("NetworkTopologyStrategy");
private String value;
private ReplicationStrategy(String value) {
this.value = value;
}
public String getValue() {
return value;
}
public String toString() {
return getValue();
}
}
}

188
spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/KeyspaceOptionsSpecification.java
View File

@@ -1,94 +1,94 @@
package org.springframework.cassandra.core.keyspace;
import static org.springframework.cassandra.core.cql.CqlStringUtils.escapeSingle;
import static org.springframework.cassandra.core.cql.CqlStringUtils.singleQuote;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
import org.springframework.cassandra.core.cql.CqlStringUtils;
/**
* Abstract builder class to support the construction of table specifications that have table options, that is, those
* options normally specified by <code>WITH ... AND ...</code>.
* <p/>
* It is important to note that although this class depends on {@link KeyspaceOption} for convenient and typesafe use, it
* ultimately stores its options in a <code>Map<String,Object></code> for flexibility. This means that
* {@link #with(KeyspaceOption)} and {@link #with(KeyspaceOption, Object)} delegate to
* {@link #with(String, Object, boolean, boolean)}. This design allows the API to support new Cassandra options as they
* are introduced without having to update the code immediately.
*
* @author John McPeek
* @param <T> The subtype of the {@link KeyspaceOptionsSpecification}.
*/
public abstract class KeyspaceOptionsSpecification<T extends KeyspaceOptionsSpecification<T>> extends
KeyspaceNameSpecification<KeyspaceOptionsSpecification<T>> {
protected Map<String, Object> options = new LinkedHashMap<String, Object>();
@SuppressWarnings( "unchecked" )
public T name(String name) {
return (T) super.name(name);
}
/**
* Convenience method that calls <code>with(option, null)</code>.
*
* @return this
*/
public T with(KeyspaceOption option) {
return with(option, null);
}
/**
* Sets the given table option. This is a convenience method that calls
* {@link #with(String, Object, boolean, boolean)} appropriately from the given {@link KeyspaceOption} and value for that
* option.
*
* @param option The option to set.
* @param value The value of the option. Must be type-compatible with the {@link KeyspaceOption}.
* @return this
* @see #with(String, Object, boolean, boolean)
*/
public T with(KeyspaceOption option, Object value) {
option.checkValue(value);
return (T) with(option.getName(), value, option.escapesValue(), option.quotesValue());
}
/**
* Adds the given option by name to this keyspaces's options.
* <p/>
* Options that have <code>null</code> values are considered single string options where the name of the option is the
* string to be used. Otherwise, the result of {@link Object#toString()} is considered to be the value of the option
* with the given name. The value, after conversion to string, may have embedded single quotes escaped according to
* parameter <code>escape</code> and may be single-quoted according to parameter <code>quote</code>.
*
* @param name The name of the option
* @param value The value of the option. If <code>null</code>, the value is ignored and the option is considered to be
* composed of only the name, otherwise the value's {@link Object#toString()} value is used.
* @param escape Whether to escape the value via {@link CqlStringUtils#escapeSingle(Object)}. Ignored if given value
* is an instance of a {@link Map}.
* @param quote Whether to quote the value via {@link CqlStringUtils#singleQuote(Object)}. Ignored if given value is
* an instance of a {@link Map}.
* @return this
*/
@SuppressWarnings("unchecked")
public T with(String name, Object value, boolean escape, boolean quote) {
if (!(value instanceof Map)) {
if (escape) {
value = escapeSingle(value);
}
if (quote) {
value = singleQuote(value);
}
}
options.put(name, value);
return (T) this;
}
public Map<String, Object> getOptions() {
return Collections.unmodifiableMap(options);
}
}
package org.springframework.cassandra.core.keyspace;
import static org.springframework.cassandra.core.cql.CqlStringUtils.escapeSingle;
import static org.springframework.cassandra.core.cql.CqlStringUtils.singleQuote;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
import org.springframework.cassandra.core.cql.CqlStringUtils;
/**
* Abstract builder class to support the construction of table specifications that have table options, that is, those
* options normally specified by <code>WITH ... AND ...</code>.
* <p/>
* It is important to note that although this class depends on {@link KeyspaceOption} for convenient and typesafe use, it
* ultimately stores its options in a <code>Map<String,Object></code> for flexibility. This means that
* {@link #with(KeyspaceOption)} and {@link #with(KeyspaceOption, Object)} delegate to
* {@link #with(String, Object, boolean, boolean)}. This design allows the API to support new Cassandra options as they
* are introduced without having to update the code immediately.
*
* @author John McPeek
* @param <T> The subtype of the {@link KeyspaceOptionsSpecification}.
*/
public abstract class KeyspaceOptionsSpecification<T extends KeyspaceOptionsSpecification<T>> extends
KeyspaceActionSpecification<KeyspaceOptionsSpecification<T>> {
protected Map<String, Object> options = new LinkedHashMap<String, Object>();
@SuppressWarnings( "unchecked" )
public T name(String name) {
return (T) super.name(name);
}
/**
* Convenience method that calls <code>with(option, null)</code>.
*
* @return this
*/
public T with(KeyspaceOption option) {
return with(option, null);
}
/**
* Sets the given table option. This is a convenience method that calls
* {@link #with(String, Object, boolean, boolean)} appropriately from the given {@link KeyspaceOption} and value for that
* option.
*
* @param option The option to set.
* @param value The value of the option. Must be type-compatible with the {@link KeyspaceOption}.
* @return this
* @see #with(String, Object, boolean, boolean)
*/
public T with(KeyspaceOption option, Object value) {
option.checkValue(value);
return (T) with(option.getName(), value, option.escapesValue(), option.quotesValue());
}
/**
* Adds the given option by name to this keyspaces's options.
* <p/>
* Options that have <code>null</code> values are considered single string options where the name of the option is the
* string to be used. Otherwise, the result of {@link Object#toString()} is considered to be the value of the option
* with the given name. The value, after conversion to string, may have embedded single quotes escaped according to
* parameter <code>escape</code> and may be single-quoted according to parameter <code>quote</code>.
*
* @param name The name of the option
* @param value The value of the option. If <code>null</code>, the value is ignored and the option is considered to be
* composed of only the name, otherwise the value's {@link Object#toString()} value is used.
* @param escape Whether to escape the value via {@link CqlStringUtils#escapeSingle(Object)}. Ignored if given value
* is an instance of a {@link Map}.
* @param quote Whether to quote the value via {@link CqlStringUtils#singleQuote(Object)}. Ignored if given value is
* an instance of a {@link Map}.
* @return this
*/
@SuppressWarnings("unchecked")
public T with(String name, Object value, boolean escape, boolean quote) {
if (!(value instanceof Map)) {
if (escape) {
value = escapeSingle(value);
}
if (quote) {
value = singleQuote(value);
}
}
options.put(name, value);
return (T) this;
}
public Map<String, Object> getOptions() {
return Collections.unmodifiableMap(options);
}
}

119
spring-cassandra/src/main/resources/org/springframework/cassandra/config/spring-cassandra-1.0.xsd
View File

@@ -59,22 +59,16 @@ Defines a Cassandra cluster.
<xsd:complexType name="clusterType">
<xsd:sequence>
<xsd:element name="local-pooling-options" type="poolingOptionsType"
maxOccurs="1" minOccurs="0">
<xsd:annotation>
<xsd:documentation><![CDATA[
Local pooling options.
]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="remote-pooling-options" type="poolingOptionsType"
maxOccurs="1" minOccurs="0">
<xsd:element name="local-pooling-options"
type="poolingOptionsType" maxOccurs="1" minOccurs="0">
<xsd:annotation>
<xsd:documentation><![CDATA[
Remote pooling options.
]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="remote-pooling-options" type="poolingOptionsType" minOccurs="0" maxOccurs="1"></xsd:element>
<xsd:element name="socket-options" type="socketOptionsType"
maxOccurs="1" minOccurs="0">
<xsd:annotation>
@@ -133,32 +127,16 @@ The native CQL port to connect to. Default is 9042.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="compression" default="NONE" use="optional">
<xsd:attribute name="compression" default="NONE" use="optional"
type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
The protocol compression option. Default is "NONE".
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:restriction base="xsd:string">
<xsd:enumeration value="NONE">
<xsd:annotation>
<xsd:documentation><![CDATA[
No compression.
]]></xsd:documentation>
</xsd:annotation>
</xsd:enumeration>
<xsd:enumeration value="SNAPPY">
<xsd:annotation>
<xsd:documentation><![CDATA[
SNAPPY compression algorithm.
]]></xsd:documentation>
</xsd:annotation>
</xsd:enumeration>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="auth-info-provider" use="optional">
<xsd:attribute name="auth-info-provider-ref" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
AuthInfoProvider implementation.
@@ -175,7 +153,43 @@ AuthInfoProvider implementation.
<xsd:union memberTypes="xsd:string" />
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="load-balancing-policy" use="optional">
<xsd:attribute name="username" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
When Authentication is enabled, the username to use when connecting to the Cluster.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="password" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
When Authentication is enabled, the password to use when connecting to the Cluster.
]]>
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="metricsEnabled" type="xsd:string" default="true">
<xsd:annotation>
<xsd:documentation><![CDATA[
Determine whether or not to collect metrics. Defaults to true.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="jmxReportingEnabled" type="xsd:string" default="true">
<xsd:annotation>
<xsd:documentation><![CDATA[
Determine whether or not to enable JMX Reporting. Defaults to true.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="deferredInitialization" type="xsd:string" default="false">
<xsd:annotation>
<xsd:documentation><![CDATA[
Determine if we defer initalizing the cluster until a connection is requested. Defaults to false.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="load-balancing-policy-ref" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
LoadBalancingPolicy implementation.
@@ -193,7 +207,7 @@ LoadBalancingPolicy implementation.
<xsd:union memberTypes="xsd:string" />
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="reconnection-policy" use="optional">
<xsd:attribute name="reconnection-policy-ref" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
ReconnectionPolicy implementation.
@@ -211,7 +225,7 @@ ReconnectionPolicy implementation.
<xsd:union memberTypes="xsd:string" />
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="retry-policy" use="optional">
<xsd:attribute name="retry-policy-ref" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
RetryPolicy implementation.
@@ -230,7 +244,6 @@ RetryPolicy implementation.
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
<xsd:simpleType name="clusterRef" final="union">
<xsd:annotation>
<xsd:appinfo>
@@ -285,7 +298,7 @@ More connections are created up to a configurable maximum number of connections.
</xsd:complexType>
<xsd:complexType name="socketOptionsType">
<xsd:attribute name="connect-timeout-mls" type="xsd:string">
<xsd:attribute name="connect-timeout-millis" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Sets connection timeout for client socket in milliseconds.
@@ -299,6 +312,13 @@ Sets the SO_KEEPALIVE socket option.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="read-timeout-millis" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Sets read timeout for client socket in milliseconds.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reuse-address" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -357,7 +377,7 @@ Arbitrary CQL script to be executed against the session's keyspace during bean d
</xsd:annotation>
</xsd:element>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:ID" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -422,32 +442,14 @@ The name of this keyspace. Required.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="action" use="required">
<xsd:attribute name="action" use="required" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
The keyspace action to take at startup and possibly shutdown.
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:restriction base="xsd:string">
<xsd:enumeration value="CREATE">
<xsd:annotation>
<xsd:documentation><![CDATA[
Action value that causes keyspace creation during bean initialization.
]]></xsd:documentation>
</xsd:annotation>
</xsd:enumeration>
<xsd:enumeration value="CREATE-DROP">
<xsd:annotation>
<xsd:documentation><![CDATA[
Action value that causes keyspace creation during bean initialization and keyspace dropping during bean destruction.
]]></xsd:documentation>
</xsd:annotation>
</xsd:enumeration>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="durable-writes" type="xsd:boolean"
<xsd:attribute name="durable-writes" type="xsd:string"
use="optional" default="false">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -481,7 +483,7 @@ The name of the replication class; default is "SimpleStrategy".
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="replication-factor" type="xsd:integer"
<xsd:attribute name="replication-factor" type="xsd:string"
use="optional" default="1">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -490,7 +492,6 @@ The replication factor; default is 1.
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:complexType name="datacenterType">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -504,7 +505,7 @@ The name of the data center.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="replication-factor" type="xsd:integer"
<xsd:attribute name="replication-factor" type="xsd:string"
use="required">
<xsd:annotation>
<xsd:documentation><![CDATA[

43
spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/config/xml/PropertyPlaceholderNamespaceCreatingXmlConfigTest.java Normal file
View File

@@ -0,0 +1,43 @@
package org.springframework.cassandra.test.integration.config.xml;
import static org.junit.Assert.assertNotNull;
import javax.inject.Inject;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cassandra.core.CassandraOperations;
import org.springframework.cassandra.test.integration.AbstractEmbeddedCassandraIntegrationTest;
import org.springframework.cassandra.test.integration.config.IntegrationTestUtils;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.datastax.driver.core.Session;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class PropertyPlaceholderNamespaceCreatingXmlConfigTest extends AbstractEmbeddedCassandraIntegrationTest {
private static Logger log = LoggerFactory.getLogger(PropertyPlaceholderNamespaceCreatingXmlConfigTest.class);
@Override
protected String keyspace() {
return "ppncxcx";
}
@Inject
Session s;
@Inject
CassandraOperations ops;
@Test
public void test() {
IntegrationTestUtils.assertSession(s);
IntegrationTestUtils.assertKeyspaceExists(keyspace(), s);
assertNotNull(ops);
}
}

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

@@ -3,7 +3,7 @@ package org.springframework.cassandra.test.unit.core.cql.generator;
import org.junit.Test;
import org.springframework.cassandra.core.cql.generator.KeyspaceNameCqlGenerator;
import org.springframework.cassandra.core.cql.generator.TableNameCqlGenerator;
import org.springframework.cassandra.core.keyspace.KeyspaceNameSpecification;
import org.springframework.cassandra.core.keyspace.KeyspaceActionSpecification;
import org.springframework.cassandra.core.keyspace.TableNameSpecification;
/**
@@ -16,7 +16,7 @@ import org.springframework.cassandra.core.keyspace.TableNameSpecification;
* @param <S> The type of the {@link TableNameSpecification}
* @param <G> The type of the {@link TableNameCqlGenerator}
*/
public abstract class KeyspaceOperationCqlGeneratorTest<S extends KeyspaceNameSpecification<?>, G extends KeyspaceNameCqlGenerator<?>> {
public abstract class KeyspaceOperationCqlGeneratorTest<S extends KeyspaceActionSpecification<?>, G extends KeyspaceNameCqlGenerator<?>> {
public abstract S specification();

4
spring-cassandra/src/test/resources/logback-test.xml
View File

@@ -10,8 +10,8 @@
</appender>
<logger name="org.springframework.context" level="info" />
<logger name="org.springframework.cassandra" level="info" />
<logger name="org.springframework.data.cassandra" level="info" />
<logger name="org.springframework.cassandra" level="debug" />
<logger name="org.springframework.data.cassandra" level="debug" />
<logger name="com.datastax" level="info" />
<root level="warn">

10
spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/FullySpecifiedKeyspaceCreatingXmlConfigTest-context.xml
View File

@@ -11,18 +11,16 @@
location="classpath:/org/springframework/cassandra/test/integration/config/xml/FullySpecifiedKeyspaceCreatingXmlConfigTest.properties" />
<cass:cluster>
<cass:keyspace action="CREATE-DROP" durable-writes="true"
<cass:keyspace action="CREATE_DROP" durable-writes="true"
name="full1">
<cass:replication class="SimpleStrategy"
replication-factor="1">
<cass:replication class="SIMPLE_STRATEGY">
<cass:data-center name="foo" replication-factor="1" />
<cass:data-center name="bar" replication-factor="2" />
</cass:replication>
</cass:keyspace>
<cass:keyspace action="CREATE-DROP" durable-writes="true"
<cass:keyspace action="CREATE_DROP" durable-writes="true"
name="full2">
<cass:replication class="SimpleStrategy"
replication-factor="1">
<cass:replication class="SIMPLE_STRATEGY">
<cass:data-center name="foo" replication-factor="1" />
<cass:data-center name="bar" replication-factor="2" />
</cass:replication>

2
spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/MinimalKeyspaceCreatingXmlConfigTest-context.xml
View File

@@ -8,7 +8,7 @@
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd">
<cass:cluster>
<cass:keyspace action="CREATE-DROP" name="minimal" />
<cass:keyspace action="CREATE_DROP" name="minimal" />
</cass:cluster>
<cass:session keyspace-name="minimal" />

72
spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/PropertyPlaceholderNamespaceCreatingXmlConfigTest-context.xml Normal file
View File

@@ -0,0 +1,72 @@
<?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:cassandra="http://www.springframework.org/schema/cassandra"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/cassandra http://www.springframework.org/schema/cassandra/spring-cassandra-1.0.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd">
<context:property-placeholder
location="classpath:/org/springframework/cassandra/test/integration/config/xml/ppncxct.properties" />
<bean id="authProvider" class="com.datastax.driver.core.sasl.DseAuthProvider" />
<bean id="loadBalancingPolicy" class="com.datastax.driver.core.policies.DCAwareRoundRobinPolicy">
<constructor-arg name="localDc" value="${lb.policy.dcAware.localDc}" />
<constructor-arg name="usedHostsPerRemoteDc" value="${lb.policy.dcAware.remoteHosts}" />
</bean>
<bean id="reconnectionPolicy" class="com.datastax.driver.core.policies.ConstantReconnectionPolicy">
<constructor-arg name="constantDelayMs" value="${cluster.reconnection.delayMillis}"/>
</bean>
<bean id="retryPolicy" class="com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy" />
<cassandra:cluster id="cassandra-cluster"
contactPoints="${cluster.contactPoints}" port="${cluster.port}"
compression="${cluster.compression}" auth-info-provider-ref="authProvider"
load-balancing-policy-ref="loadBalancingPolicy" username="${auth.username}" password="${auth.password}"
deferredInitialization="${cluster.deferredInit}" metricsEnabled="${cluster.metricsEnabled}"
jmxReportingEnabled="${cluster.jmxReportingEnabled}"
reconnection-policy-ref="reconnectionPolicy"
retry-policy-ref="retryPolicy">
<cassandra:local-pooling-options
min-simultaneous-requests="${local.min.requests}"
max-simultaneous-requests="${local.max.requests}"
core-connections="${local.core.connections}"
max-connections="${local.max.connections}"
/>
<cassandra:remote-pooling-options
min-simultaneous-requests="${remote.min.requests}"
max-simultaneous-requests="${remote.max.requests}"
core-connections="${remote.core.connections}"
max-connections="${remote.max.connections}"
/>
<cassandra:socket-options
connect-timeout-millis="${socket.connectTimeoutMillis}"
keep-alive="${socket.keepAlive}"
read-timeout-millis="${socket.readTimeoutMillis}"
reuse-address="${socket.reuseAddress}"
so-linger="${socket.soLinger}"
tcp-no-delay="${socket.tcpNoDelay}"
receive-buffer-size="${socket.receiveBufferSize}"
send-buffer-size="${socket.sendBufferSize}"
/>
<cassandra:keyspace name="${keyspace.name}" action="${keyspace.action}"/>
<cassandra:keyspace name="Foo" action="CREATE_DROP" durable-writes="true">
<cassandra:replication class="NETWORK_TOPOLOGY_STRATEGY">
<cassandra:data-center replication-factor="${dc1.rf}" name="${dc1.name}"/>
<cassandra:data-center replication-factor="${dc1.rf}" name="${dc2.name}"/>
</cassandra:replication>
</cassandra:keyspace>
</cassandra:cluster>
<cassandra:session id="cassandra-session"
keyspace-name="system">
</cassandra:session>
<bean id="cassandraTemplate" class="org.springframework.cassandra.core.CassandraTemplate">
<constructor-arg ref="cassandra-session" />
</bean>
</beans>

2
spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/XmlConfigTest-context.xml
View File

@@ -15,7 +15,7 @@
min-simultaneous-requests="25" max-simultaneous-requests="100"
core-connections="1" max-connections="2" />
<cassandra:socket-options
connect-timeout-mls="5000" keep-alive="true" reuse-address="true"
connect-timeout-millis="5000" keep-alive="true" read-timeout-millis="60000" reuse-address="true"
so-linger="60" tcp-no-delay="true" receive-buffer-size="65536"
send-buffer-size="65536" />
</cassandra:cluster>

33
spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/ppncxct.properties Normal file
View File

@@ -0,0 +1,33 @@
cluster.contactPoints=localhost
cluster.port=9042
cluster.compression=SNAPPY
cluster.deferredInit=true
cluster.metricsEnabled=false
cluster.jmxReportingEnabled=false
cluster.reconnection.delayMillis=5000
keyspace.name=ppncxct
keyspace.action=CREATE
dc1.name=DCJAX
dc1.rf=2
dc2.name=DCCTL
dc2.rf=3
lb.policy.dcAware.remoteHosts=4
lb.policy.dcAware.localDc=DCJAX
auth.username=test
auth.password=pass
socket.connectTimeoutMillis=5000
socket.keepAlive=true
socket.readTimeoutMillis=60000
socket.receiveBufferSize=1024
socket.sendBufferSize=2048
socket.reuseAddress=true
socket.soLinger=5
socket.tcpNoDelay=false
local.min.requests=10
local.max.requests=20
local.core.connections=30
local.max.connections=40
remote.min.requests=5
remote.max.requests=10
remote.core.connections=15
remote.max.connections=20