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

DATACASS-766 - Add support for port per endpoint.

Browse Source 
Original pull request: #178.
This commit is contained in:
tomekl007
2020-08-10 14:29:03 +02:00
committed by Mark Paluch
parent 8fff511c0f
commit 2591f3a197
2 changed files with 206 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -23,6 +23,9 @@ import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.function.IntFunction;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.slf4j.Logger;
@@ -64,6 +67,7 @@ import com.datastax.oss.driver.api.core.CqlSessionBuilder;
* @author Matthew T. Adams
* @author John Blum
* @author Mark Paluch
* @author Tomasz Lelek
* @since 3.0
*/
public class CqlSessionFactoryBean
@@ -107,12 +111,22 @@ public class CqlSessionFactoryBean
private @Nullable SessionBuilderConfigurer sessionBuilderConfigurer;
private String contactPoints = DEFAULT_CONTACT_POINTS;
private IntFunction<Collection<InetSocketAddress>> contactPoints = port -> createInetSocketAddresses(
DEFAULT_CONTACT_POINTS, port);
private @Nullable String keyspaceName;
private @Nullable String localDatacenter;
private @Nullable String password;
private @Nullable String username;
private Supplier<CqlSessionBuilder> setCqlSessionBuilderSupplier;
public CqlSessionFactoryBean() {
this(CqlSession::builder);
}
CqlSessionFactoryBean(Supplier<CqlSessionBuilder> setCqlSessionBuilderSupplier) {
this.setCqlSessionBuilderSupplier = setCqlSessionBuilderSupplier;
}
/**
* Null-safe operation to determine whether the Cassandra {@link CqlSession} is connected or not.
@@ -130,14 +144,25 @@ public class CqlSessionFactoryBean
/**
* Set a comma-delimited string of the contact points (hosts) to connect to. Default is {@code localhost}; see
* {@link #DEFAULT_CONTACT_POINTS}.
* {@link #DEFAULT_CONTACT_POINTS}. It can be in the form 'host:port', or a simple 'host' to use the configured port.
*
* @param contactPoints the contact points used by the new cluster.
*/
public void setContactPoints(String contactPoints) {
this.contactPoints = contactPoints;
this.contactPoints = port -> createInetSocketAddresses(contactPoints, port);
}
/**
* Set a collection of the contact points (hosts) to connect to. Default is {@code localhost}; see
* {@link #DEFAULT_CONTACT_POINTS}.
*
* @param contactPoints the contact points used by the new cluster.
*/
public void setContactPoints(Collection<InetSocketAddress> contactPoints) {
this.contactPoints = unusedPort -> contactPoints;
}
/**
* Sets the name of the local datacenter.
*
@@ -439,13 +464,12 @@ public class CqlSessionFactoryBean
}
protected CqlSessionBuilder buildBuilder() {
Collection<InetSocketAddress> addresses = contactPoints.apply(this.port);
Assert.notEmpty(addresses, "At least one server is required");
Assert.hasText(this.contactPoints, "At least one server is required");
CqlSessionBuilder sessionBuilder = setCqlSessionBuilderSupplier.get();
CqlSessionBuilder sessionBuilder = CqlSession.builder();
StringUtils.commaDelimitedListToSet(this.contactPoints).forEach(host ->
sessionBuilder.addContactPoint(InetSocketAddress.createUnresolved(host, this.port)));
addresses.forEach(sessionBuilder::addContactPoint);
if (StringUtils.hasText(this.username)) {
sessionBuilder.withAuthCredentials(this.username, this.password);
@@ -460,6 +484,45 @@ public class CqlSessionFactoryBean
: sessionBuilder;
}
private Collection<InetSocketAddress> createInetSocketAddresses(String contactPoints, int port) {
return StringUtils.commaDelimitedListToSet(contactPoints).stream().map(candidate -> toHostAndPort(candidate, port))
.map(hostAndPort -> InetSocketAddress.createUnresolved(hostAndPort.host, hostAndPort.port))
.collect(Collectors.toList());
}
private static class HostAndPort {
private final String host;
private final int port;
private HostAndPort(String host, int port) {
this.host = host;
this.port = port;
}
}
private HostAndPort toHostAndPort(String candidate, int port) {
int i = candidate.lastIndexOf(':');
if (i == -1 || !isPort(() -> candidate.substring(i + 1))) {
return new HostAndPort(candidate, port);
} else {
String[] hostAndPort = candidate.split(":");
if (hostAndPort.length != 2) {
throw new IllegalArgumentException(
String.format("The provided contact point: %s has wrong format.", candidate));
}
return new HostAndPort(hostAndPort[0], Integer.parseInt(hostAndPort[1]));
}
}
private boolean isPort(Supplier<String> value) {
try {
int i = Integer.parseInt(value.get());
return i > 0 && i < 65535;
} catch (Exception ex) {
return false;
}
}
/**
* Build the Cassandra {@link CqlSession System Session}.
*

135
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CqlSessionFactoryBeanUnitTests.java Normal file
View File

@@ -0,0 +1,135 @@
/*
* Copyright 2017-2020 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
*
* https://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.data.cassandra.config;
import static org.mockito.Mockito.*;
import java.net.InetSocketAddress;
import java.util.Arrays;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import com.datastax.oss.driver.api.core.CqlSessionBuilder;
/**
* @author Tomasz Lelek
*/
@ExtendWith(MockitoExtension.class)
class CqlSessionFactoryBeanUnitTests {
@Mock CqlSessionBuilder cqlSessionBuilder;
@Test
public void constructCqlSessionBuilderWithDefaultHostAndPort() {
new CqlSessionFactoryBean(() -> cqlSessionBuilder).buildBuilder();
verify(cqlSessionBuilder, times(1)).addContactPoint(InetSocketAddress
.createUnresolved(CqlSessionFactoryBean.DEFAULT_CONTACT_POINTS, CqlSessionFactoryBean.DEFAULT_PORT));
}
@Test
public void constructCqlSessionBuilderWithDefaultHostAndNonDefaultPort() {
CqlSessionFactoryBean cqlSessionFactoryBean = new CqlSessionFactoryBean(() -> cqlSessionBuilder);
cqlSessionFactoryBean.setPort(1000);
cqlSessionFactoryBean.buildBuilder();
verify(cqlSessionBuilder, times(1))
.addContactPoint(InetSocketAddress.createUnresolved(CqlSessionFactoryBean.DEFAULT_CONTACT_POINTS, 1000));
}
@Test
public void constructCqlSessionBuilderWithMultipleContactPointsSamePort() {
CqlSessionFactoryBean cqlSessionFactoryBean = new CqlSessionFactoryBean(() -> cqlSessionBuilder);
cqlSessionFactoryBean.setPort(1000);
cqlSessionFactoryBean.setContactPoints("a,b,c");
cqlSessionFactoryBean.buildBuilder();
verify(cqlSessionBuilder, times(1)).addContactPoint(InetSocketAddress.createUnresolved("a", 1000));
verify(cqlSessionBuilder, times(1)).addContactPoint(InetSocketAddress.createUnresolved("b", 1000));
verify(cqlSessionBuilder, times(1)).addContactPoint(InetSocketAddress.createUnresolved("c", 1000));
}
@Test
public void constructCqlSessionBuilderWithMultipleContactPointsDifferentPorts() {
CqlSessionFactoryBean cqlSessionFactoryBean = new CqlSessionFactoryBean(() -> cqlSessionBuilder);
cqlSessionFactoryBean.setContactPoints("a:1000,b:1001,c:1002");
cqlSessionFactoryBean.buildBuilder();
verify(cqlSessionBuilder, times(1)).addContactPoint(InetSocketAddress.createUnresolved("a", 1000));
verify(cqlSessionBuilder, times(1)).addContactPoint(InetSocketAddress.createUnresolved("b", 1001));
verify(cqlSessionBuilder, times(1)).addContactPoint(InetSocketAddress.createUnresolved("c", 1002));
}
@Test
public void throwWhenContactPointsWithPortHasWrongFormat() {
CqlSessionFactoryBean cqlSessionFactoryBean = new CqlSessionFactoryBean(() -> cqlSessionBuilder);
cqlSessionFactoryBean.setContactPoints("a:1000:100");
Assertions.assertThatThrownBy(cqlSessionFactoryBean::buildBuilder).isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("The provided contact point: a:1000:100 has wrong format.");
}
@Test
public void constructCqlSessionBuilderWithExplictPortsAndDefaultPort() {
CqlSessionFactoryBean cqlSessionFactoryBean = new CqlSessionFactoryBean(() -> cqlSessionBuilder);
cqlSessionFactoryBean.setContactPoints("a:1000,b:2000,c");
cqlSessionFactoryBean.buildBuilder();
verify(cqlSessionBuilder, times(1)).addContactPoint(InetSocketAddress.createUnresolved("a", 1000));
verify(cqlSessionBuilder, times(1)).addContactPoint(InetSocketAddress.createUnresolved("b", 2000));
verify(cqlSessionBuilder, times(1))
.addContactPoint(InetSocketAddress.createUnresolved("c", CqlSessionFactoryBean.DEFAULT_PORT));
}
@Test
public void constructCqlSessionBuilderWithIpv6ContactPoint() {
CqlSessionFactoryBean cqlSessionFactoryBean = new CqlSessionFactoryBean(() -> cqlSessionBuilder);
cqlSessionFactoryBean.setPort(1000);
cqlSessionFactoryBean.setContactPoints("[2001:db8:85a3:8d3:1319:8a2e:370:7348]");
cqlSessionFactoryBean.buildBuilder();
verify(cqlSessionBuilder, times(1))
.addContactPoint(InetSocketAddress.createUnresolved("[2001:db8:85a3:8d3:1319:8a2e:370:7348]", 1000));
}
@Test
public void constructCqlSessionBuilderWithContactPointsProvidedAsInetSocketAddress() {
CqlSessionFactoryBean cqlSessionFactoryBean = new CqlSessionFactoryBean(() -> cqlSessionBuilder);
cqlSessionFactoryBean.setContactPoints(Arrays.asList(InetSocketAddress.createUnresolved("a", 1000)));
cqlSessionFactoryBean.buildBuilder();
verify(cqlSessionBuilder, times(1)).addContactPoint(InetSocketAddress.createUnresolved("a", 1000));
}
@Test
public void constructCqlSessionBuilderLastSetContactPointsOverridePreviousInet() {
CqlSessionFactoryBean cqlSessionFactoryBean = new CqlSessionFactoryBean(() -> cqlSessionBuilder);
cqlSessionFactoryBean.setContactPoints(Arrays.asList(InetSocketAddress.createUnresolved("a", 1000)));
cqlSessionFactoryBean.setContactPoints("b:1000");
cqlSessionFactoryBean.buildBuilder();
verify(cqlSessionBuilder, times(1)).addContactPoint(InetSocketAddress.createUnresolved("b", 1000));
}
@Test
public void constructCqlSessionBuilderLastSetContactPointsOverridePreviousString() {
CqlSessionFactoryBean cqlSessionFactoryBean = new CqlSessionFactoryBean(() -> cqlSessionBuilder);
cqlSessionFactoryBean.setContactPoints("b:1000");
cqlSessionFactoryBean.setContactPoints(Arrays.asList(InetSocketAddress.createUnresolved("a", 1000)));
cqlSessionFactoryBean.buildBuilder();
verify(cqlSessionBuilder, times(1)).addContactPoint(InetSocketAddress.createUnresolved("a", 1000));
}
}