From 2591f3a197a75285f370252660fbd8a105e9b4a9 Mon Sep 17 00:00:00 2001 From: tomekl007 Date: Mon, 10 Aug 2020 14:29:03 +0200 Subject: [PATCH] DATACASS-766 - Add support for port per endpoint. Original pull request: #178. --- .../config/CqlSessionFactoryBean.java | 79 ++++++++-- .../CqlSessionFactoryBeanUnitTests.java | 135 ++++++++++++++++++ 2 files changed, 206 insertions(+), 8 deletions(-) create mode 100644 spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CqlSessionFactoryBeanUnitTests.java diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CqlSessionFactoryBean.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CqlSessionFactoryBean.java index 32a0c75ba..8d36521e3 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CqlSessionFactoryBean.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CqlSessionFactoryBean.java @@ -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> 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 setCqlSessionBuilderSupplier; + + public CqlSessionFactoryBean() { + this(CqlSession::builder); + } + + CqlSessionFactoryBean(Supplier 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 contactPoints) { + this.contactPoints = unusedPort -> contactPoints; + } + + /** * Sets the name of the local datacenter. * @@ -439,13 +464,12 @@ public class CqlSessionFactoryBean } protected CqlSessionBuilder buildBuilder() { + Collection 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 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 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}. * diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CqlSessionFactoryBeanUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CqlSessionFactoryBeanUnitTests.java new file mode 100644 index 000000000..5f5ae847f --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CqlSessionFactoryBeanUnitTests.java @@ -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)); + } +}