DATACASS-71 - XML parsing fails on property placeholders

Added BeanFeinitions in Parser to PPH works.

spring-cassandra/pom.xml

@@ -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>

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

@@ -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;
@@ -51,6 +53,7 @@ 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 {
@@ -62,9 +65,10 @@ public class CassandraClusterFactoryBean implements FactoryBean<Cluster>, Initia
 	protected static final Logger log = LoggerFactory.getLogger(CassandraClusterFactoryBean.class);
 
 	private Cluster cluster;
+	private boolean accumulating = true;
 
-	/**
-	 * Comma-delimited string of servers.
+	/*
+	 * Attributes needed for cluster builder
 	 */
 	private String contactPoints = DEFAULT_CONTACT_POINTS;
 	private int port = CassandraClusterFactoryBean.DEFAULT_PORT;
@@ -77,6 +81,7 @@ public class CassandraClusterFactoryBean implements FactoryBean<Cluster>, Initia
 	private ReconnectionPolicy reconnectionPolicy;
 	private RetryPolicy retryPolicy;
 	private boolean metricsEnabled = DEFAULT_METRICS_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>();
@@ -152,9 +157,31 @@ public class CassandraClusterFactoryBean implements FactoryBean<Cluster>, Initia
 		}
 
 		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 +194,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();
@@ -343,4 +370,35 @@ public class CassandraClusterFactoryBean implements FactoryBean<Cluster>, Initia
 
 		return socketOptions;
 	}
+
+	/**
+	 * @return Returns the keyspaceSpecifications.
+	 */
+	public Set<KeyspaceActionSpecification<?>> getKeyspaceSpecifications() {
+		return keyspaceSpecifications;
+	}
+
+	/**
+	 * 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;
+	}
+
+	/**
+	 * @return Returns the accumulating.
+	 */
+	public boolean isAccumulating() {
+		return accumulating;
+	}
+
+	/**
+	 * @param accumulating The accumulating to set.
+	 */
+	public void setAccumulating(boolean accumulating) {
+		this.accumulating = accumulating;
+	}
 }

spring-cassandra/src/main/java/org/springframework/cassandra/config/KeyspaceAction.java

@@ -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;
+}

spring-cassandra/src/main/java/org/springframework/cassandra/config/KeyspaceActionSpecificationFactoryBean.java

@@ -0,0 +1,185 @@
+/*
+ * 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.LinkedHashMap;
+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.Option;
+import org.springframework.util.Assert;
+
+/**
+ * @author David Webb (dwebb@brightmove.com)
+ * 
+ */
+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 Map<Option, Object> replicationOptions = new LinkedHashMap<Option, Object>();
+	private boolean durableWrites = false;
+	private boolean ifNotExists = false;
+
+	private Set<KeyspaceActionSpecification<?>> specs = new HashSet<KeyspaceActionSpecification<?>>();
+
+	@Override
+	public void destroy() throws Exception {
+		name = null;
+		replicationOptions = 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:
+			specs.add(generateCreateKeyspaceSpecification());
+			break;
+		case ALTER:
+			break;
+		}
+
+	}
+
+	private CreateKeyspaceSpecification generateCreateKeyspaceSpecification() {
+		CreateKeyspaceSpecification create = new CreateKeyspaceSpecification();
+		create.name(name).ifNotExists(ifNotExists).with(KeyspaceOption.DURABLE_WRITES, durableWrites);
+		if (replicationOptions != null && replicationOptions.size() > 0) {
+			create.with(KeyspaceOption.REPLICATION, replicationOptions);
+		} else {
+			Map<Option, Object> defaultReplicationStrategyMap = new HashMap<Option, Object>();
+			defaultReplicationStrategyMap.put(new DefaultOption("class", String.class, true, false, true),
+					KeyspaceOption.ReplicationStrategy.SIMPLE_STRATEGY);
+			defaultReplicationStrategyMap.put(new DefaultOption("replication_factor", String.class, true, false, false), "1");
+			create.with(KeyspaceOption.REPLICATION, defaultReplicationStrategyMap);
+		}
+		return create;
+	}
+
+	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 replicationOptions.
+	 */
+	public Map<Option, Object> getReplicationOptions() {
+		return replicationOptions;
+	}
+
+	/**
+	 * @param replicationOptions The replicationOptions to set.
+	 */
+	public void setReplicationOptions(Map<Option, Object> replicationOptions) {
+		this.replicationOptions = replicationOptions;
+	}
+
+	/**
+	 * @return Returns the durableWrites.
+	 */
+	public boolean isDurableWrites() {
+		return durableWrites;
+	}
+
+	/**
+	 * @param durableWrites The durableWrites to set.
+	 */
+	public void setDurableWrites(boolean durableWrites) {
+		this.durableWrites = durableWrites;
+	}
+
+}

spring-cassandra/src/main/java/org/springframework/cassandra/config/MultiLevelListFlattenerFactoryBean.java

@@ -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;
+	}
+
+}

spring-cassandra/src/main/java/org/springframework/cassandra/config/MultiLevelSetFlattenerFactoryBean.java

@@ -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;
+	}
+
+}

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

@@ -8,8 +8,6 @@ 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;

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

@@ -15,44 +15,50 @@
  */
 package org.springframework.cassandra.config.xml;
 
+import static org.springframework.data.config.ParsingUtils.getSourceBeanDefinition;
+
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+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.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.MultiLevelSetFlattenerFactoryBean;
 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.KeyspaceOption.ReplicationStrategy;
 import org.springframework.cassandra.core.keyspace.Option;
 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.AuthProvider;
 
 /**
- * Parser for <cluster;gt; definitions.
- * 
  * @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 Class<?> getBeanClass(Element element) {
+	// return CassandraClusterFactoryBean.class;
+	// }
 
 	@Override
 	protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
@@ -63,7 +69,27 @@ 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();
+	}
+
+	protected void doParse(Element element, ParserContext context, BeanDefinitionBuilder builder) {
 
 		String contactPoints = element.getAttribute("contactPoints");
 		if (StringUtils.hasText(contactPoints)) {
@@ -77,23 +103,30 @@ 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)) {
+			log.info(authProvider);
+			builder.addPropertyReference("authProvider", authProvider);
+		}
+
+		parseChildElements(element, context, builder);
 	}
 
-	protected void parseChildElements(BeanDefinitionBuilder builder, Element element) {
+	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
 		for (Element subElement : elements) {
+
 			String name = subElement.getLocalName();
 
 			if ("local-pooling-options".equals(name)) {
@@ -104,14 +137,9 @@ public class CassandraClusterParser extends AbstractSimpleBeanDefinitionParser {
 				builder.addPropertyValue("socketOptions", parseSocketOptions(subElement));
 			} 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,83 +147,80 @@ public class CassandraClusterParser extends AbstractSimpleBeanDefinitionParser {
 			}
 		}
 
-		builder.addPropertyValue("keyspaceCreations", creates);
-		builder.addPropertyValue("keyspaceDrops", drops);
+		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
+	 * @param context
+	 * @param keyspaceActionSpecificationBeanDefinitions
+	 * @return
+	 */
+	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 BeanDefinitionBuilder.
+	 * 
+	 * @param element
+	 * @param builder
+	 */
+	/**
+	 * @param element
+	 * @param builder
+	 */
+	protected void parseReplication(Element element, BeanDefinitionBuilder builder) {
 
-		String strategyClass = null;
-		if (element != null) {
-			strategyClass = element.getAttribute("class");
+		if (element == null) {
+			return;
 		}
-		if (strategyClass == null || (strategyClass = strategyClass.trim()).length() == 0) {
+
+		String strategyClass = element.getAttribute("class");
+		if (!StringUtils.hasText(strategyClass)) {
 			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;
+		String replicationFactor = null;
+		if (strategyClass.equals(ReplicationStrategy.SIMPLE_STRATEGY.getValue())) {
+
+			replicationFactor = element.getAttribute("replication-factor");
+
+			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 (replicationFactor != null) {
+			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),
+		/*
+		 * DataCenters only apply to NetworkTolopogyStrategy
+		 */
+		if (strategyClass.equals(ReplicationStrategy.NETWORK_TOPOLOGY_STRATEGY.getValue())) {
+			List<Element> dcElements = DomUtils.getChildElementsByTagName(element, "data-center");
+			for (Element dataCenter : dcElements) {
+				replicationMap.put(new DefaultOption(dataCenter.getAttribute("name"), Long.class, true, false, true),
 						dataCenter.getAttribute("replication-factor"));
 			}
 		}
 
-		return create.with(KeyspaceOption.REPLICATION, replicationMap);
+		builder.addPropertyValue("replicationOptions", replicationMap);
+
 	}
 
 	protected String parseScript(Element element) {
@@ -227,15 +252,35 @@ public class CassandraClusterParser extends AbstractSimpleBeanDefinitionParser {
 		return builder.getBeanDefinition();
 	}
 
-	protected static class KeyspaceSpecifications {
+	/**
+	 * Returns a {@link BeanDefinition} for a {@link AuthProvider} object.
+	 * 
+	 * @param element
+	 * @param context
+	 * @return the {@link BeanDefinition} or {@literal null} if auth-info-provider is not given.
+	 */
+	private BeanDefinition getKeyspaceSpecificationBeanDefinition(Element element, ParserContext context) {
 
-		public KeyspaceSpecifications(CreateKeyspaceSpecification create, DropKeyspaceSpecification drop) {
-			this.create = create;
-			this.drop = drop;
+		String name = element.getAttribute("name");
+		String action = element.getAttribute("action");
+		String durableWrites = element.getAttribute("durable-writes");
+
+		if (!StringUtils.hasText(action)) {
+			return null;
 		}
 
-		public CreateKeyspaceSpecification create;
-		public DropKeyspaceSpecification drop;
-		// TODO: public AlterKeyspaceSpecification alter;
+		BeanDefinitionBuilder keyspaceBuilder = BeanDefinitionBuilder
+				.genericBeanDefinition(KeyspaceActionSpecificationFactoryBean.class);
+		keyspaceBuilder.addPropertyValue("name", name);
+		keyspaceBuilder.addPropertyValue("action", action);
+		keyspaceBuilder.addPropertyValue("durableWrites", durableWrites);
+
+		Element replicationElement = DomUtils.getChildElementByTagName(element, "replication");
+		if (replicationElement != null) {
+			parseReplication(replicationElement, keyspaceBuilder);
+		}
+
+		return getSourceBeanDefinition(keyspaceBuilder, context, element);
 	}
+
 }

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

@@ -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;
 	}

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

@@ -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();
+	}
+
+}

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/KeyspaceActionSpecification.java

@@ -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();
+	}
+
+}

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

@@ -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);
-	}
-
-}

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

@@ -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();
+		}
+	}
+}

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

@@ -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);
+	}
+
+}

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

@@ -133,32 +133,14 @@ 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.
@@ -422,32 +404,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 +445,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[
@@ -504,7 +468,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[

spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/config/xml/PropertyPlaceholderNamespaceCreatingXmlConfigTest.java

@@ -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);
+	}
+}

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

@@ -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();
 

spring-cassandra/src/test/resources/logback-test.xml

@@ -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">

spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/PropertyPlaceholderNamespaceCreatingXmlConfigTest-context.xml

@@ -0,0 +1,34 @@
+<?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" />
+
+	<cassandra:cluster id="cassandra-cluster"
+		contactPoints="${cluster.contactPoints}" port="${cluster.port}"
+		compression="${cluster.compression}" auth-info-provider-ref="authProvider">
+		<cassandra:keyspace name="${keyspace.name}" action="${keyspace.action}"/>
+		<cassandra:keyspace name="Foo" action="CREATE_DROP" durable-writes="true">
+			<cassandra:replication class="NetworkTopologyStrategy">
+				<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>

spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/ppncxct.properties

@@ -0,0 +1,9 @@
+cluster.contactPoints=localhost
+cluster.port=9042
+cluster.compression=SNAPPY
+keyspace.name=ppncxct
+keyspace.action=CREATE
+dc1.name=DCJAX
+dc1.rf=2
+dc2.name=DCCTL
+dc2.rf=3