diff --git a/spring-cassandra/src/main/java/org/springframework/cassandra/config/CassandraClusterFactoryBean.java b/spring-cassandra/src/main/java/org/springframework/cassandra/config/CassandraClusterFactoryBean.java index 5c7610159..69453baec 100644 --- a/spring-cassandra/src/main/java/org/springframework/cassandra/config/CassandraClusterFactoryBean.java +++ b/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, 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, 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> keyspaceSpecifications = new HashSet>(); private List keyspaceCreations = new ArrayList(); private List keyspaceDrops = new ArrayList(); @@ -124,20 +127,20 @@ public class CassandraClusterFactoryBean implements FactoryBean, 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, 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, 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, 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, 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; } } diff --git a/spring-cassandra/src/main/java/org/springframework/cassandra/config/KeyspaceActionSpecificationFactoryBean.java b/spring-cassandra/src/main/java/org/springframework/cassandra/config/KeyspaceActionSpecificationFactoryBean.java index 5fb25d3fe..e44d16587 100644 --- a/spring-cassandra/src/main/java/org/springframework/cassandra/config/KeyspaceActionSpecificationFactoryBean.java +++ b/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>>, @@ -45,7 +52,10 @@ public class KeyspaceActionSpecificationFactoryBean implements FactoryBean replicationOptions = new LinkedHashMap(); + private List networkTopologyDataCenters = new LinkedList(); + private List networkTopologyReplicationFactors = new LinkedList(); + private String replicationStrategy; + private long replicationFactor; private boolean durableWrites = false; private boolean ifNotExists = false; @@ -53,8 +63,11 @@ public class KeyspaceActionSpecificationFactoryBean implements FactoryBean 0) { - create.with(KeyspaceOption.REPLICATION, replicationOptions); - } else { - Map defaultReplicationStrategyMap = new HashMap(); - 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 replicationStrategyMap = new HashMap(); + 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 getReplicationOptions() { - return replicationOptions; - } - - /** - * @param replicationOptions The replicationOptions to set. - */ - public void setReplicationOptions(Map replicationOptions) { - this.replicationOptions = replicationOptions; - } - /** * @return Returns the durableWrites. */ @@ -182,4 +208,60 @@ public class KeyspaceActionSpecificationFactoryBean implements FactoryBean getNetworkTopologyDataCenters() { + return networkTopologyDataCenters; + } + + /** + * @param networkTopologyDataCenters The networkTopologyDataCenters to set. + */ + public void setNetworkTopologyDataCenters(List networkTopologyDataCenters) { + this.networkTopologyDataCenters = networkTopologyDataCenters; + } + + /** + * @return Returns the networkTopologyReplicationFactors. + */ + public List getNetworkTopologyReplicationFactors() { + return networkTopologyReplicationFactors; + } + + /** + * @param networkTopologyReplicationFactors The networkTopologyReplicationFactors to set. + */ + public void setNetworkTopologyReplicationFactors(List 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; + } + } diff --git a/spring-cassandra/src/main/java/org/springframework/cassandra/config/PoolingOptionsConfig.java b/spring-cassandra/src/main/java/org/springframework/cassandra/config/PoolingOptionsConfig.java deleted file mode 100644 index e982e217d..000000000 --- a/spring-cassandra/src/main/java/org/springframework/cassandra/config/PoolingOptionsConfig.java +++ /dev/null @@ -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; - } -} diff --git a/spring-cassandra/src/main/java/org/springframework/cassandra/config/PoolingOptionsFactoryBean.java b/spring-cassandra/src/main/java/org/springframework/cassandra/config/PoolingOptionsFactoryBean.java new file mode 100644 index 000000000..d9f4489b3 --- /dev/null +++ b/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, 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; + } + +} diff --git a/spring-cassandra/src/main/java/org/springframework/cassandra/config/SocketOptionsConfig.java b/spring-cassandra/src/main/java/org/springframework/cassandra/config/SocketOptionsConfig.java deleted file mode 100644 index 562377415..000000000 --- a/spring-cassandra/src/main/java/org/springframework/cassandra/config/SocketOptionsConfig.java +++ /dev/null @@ -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; - } -} diff --git a/spring-cassandra/src/main/java/org/springframework/cassandra/config/SocketOptionsFactoryBean.java b/spring-cassandra/src/main/java/org/springframework/cassandra/config/SocketOptionsFactoryBean.java new file mode 100644 index 000000000..164aac8e5 --- /dev/null +++ b/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, 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; + } + +} diff --git a/spring-cassandra/src/main/java/org/springframework/cassandra/config/java/AbstractCassandraConfiguration.java b/spring-cassandra/src/main/java/org/springframework/cassandra/config/java/AbstractCassandraConfiguration.java index 9ec7b70e1..6e116716f 100644 --- a/spring-cassandra/src/main/java/org/springframework/cassandra/config/java/AbstractCassandraConfiguration.java +++ b/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; } diff --git a/spring-cassandra/src/main/java/org/springframework/cassandra/config/xml/CassandraClusterParser.java b/spring-cassandra/src/main/java/org/springframework/cassandra/config/xml/CassandraClusterParser.java index 231aa1e3c..aafdd3a38 100644 --- a/spring-cassandra/src/main/java/org/springframework/cassandra/config/xml/CassandraClusterParser.java +++ b/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 } 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 keyspaceActionSpecificationBeanDefinitions = new ManagedSet(); @@ -124,17 +174,26 @@ public class CassandraClusterParser extends AbstractBeanDefinitionParser { List 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> * - * @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 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 networkTopologyDataCenters = new ManagedList(); + ManagedList networkTopologyReplicationFactors = new ManagedList(); + 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 replicationMap = new HashMap(); - 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 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); } diff --git a/spring-cassandra/src/main/resources/org/springframework/cassandra/config/spring-cassandra-1.0.xsd b/spring-cassandra/src/main/resources/org/springframework/cassandra/config/spring-cassandra-1.0.xsd index 03fdb1ac6..45bb9ddfc 100644 --- a/spring-cassandra/src/main/resources/org/springframework/cassandra/config/spring-cassandra-1.0.xsd +++ b/spring-cassandra/src/main/resources/org/springframework/cassandra/config/spring-cassandra-1.0.xsd @@ -59,22 +59,16 @@ Defines a Cassandra cluster. - - - - - - + + + @@ -133,13 +127,15 @@ The native CQL port to connect to. Default is 9042. ]]> - + + - + + + + + + + + + + + + + + + + + + + + + + + + + + + - + - + - @@ -267,7 +298,7 @@ More connections are created up to a configurable maximum number of connections. - + + + + + + - + - - - + - - + diff --git a/spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/MinimalKeyspaceCreatingXmlConfigTest-context.xml b/spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/MinimalKeyspaceCreatingXmlConfigTest-context.xml index 688cb3dba..05939b0d0 100644 --- a/spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/MinimalKeyspaceCreatingXmlConfigTest-context.xml +++ b/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"> - + diff --git a/spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/PropertyPlaceholderNamespaceCreatingXmlConfigTest-context.xml b/spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/PropertyPlaceholderNamespaceCreatingXmlConfigTest-context.xml index 2c7f28a31..81d7ff8ce 100644 --- a/spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/PropertyPlaceholderNamespaceCreatingXmlConfigTest-context.xml +++ b/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" /> + + + + + + + + + + + + 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"> + + + - + diff --git a/spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/XmlConfigTest-context.xml b/spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/XmlConfigTest-context.xml index 39e2ff612..3681a711e 100644 --- a/spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/XmlConfigTest-context.xml +++ b/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" /> diff --git a/spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/ppncxct.properties b/spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/ppncxct.properties index b89cbde8e..b8d517dc4 100644 --- a/spring-cassandra/src/test/resources/org/springframework/cassandra/test/integration/config/xml/ppncxct.properties +++ b/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