DATACASS-71 - Completed refactoring for compliance with property

placeholders.

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

@@ -39,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;
@@ -60,12 +59,13 @@ public class CassandraClusterFactoryBean implements FactoryBean<Cluster>, Initia
 
 	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;
-	private boolean accumulating = true;
 
 	/*
 	 * Attributes needed for cluster builder
@@ -73,14 +73,17 @@ public class CassandraClusterFactoryBean implements FactoryBean<Cluster>, Initia
 	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>();
@@ -124,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) {
@@ -152,10 +155,18 @@ 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();
@@ -256,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;
 	}
 
@@ -322,55 +329,6 @@ 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;
-	}
-
-	private static SocketOptions configSocketOptions(SocketOptionsConfig config) {
-		SocketOptions socketOptions = new SocketOptions();
-
-		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());
-		}
-
-		return socketOptions;
-	}
-
 	/**
 	 * @return Returns the keyspaceSpecifications.
 	 */
@@ -389,16 +347,30 @@ public class CassandraClusterFactoryBean implements FactoryBean<Cluster>, Initia
 	}
 
 	/**
-	 * @return Returns the accumulating.
+	 * @param username The username to set.
 	 */
-	public boolean isAccumulating() {
-		return accumulating;
+	public void setUsername(String username) {
+		this.username = username;
 	}
 
 	/**
-	 * @param accumulating The accumulating to set.
+	 * @param password The password to set.
 	 */
-	public void setAccumulating(boolean accumulating) {
-		this.accumulating = accumulating;
+	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;
 	}
 }

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

@@ -17,7 +17,8 @@ package org.springframework.cassandra.config;
 
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
@@ -31,11 +32,17 @@ 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;
 
 /**
- * @author David Webb (dwebb@brightmove.com)
+ * 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<?>>>,
@@ -45,7 +52,10 @@ public class KeyspaceActionSpecificationFactoryBean implements FactoryBean<Set<K
 
 	private KeyspaceAction action;
 	private String name;
-	private Map<Option, Object> replicationOptions = new LinkedHashMap<Option, Object>();
+	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;
 
@@ -53,8 +63,11 @@ public class KeyspaceActionSpecificationFactoryBean implements FactoryBean<Set<K
 
 	@Override
 	public void destroy() throws Exception {
+		action = null;
 		name = null;
-		replicationOptions = null;
+		networkTopologyDataCenters = null;
+		networkTopologyReplicationFactors = null;
+		replicationStrategy = null;
 		specs = null;
 	}
 
@@ -68,6 +81,7 @@ public class KeyspaceActionSpecificationFactoryBean implements FactoryBean<Set<K
 		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:
@@ -76,21 +90,47 @@ public class KeyspaceActionSpecificationFactoryBean implements FactoryBean<Set<K
 
 	}
 
+	/**
+	 * 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);
-		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);
+
+		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());
@@ -154,20 +194,6 @@ public class KeyspaceActionSpecificationFactoryBean implements FactoryBean<Set<K
 		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.
 	 */
@@ -182,4 +208,60 @@ public class KeyspaceActionSpecificationFactoryBean implements FactoryBean<Set<K
 		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;
+	}
+
 }

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

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

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

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

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

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

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

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

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

@@ -6,8 +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.keyspace.CreateKeyspaceSpecification;
 import org.springframework.cassandra.core.keyspace.DropKeyspaceSpecification;
 import org.springframework.context.annotation.Bean;
@@ -15,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;
@@ -40,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());
@@ -69,7 +68,7 @@ public abstract class AbstractCassandraConfiguration {
 		return Collections.emptyList();
 	}
 
-	protected SocketOptionsConfig getSocketOptions() {
+	protected SocketOptions getSocketOptions() {
 		return null;
 	}
 
@@ -85,7 +84,7 @@ public abstract class AbstractCassandraConfiguration {
 		return null;
 	}
 
-	protected PoolingOptionsConfig getRemotePoolingOptions() {
+	protected PoolingOptions getPoolingOptions() {
 		return null;
 	}
 
@@ -97,10 +96,6 @@ public abstract class AbstractCassandraConfiguration {
 		return CassandraClusterFactoryBean.DEFAULT_METRICS_ENABLED;
 	}
 
-	protected PoolingOptionsConfig getLocalPoolingOptions() {
-		return null;
-	}
-
 	protected LoadBalancingPolicy getLoadBalancingPolicy() {
 		return null;
 	}

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

@@ -18,9 +18,7 @@ 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;
@@ -28,6 +26,7 @@ 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.support.ManagedList;
 import org.springframework.beans.factory.support.ManagedSet;
 import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
@@ -35,19 +34,22 @@ import org.springframework.cassandra.config.CassandraClusterFactoryBean;
 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.DefaultOption;
+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.cassandra.core.keyspace.Option;
+import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
 import org.w3c.dom.Element;
 
-import com.datastax.driver.core.AuthProvider;
+import com.datastax.driver.core.HostDistance;
+import com.datastax.driver.core.PoolingOptions;
+import com.datastax.driver.core.SocketOptions;
 
 /**
- * @author Alex Shvid
+ * Parses the {@literal <cluster>} element of the XML Configuration.
+ * 
  * @author Matthew T. Adams
  * @author David Webb
  */
@@ -55,11 +57,6 @@ public class CassandraClusterParser extends AbstractBeanDefinitionParser {
 
 	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)
 			throws BeanDefinitionStoreException {
@@ -89,6 +86,13 @@ public class CassandraClusterParser extends AbstractBeanDefinitionParser {
 		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");
@@ -108,13 +112,59 @@ public class CassandraClusterParser extends AbstractBeanDefinitionParser {
 
 		String authProvider = element.getAttribute("auth-info-provider-ref");
 		if (StringUtils.hasText(authProvider)) {
-			log.info(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);
 	}
 
+	/**
+	 * 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) {
 
 		ManagedSet<BeanDefinition> keyspaceActionSpecificationBeanDefinitions = new ManagedSet<BeanDefinition>();
@@ -124,17 +174,26 @@ public class CassandraClusterParser extends AbstractBeanDefinitionParser {
 		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)) {
 
 				keyspaceActionSpecificationBeanDefinition = getKeyspaceSpecificationBeanDefinition(subElement, context);
@@ -147,6 +206,13 @@ public class CassandraClusterParser extends AbstractBeanDefinitionParser {
 			}
 		}
 
+		/*
+		 * 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);
@@ -156,10 +222,10 @@ public class CassandraClusterParser extends AbstractBeanDefinitionParser {
 	/**
 	 * Create the Single Factory Bean that will flatten all List<List<KeyspaceActionSpecificationFactoryBean>>
 	 * 
-	 * @param element
-	 * @param context
-	 * @param keyspaceActionSpecificationBeanDefinitions
-	 * @return
+	 * @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) {
@@ -171,114 +237,136 @@ public class CassandraClusterParser extends AbstractBeanDefinitionParser {
 	}
 
 	/**
-	 * Parses the keyspace replication options and adds them to the supplied BeanDefinitionBuilder.
+	 * Parses the keyspace replication options and adds them to the supplied {@link BeanDefinitionBuilder}.
 	 * 
-	 * @param element
-	 * @param builder
-	 */
-	/**
-	 * @param element
-	 * @param builder
+	 * @param element The Element being parsed
+	 * @param builder The {@link BeanDefinitionBuilder} to add the replication to
 	 */
 	protected void parseReplication(Element element, BeanDefinitionBuilder builder) {
 
-		if (element == null) {
-			return;
-		}
-
-		String strategyClass = element.getAttribute("class");
-		if (!StringUtils.hasText(strategyClass)) {
-			strategyClass = KeyspaceAttributes.DEFAULT_REPLICATION_STRATEGY;
-		}
-
+		ManagedList<String> networkTopologyDataCenters = new ManagedList<String>();
+		ManagedList<String> networkTopologyReplicationFactors = new ManagedList<String>();
+		String strategyClass = null;
 		String replicationFactor = null;
-		if (strategyClass.equals(ReplicationStrategy.SIMPLE_STRATEGY.getValue())) {
+
+		if (element != null) {
+
+			strategyClass = element.getAttribute("class");
+			if (!StringUtils.hasText(strategyClass)) {
+				strategyClass = KeyspaceAttributes.DEFAULT_REPLICATION_STRATEGY;
+			}
 
 			replicationFactor = element.getAttribute("replication-factor");
-
-			if (replicationFactor == null) {
+			if (!StringUtils.hasText(replicationFactor)) {
 				replicationFactor = KeyspaceAttributes.DEFAULT_REPLICATION_FACTOR + "";
 			}
 
-		}
-
-		Map<Option, Object> replicationMap = new HashMap<Option, Object>();
-		replicationMap.put(new DefaultOption("class", String.class, false, false, true), strategyClass);
-		if (replicationFactor != null) {
-			replicationMap.put(new DefaultOption("replication_factor", Long.class, true, false, false), replicationFactor);
-		}
-
-		/*
-		 * DataCenters only apply to NetworkTolopogyStrategy
-		 */
-		if (strategyClass.equals(ReplicationStrategy.NETWORK_TOPOLOGY_STRATEGY.getValue())) {
+			/*
+			 * DataCenters only apply to NetworkTolopogyStrategy
+			 */
 			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"));
+				networkTopologyDataCenters.add(dataCenter.getAttribute("name"));
+				networkTopologyReplicationFactors.add(dataCenter.getAttribute("replication-factor"));
 			}
+		} else {
+			strategyClass = ReplicationStrategy.SIMPLE_STRATEGY.name();
+			replicationFactor = KeyspaceAttributes.DEFAULT_REPLICATION_FACTOR + "";
 		}
 
-		builder.addPropertyValue("replicationOptions", 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);
 	}
 
 	/**
-	 * Returns a {@link BeanDefinition} for a {@link AuthProvider} object.
+	 * Returns a {@link BeanDefinition} for a {@link KeyspaceActionSpecification} object.
 	 * 
-	 * @param element
-	 * @param context
-	 * @return the {@link BeanDefinition} or {@literal null} if auth-info-provider is not given.
+	 * @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) {
 
-		String name = element.getAttribute("name");
 		String action = element.getAttribute("action");
-		String durableWrites = element.getAttribute("durable-writes");
 
-		if (!StringUtils.hasText(action)) {
-			return null;
-		}
+		Assert.notNull(action, "Keyspace Action must not be null!");
 
 		BeanDefinitionBuilder keyspaceBuilder = BeanDefinitionBuilder
 				.genericBeanDefinition(KeyspaceActionSpecificationFactoryBean.class);
-		keyspaceBuilder.addPropertyValue("name", name);
-		keyspaceBuilder.addPropertyValue("action", action);
-		keyspaceBuilder.addPropertyValue("durableWrites", durableWrites);
+
+		ParsingUtils.setPropertyValue(keyspaceBuilder, element, "name", "name");
+		ParsingUtils.setPropertyValue(keyspaceBuilder, element, "action", "action");
+		ParsingUtils.setPropertyValue(keyspaceBuilder, element, "durableWrites", "durableWrites");
 
 		Element replicationElement = DomUtils.getChildElementByTagName(element, "replication");
-		if (replicationElement != null) {
-			parseReplication(replicationElement, keyspaceBuilder);
-		}
+		parseReplication(replicationElement, keyspaceBuilder);
 
 		return getSourceBeanDefinition(keyspaceBuilder, context, element);
 	}

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

@@ -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,13 +127,15 @@ The native CQL port to connect to.  Default is 9042.
 				]]></xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
-		<xsd:attribute name="compression" default="NONE" use="optional" type="xsd:string">
+		<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:attribute>
+
 		<xsd:attribute name="auth-info-provider-ref" use="optional">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
@@ -157,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.
@@ -175,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.
@@ -193,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.
@@ -212,7 +244,6 @@ RetryPolicy implementation.
 			</xsd:simpleType>
 		</xsd:attribute>
 	</xsd:complexType>
-
 	<xsd:simpleType name="clusterRef" final="union">
 		<xsd:annotation>
 			<xsd:appinfo>
@@ -267,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.
@@ -281,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[
@@ -339,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[
@@ -454,7 +492,6 @@ The replication factor; default is 1.
 			</xsd:annotation>
 		</xsd:attribute>
 	</xsd:complexType>
-
 	<xsd:complexType name="datacenterType">
 		<xsd:annotation>
 			<xsd:documentation><![CDATA[

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

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

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

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

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

@@ -10,13 +10,51 @@
 		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">
+		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="NetworkTopologyStrategy">
+			<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>

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

@@ -18,7 +18,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>

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

@@ -1,9 +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