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

Support awareness and discovery of secure (Auth and SSL eanbled) Apache Geode clusters using @EnableClusterAware annotation functionality.

Browse Source 
Improves log details about the runtime environment (e.g. cloud, standalone/self-managed).

Resolves gh-57.

Resolves gh-99.
This commit is contained in:
John Blum
2021-01-15 01:32:11 -08:00
parent d08525708b
commit 4a20f7be61
4 changed files with 1059 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

224
spring-geode-autoconfigure/src/test/java/org/springframework/geode/boot/autoconfigure/cluster/aware/SecureClusterAwareConfigurationIntegrationTests.java Normal file
View File

@@ -0,0 +1,224 @@
/*
* Copyright 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.geode.boot.autoconfigure.cluster.aware;
import static org.assertj.core.api.Assertions.assertThat;
import java.io.File;
import java.io.IOException;
import java.security.KeyStore;
import java.util.List;
import java.util.stream.Collectors;
import javax.net.ssl.SSLContext;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.apache.geode.cache.GemFireCache;
import org.apache.geode.cache.Region;
import org.apache.http.client.HttpClient;
import org.apache.http.conn.ssl.NoopHostnameVerifier;
import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
import org.apache.http.conn.ssl.TrustAllStrategy;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.ssl.SSLContexts;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Profile;
import org.springframework.core.env.Environment;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.gemfire.GemfireTemplate;
import org.springframework.data.gemfire.GemfireUtils;
import org.springframework.data.gemfire.config.annotation.CacheServerApplication;
import org.springframework.data.gemfire.config.annotation.EnableEntityDefinedRegions;
import org.springframework.data.gemfire.config.annotation.EnableManager;
import org.springframework.data.gemfire.config.support.RestTemplateConfigurer;
import org.springframework.data.gemfire.tests.integration.ForkingClientServerIntegrationTestsSupport;
import org.springframework.geode.config.annotation.EnableClusterAware;
import org.springframework.geode.security.TestSecurityManager;
import org.springframework.geode.util.GeodeConstants;
import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.junit4.SpringRunner;
import example.app.books.model.Book;
import example.app.books.model.ISBN;
/**
* Integration Tests testing the {@link EnableClusterAware} annotation configuration when the Apache Geode cluster
* (server(s)) are secure (i.e. when both Authentication and TLS/SSL are enabled).
*
* @author John Blum
* @see org.junit.Test
* @see org.apache.geode.cache.GemFireCache
* @see org.apache.geode.cache.Region
* @see org.springframework.boot.ApplicationRunner
* @see org.springframework.boot.autoconfigure.SpringBootApplication
* @see org.springframework.boot.builder.SpringApplicationBuilder
* @see org.springframework.boot.test.context.SpringBootTest
* @see org.springframework.context.annotation.Bean
* @see org.springframework.context.annotation.Profile
* @see org.springframework.data.gemfire.GemfireTemplate
* @see org.springframework.data.gemfire.config.annotation.CacheServerApplication
* @see org.springframework.data.gemfire.config.annotation.EnableManager
* @see org.springframework.data.gemfire.tests.integration.ForkingClientServerIntegrationTestsSupport
* @see org.springframework.geode.config.annotation.EnableClusterAware
* @see org.springframework.geode.security.TestSecurityManager
* @see org.springframework.test.annotation.DirtiesContext
* @see org.springframework.test.context.ActiveProfiles
* @see org.springframework.test.context.junit4.SpringRunner
* @since 1.4.1
*/
@ActiveProfiles({ "cluster-aware-with-secure-client", "ssl" })
@DirtiesContext
@RunWith(SpringRunner.class)
@SpringBootTest(
classes = SecureClusterAwareConfigurationIntegrationTests.TestGeodeClientConfiguration.class,
properties = {
"spring.data.gemfire.management.require-https=true",
"spring.data.gemfire.security.username=test",
"spring.data.gemfire.security.password=test"
},
webEnvironment = SpringBootTest.WebEnvironment.NONE
)
@SuppressWarnings("unused")
public class SecureClusterAwareConfigurationIntegrationTests extends ForkingClientServerIntegrationTestsSupport {
@BeforeClass
public static void startGeodeServer() throws IOException {
startGemFireServer(TestGeodeServerConfiguration.class,
"-Dspring.profiles.active=cluster-aware-with-secure-server,ssl");
}
@Autowired
@Qualifier("booksTemplate")
private GemfireTemplate booksTemplate;
@Test
public void clientServerConfigurationAndConfigurationIsSuccessful() {
Book book = Book.newBook("Book of Job")
.identifiedBy(ISBN.autoGenerated());
this.booksTemplate.put(book.getIsbn(), book);
Book returnedBook = this.booksTemplate.get(book.getIsbn());
assertThat(returnedBook).isNotNull();
assertThat(returnedBook).isEqualTo(book);
assertThat(returnedBook).isNotSameAs(book);
}
@SpringBootApplication
@EnableClusterAware
@EnableEntityDefinedRegions(basePackageClasses = Book.class)
@Profile(("cluster-aware-with-secure-client"))
static class TestGeodeClientConfiguration {
static final String TEST_TRUSTED_KEYSTORE_FILENAME = "test-trusted.keystore";
@Bean
RestTemplateConfigurer secureClientHttpRequestConfigurer(Environment environment) {
return restTemplate -> {
try {
KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());
keyStore.load(new ClassPathResource(TEST_TRUSTED_KEYSTORE_FILENAME).getInputStream(),
environment.getProperty("spring.data.gemfire.security.ssl.truststore.password").toCharArray());
SSLContext sslContext = SSLContexts.custom()
.loadTrustMaterial(keyStore, TrustAllStrategy.INSTANCE)
.build();
HttpClient httpClient = HttpClients.custom()
.setSSLSocketFactory(new SSLConnectionSocketFactory(sslContext, new NoopHostnameVerifier()))
.build();
restTemplate.setRequestFactory(new HttpComponentsClientHttpRequestFactory(httpClient));
}
catch (Exception cause) {
throw new RuntimeException(cause);
}
};
}
}
@SpringBootApplication
@CacheServerApplication(name = "SecureClusterAwareConfigurationIntegrationTestsServer")
@EnableManager(start = true)
@Profile("cluster-aware-with-secure-server")
static class TestGeodeServerConfiguration {
private static final String GEODE_HOME_PROPERTY = GeodeConstants.GEMFIRE_PROPERTY_PREFIX + "home";
public static void main(String[] args) throws IOException {
resolveAndConfigureGeodeHome();
new SpringApplicationBuilder(TestGeodeServerConfiguration.class)
.web(WebApplicationType.NONE)
.build()
.run(args);
}
private static void resolveAndConfigureGeodeHome() throws IOException {
ClassPathResource resource = new ClassPathResource("/geode-home");
File resourceFile = resource.getFile();
System.setProperty(GEODE_HOME_PROPERTY, resourceFile.getAbsolutePath());
}
@Bean
TestSecurityManager securityManager() {
return new TestSecurityManager();
}
@Bean
ApplicationRunner peerCacheVerifier(GemFireCache cache) {
return args -> {
assertThat(cache).isNotNull();
assertThat(GemfireUtils.isPeer(cache)).isTrue();
assertThat(cache.getName()).isEqualTo("SecureClusterAwareConfigurationIntegrationTestsServer");
List<String> regionNames = cache.rootRegions().stream()
.map(Region::getName)
.collect(Collectors.toList());
assertThat(regionNames)
.describedAs("Expected no Regions; but was [%s]", regionNames)
.isEmpty();
};
}
}
}

349
spring-geode/src/main/java/org/springframework/geode/config/annotation/ClusterAwareConfiguration.java
View File

@@ -22,12 +22,25 @@ import java.net.Socket;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.function.UnaryOperator;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import org.apache.geode.cache.client.ClientCache;
import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.server.CacheServer;
import org.springframework.boot.cloud.CloudPlatform;
@@ -48,6 +61,8 @@ import org.springframework.data.gemfire.config.annotation.support.AbstractAnnota
import org.springframework.data.gemfire.support.ConnectionEndpoint;
import org.springframework.data.gemfire.support.ConnectionEndpointList;
import org.springframework.data.gemfire.util.ArrayUtils;
import org.springframework.data.gemfire.util.CollectionUtils;
import org.springframework.geode.cache.SimpleCacheResolver;
import org.springframework.geode.core.util.ObjectUtils;
import org.springframework.lang.NonNull;
import org.springframework.lang.Nullable;
@@ -63,9 +78,13 @@ import org.slf4j.LoggerFactory;
*
* @author John Blum
* @see java.lang.annotation.Annotation
* @see java.net.InetSocketAddress
* @see java.net.Socket
* @see java.net.SocketAddress
* @see org.apache.geode.cache.client.ClientCache
* @see org.apache.geode.cache.client.ClientRegionShortcut
* @see org.apache.geode.cache.client.Pool
* @see org.apache.geode.cache.client.PoolManager
* @see org.apache.geode.cache.server.CacheServer
* @see org.springframework.boot.cloud.CloudPlatform
* @see org.springframework.context.ApplicationListener
@@ -83,12 +102,15 @@ import org.slf4j.LoggerFactory;
* @see org.springframework.data.gemfire.config.annotation.support.AbstractAnnotationConfigSupport
* @see org.springframework.data.gemfire.support.ConnectionEndpoint
* @see org.springframework.data.gemfire.support.ConnectionEndpointList
* @see org.springframework.geode.cache.SimpleCacheResolver
* @since 1.2.0
*/
@Configuration
@Import({ ClusterAvailableConfiguration.class, ClusterNotAvailableConfiguration.class })
public class ClusterAwareConfiguration extends AbstractAnnotationConfigSupport {
static final boolean DEFAULT_CLUSTER_CONDITION_MATCH = false;
static final int DEFAULT_CACHE_SERVER_PORT = CacheServer.DEFAULT_PORT;
static final int DEFAULT_LOCATOR_PORT = 10334;
static final int DEFAULT_TIMEOUT_IN_MILLISECONDS = 500;
@@ -106,6 +128,13 @@ public class ClusterAwareConfiguration extends AbstractAnnotationConfigSupport {
static final String SPRING_DATA_GEMFIRE_CACHE_CLIENT_REGION_SHORTCUT_PROPERTY =
"spring.data.gemfire.cache.client.region.shortcut";
private static final Function<ConditionContext, Boolean> configuredMatchFunction = conditionContext ->
Optional.ofNullable(conditionContext)
.map(ConditionContext::getEnvironment)
.map(environment -> environment.getProperty(SPRING_BOOT_DATA_GEMFIRE_CLUSTER_CONDITION_MATCH_PROPERTY,
Boolean.class, DEFAULT_CLUSTER_CONDITION_MATCH))
.orElse(DEFAULT_CLUSTER_CONDITION_MATCH);
@Override
protected Class<? extends Annotation> getAnnotationType() {
return EnableClusterAware.class;
@@ -116,7 +145,7 @@ public class ClusterAwareConfiguration extends AbstractAnnotationConfigSupport {
private static final AtomicReference<Boolean> clusterAvailable = new AtomicReference<>(null);
private static ApplicationListener<ContextClosedEvent> clusterAwareConditionResetApplicationListener() {
private static ApplicationListener<ContextClosedEvent> clusterAwareConditionResetOnContextClosedApplicationListener() {
return contextClosedEvent-> reset();
}
@@ -132,68 +161,96 @@ public class ClusterAwareConfiguration extends AbstractAnnotationConfigSupport {
* @inheritDoc
*/
@Override
public synchronized boolean matches(@NonNull ConditionContext context, @NonNull AnnotatedTypeMetadata metadata) {
public synchronized boolean matches(@NonNull ConditionContext conditionContext,
@NonNull AnnotatedTypeMetadata metadata) {
if (clusterAvailable.get() == null) {
registerApplicationListener(context);
doMatch(context);
}
return isMatch(context);
return isMatch(conditionContext) || doCachedMatch(conditionContext);
}
@NonNull ConditionContext registerApplicationListener(@NonNull ConditionContext conditionContext) {
boolean isMatch(@NonNull ConditionContext conditionContext) {
return isAvailable() || configuredMatchFunction.apply(conditionContext);
}
/**
* Caches the result of the computed {@link #doMatch(ConditionContext)} operation.
*
* Subsequent calls returns the cached value of the computed (once) {@link #doMatch(ConditionContext)}
* operation.
*
* @param conditionContext Spring {@link ConditionContext} capturing the context in which the conditions
* are evaluated; must not be {@literal null}.
* @return a boolean value indicating whether the conditions match (i.e. {@literal true}).
* @see org.springframework.context.annotation.ConditionContext
* @see #registerApplicationListener(ConditionContext)
* @see #doMatch(ConditionContext)
*/
protected boolean doCachedMatch(@NonNull ConditionContext conditionContext) {
Supplier<Boolean> evaluateConditionMatch = () -> {
registerApplicationListener(conditionContext);
return doMatch(conditionContext);
};
UnaryOperator<Boolean> clusterAvailableUpdateFunction = currentClusterAvailableState ->
ObjectUtils.initialize(currentClusterAvailableState, evaluateConditionMatch);
return clusterAvailable.updateAndGet(clusterAvailableUpdateFunction);
}
protected @NonNull ConditionContext registerApplicationListener(@NonNull ConditionContext conditionContext) {
Optional.ofNullable(conditionContext)
.map(ConditionContext::getResourceLoader)
.filter(ConfigurableApplicationContext.class::isInstance)
.map(ConfigurableApplicationContext.class::cast)
.ifPresent(applicationContext ->
applicationContext.addApplicationListener(clusterAwareConditionResetApplicationListener()));
.ifPresent(applicationContext -> applicationContext
.addApplicationListener(clusterAwareConditionResetOnContextClosedApplicationListener()));
return conditionContext;
}
boolean isMatch(@NonNull ConditionContext context) {
Environment environment = context.getEnvironment();
return isAvailable()
|| Boolean.TRUE.equals(environment.getProperty(SPRING_BOOT_DATA_GEMFIRE_CLUSTER_CONDITION_MATCH_PROPERTY,
Boolean.class, false));
}
/**
* Performs the actual conditional match to determine whether this Spring Boot for Apache Geode application
* can connect to an available Apache Geode cluster available in any environment (e.g. Standalone or Cloud).
*
* @param conditionContext Spring {@link ConditionContext} which captures the context in which the condition(s)
* @param conditionContext Spring {@link ConditionContext} capturing the context in which the condition(s)
* are evaluated; must not be {@literal null}.
* @return the given {@link ConditionContext}.
* @see org.springframework.context.annotation.ConditionContext
* @see #getConnectionEndpoints(Environment)
* @see #countConnections(ConnectionEndpointList)
* @see #configureTopology(Environment, ConnectionEndpointList, int)
* @see #isMatch(ConnectionEndpointList, int)
*/
@NonNull ConditionContext doMatch(@NonNull ConditionContext conditionContext) {
protected boolean doMatch(@NonNull ConditionContext conditionContext) {
Environment environment = conditionContext.getEnvironment();
ConnectionEndpointList connectionEndpoints =
new ConnectionEndpointList(getDefaultConnectionEndpoints())
.add(getConfiguredConnectionEndpoints(environment));
ConnectionEndpointList connectionEndpoints = getConnectionEndpoints(environment);
int connectionCount = countConnections(connectionEndpoints);
configureTopology(environment, connectionEndpoints, connectionCount);
clusterAvailable.set(isMatch(connectionEndpoints, connectionCount));
return conditionContext;
return isMatch(connectionEndpoints, connectionCount);
}
@NonNull Logger getLogger() {
boolean isMatch(@NonNull ConnectionEndpointList connectionEndpoints, int connectionCount) {
return connectionCount > 0;
}
protected @NonNull Logger getLogger() {
return logger;
}
List<ConnectionEndpoint> getDefaultConnectionEndpoints() {
protected ConnectionEndpointList getConnectionEndpoints(@NonNull Environment environment) {
return new ConnectionEndpointList(getDefaultConnectionEndpoints(environment))
.add(getConfiguredConnectionEndpoints(environment))
.add(getPooledConnectionEndpoints(environment));
}
protected List<ConnectionEndpoint> getDefaultConnectionEndpoints(@NonNull Environment environment) {
return Arrays.asList(
new ConnectionEndpoint(LOCALHOST, DEFAULT_CACHE_SERVER_PORT),
@@ -201,7 +258,7 @@ public class ClusterAwareConfiguration extends AbstractAnnotationConfigSupport {
);
}
List<ConnectionEndpoint> getConfiguredConnectionEndpoints(@NonNull Environment environment) {
protected List<ConnectionEndpoint> getConfiguredConnectionEndpoints(@NonNull Environment environment) {
List<ConnectionEndpoint> connectionEndpoints = new ArrayList<>();
@@ -252,55 +309,122 @@ public class ClusterAwareConfiguration extends AbstractAnnotationConfigSupport {
return connectionEndpoints;
}
int countConnections(@NonNull ConnectionEndpointList connectionEndpoints) {
protected List<ConnectionEndpoint> getPooledConnectionEndpoints(@NonNull Environment environment) {
List<ConnectionEndpoint> pooledConnectionEndpoints = new ArrayList<>();
getPoolsFromApacheGeode().stream()
.filter(Objects::nonNull)
.map(ConnectionEndpointListBuilder::from)
.forEach(pooledConnectionEndpoints::addAll);
return pooledConnectionEndpoints;
}
protected Collection<Pool> getPoolsFromApacheGeode() {
Set<Pool> pools = new HashSet<>();
pools.addAll(getPoolsFromClientCache());
pools.addAll(getPoolsFromPoolManager());
return pools;
}
// Technically, should be registered with the PoolManager, but...
Collection<Pool> getPoolsFromClientCache() {
return SimpleCacheResolver.getInstance().resolveClientCache()
.map(ClientCache::getDefaultPool)
.map(Collections::singleton)
.orElseGet(Collections::emptySet);
}
Collection<Pool> getPoolsFromPoolManager() {
Map<String, Pool> namedPools = PoolManager.getAll();
return CollectionUtils.nullSafeMap(namedPools).values().stream()
.filter(Objects::nonNull)
.collect(Collectors.toSet());
}
protected int countConnections(@NonNull ConnectionEndpointList connectionEndpoints) {
int count = 0;
for (ConnectionEndpoint connectionEndpoint : connectionEndpoints) {
Socket socket = null;
try (Socket socket = connect(connectionEndpoint)){
try {
socket = connect(connectionEndpoint);
count++;
count += isConnected(socket) ? 1 : 0;
if (getLogger().isInfoEnabled()) {
getLogger().info("Successfully connected to {}", connectionEndpoint);
}
}
catch (IOException cause) {
catch (IOException | SocketCreationException cause) {
if (getLogger().isInfoEnabled()) {
getLogger().info("Failed to connect to {}", connectionEndpoint);
}
if (getLogger().isDebugEnabled()) {
getLogger().debug("Connection failure caused by:", cause);
getLogger().debug("Connection failed because:", cause);
}
}
finally {
close(socket);
}
}
return count;
}
@NonNull Socket connect(@NonNull ConnectionEndpoint connectionEndpoint) throws IOException {
protected boolean isConnected(@NonNull Socket socket) {
return socket != null && socket.isConnected();
}
SocketAddress socketAddress =
new InetSocketAddress(connectionEndpoint.getHost(), connectionEndpoint.getPort());
protected @NonNull Socket connect(@NonNull ConnectionEndpoint connectionEndpoint) throws IOException {
Socket socket = new Socket();
SocketAddress socketAddress = connectionEndpoint.toInetSocketAddress();
Socket socket = connectionEndpoint instanceof PoolConnectionEndpoint
? newSocket((PoolConnectionEndpoint) connectionEndpoint)
: newSocket(connectionEndpoint);
socket.connect(socketAddress, DEFAULT_TIMEOUT_IN_MILLISECONDS);
return socket;
}
boolean close(@Nullable Socket socket) {
protected @NonNull Socket newSocket(@NonNull ConnectionEndpoint connectionEndpoint) throws IOException {
Socket socket = new Socket();
socket.setKeepAlive(false);
socket.setReuseAddress(true);
socket.setSoLinger(false, 0);
return socket;
}
protected @NonNull Socket newSocket(@NonNull PoolConnectionEndpoint poolConnectionEndpoint) {
Function<Throwable, Socket> ioExceptionHandlingFunction = cause -> {
String message = String.format("Failed to create Socket from PoolConnectionEndpoint [%s]",
poolConnectionEndpoint);
throw new SocketCreationException(message, cause);
};
return poolConnectionEndpoint.getPool()
.map(Pool::getSocketFactory)
.map(socketFactory -> ObjectUtils.<Socket>doOperationSafely(socketFactory::createSocket,
ioExceptionHandlingFunction))
.orElseGet(() -> ObjectUtils.<Socket>doOperationSafely(() ->
newSocket((ConnectionEndpoint) poolConnectionEndpoint), ioExceptionHandlingFunction));
}
protected boolean close(@Nullable Socket socket) {
return ObjectUtils.<Boolean>doOperationSafely(() -> {
@@ -314,7 +438,7 @@ public class ClusterAwareConfiguration extends AbstractAnnotationConfigSupport {
}, cause -> false);
}
void configureTopology(@NonNull Environment environment, @NonNull ConnectionEndpointList connectionEndpoints,
protected void configureTopology(@NonNull Environment environment, @NonNull ConnectionEndpointList connectionEndpoints,
int connectionCount) {
if (connectionCount < 1) {
@@ -324,34 +448,147 @@ public class ClusterAwareConfiguration extends AbstractAnnotationConfigSupport {
}
if (getLogger().isInfoEnabled()) {
getLogger().info("No cluster found; Spring Boot application will run in standalone (LOCAL) mode");
getLogger().info("No cluster found; Spring Boot application is running in standalone [LOCAL] mode");
}
}
else {
if (getLogger().isInfoEnabled()) {
getLogger().info("Cluster was found; Auto-configuration made [{}] successful connection(s);"
+ " Spring Boot application will run in a client/server topology", connectionCount);
+ " Spring Boot application is running in a client/server topology", connectionCount);
}
if (getLogger().isInfoEnabled()) {
if (CloudPlatform.CLOUD_FOUNDRY.isActive(environment)) {
getLogger().info("Spring Boot application is running in a client/server topology,"
+ " connected to VMware Tanzu GemFire for VMs");
+ " inside a VMware Tanzu GemFire for VMs environment");
}
else if (CloudPlatform.KUBERNETES.isActive(environment)) {
getLogger().info("Spring Boot application is running in a client/server topology,"
+ " connected to VMware Tanzu GemFire for K8S");
+ " inside a VMware Tanzu GemFire for K8S environment");
}
else {
getLogger().info("Spring Boot application is running in a client/server topology,"
+ " connected to a standalone Apache Geode-based cluster");
+ " using a standalone Apache Geode-based cluster");
}
}
}
}
}
boolean isMatch(@NonNull ConnectionEndpointList connectionEndpoints, int connectionCount) {
return connectionCount > 0;
protected static class ConnectionEndpointListBuilder {
protected static @NonNull ConnectionEndpointList from(@NonNull Pool pool) {
ConnectionEndpointList list = new ConnectionEndpointList();
if (pool != null) {
Set<InetSocketAddress> poolSocketAddresses = new HashSet<>();
collect(poolSocketAddresses, pool.getLocators());
collect(poolSocketAddresses, pool.getOnlineLocators());
collect(poolSocketAddresses, pool.getServers());
poolSocketAddresses.stream()
.map(ConnectionEndpoint::from)
.map(PoolConnectionEndpoint::from)
.map(it -> it.with(pool))
.forEach(list::add);
}
return list;
}
private static <T extends Collection<InetSocketAddress>> T collect(@NonNull T collection,
@NonNull Collection<InetSocketAddress> socketAddressesToCollect) {
CollectionUtils.nullSafeCollection(socketAddressesToCollect).stream()
.filter(Objects::nonNull)
.forEach(collection::add);
return collection;
}
}
protected static class PoolConnectionEndpoint extends ConnectionEndpoint {
protected static PoolConnectionEndpoint from(@NonNull ConnectionEndpoint connectionEndpoint) {
return new PoolConnectionEndpoint(connectionEndpoint.getHost(), connectionEndpoint.getPort());
}
private Pool pool;
PoolConnectionEndpoint(@NonNull String host, int port) {
super(host, port);
}
public Optional<Pool> getPool() {
return Optional.ofNullable(this.pool);
}
public @NonNull PoolConnectionEndpoint with(@Nullable Pool pool) {
this.pool = pool;
return this;
}
/**
* @inheritDoc
*/
@Override
public boolean equals(Object obj) {
if (obj == this) {
return true;
}
if (!(obj instanceof PoolConnectionEndpoint)) {
return false;
}
PoolConnectionEndpoint that = (PoolConnectionEndpoint) obj;
return super.equals(that)
&& this.getPool().equals(that.getPool());
}
/**
* @inheritDoc
*/
@Override
public int hashCode() {
int hashValue = super.hashCode();
hashValue = 37 * hashValue + ObjectUtils.nullSafeHashCode(getPool());
return hashValue;
}
/**
* @inheritDoc
*/
@Override
public String toString() {
return String.format("ConnectionEndpoint [%1$s] from Pool [%2$s]",
super.toString(), getPool().map(Pool::getName).orElse(""));
}
}
@SuppressWarnings("unused")
protected static class SocketCreationException extends RuntimeException {
protected SocketCreationException() { }
protected SocketCreationException(String message) {
super(message);
}
protected SocketCreationException(Throwable cause) {
super(cause);
}
protected SocketCreationException(String message, Throwable cause) {
super(message, cause);
}
}
}

6
spring-geode/src/main/java/org/springframework/geode/config/annotation/ClusterNotAvailableConfiguration.java
View File

@@ -148,8 +148,10 @@ public class ClusterNotAvailableConfiguration {
public static final class ClusterNotAvailableCondition extends ClusterAwareConfiguration.ClusterAwareCondition {
@Override
public synchronized boolean matches(@NonNull ConditionContext context, @NonNull AnnotatedTypeMetadata metadata) {
return !super.matches(context, metadata);
public synchronized boolean matches(@NonNull ConditionContext conditionContext,
@NonNull AnnotatedTypeMetadata metadata) {
return !super.matches(conditionContext, metadata);
}
}

565
spring-geode/src/test/java/org/springframework/geode/config/annotation/ClusterAwareConfigurationUnitTests.java
View File

@@ -20,25 +20,38 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isA;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.Properties;
import java.util.stream.Collectors;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentMatchers;
import org.mockito.InOrder;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.client.SocketFactory;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
@@ -50,6 +63,9 @@ import org.springframework.core.env.PropertiesPropertySource;
import org.springframework.data.gemfire.support.ConnectionEndpoint;
import org.springframework.data.gemfire.support.ConnectionEndpointList;
import org.springframework.data.gemfire.tests.integration.IntegrationTestsSupport;
import org.springframework.geode.config.annotation.ClusterAwareConfiguration.PoolConnectionEndpoint;
import org.springframework.geode.config.annotation.ClusterAwareConfiguration.SocketCreationException;
import org.springframework.lang.NonNull;
import org.slf4j.Logger;
@@ -57,19 +73,26 @@ import org.slf4j.Logger;
* Unit Tests for {@link EnableClusterAware} and {@link ClusterAwareConfiguration}.
*
* @author John Blum
* @see java.net.InetSocketAddress
* @see java.net.Socket
* @see java.util.Properties
* @see org.junit.Test
* @see org.mockito.Mockito
* @see org.mockito.Spy
* @see org.apache.geode.cache.client.Pool
* @see org.apache.geode.cache.client.SocketFactory
* @see org.springframework.context.ApplicationListener
* @see org.springframework.context.ConfigurableApplicationContext
* @see org.springframework.context.annotation.ConditionContext
* @see org.springframework.core.env.ConfigurableEnvironment
* @see org.springframework.core.env.Environment
* @see org.springframework.core.env.MutablePropertySources
* @see org.springframework.core.env.PropertiesPropertySource
* @see org.springframework.data.gemfire.support.ConnectionEndpoint
* @see org.springframework.data.gemfire.support.ConnectionEndpointList
* @see org.springframework.data.gemfire.tests.integration.IntegrationTestsSupport
* @see org.springframework.geode.config.annotation.ClusterAwareConfiguration
* @see org.springframework.geode.config.annotation.ClusterAwareConfiguration.PoolConnectionEndpoint
* @see org.springframework.geode.config.annotation.EnableClusterAware
* @since 1.2.0
*/
@@ -78,43 +101,200 @@ public class ClusterAwareConfigurationUnitTests extends IntegrationTestsSupport
private ClusterAwareConfiguration.ClusterAwareCondition condition =
spy(new ClusterAwareConfiguration.ClusterAwareCondition());
private @NonNull InetSocketAddress newSocketAddress(@NonNull String host, int port) {
return new InetSocketAddress(host, port);
}
@Before @After
public void setupAndTearDown() {
System.clearProperty(ClusterAwareConfiguration.SPRING_DATA_GEMFIRE_CACHE_CLIENT_REGION_SHORTCUT_PROPERTY);
ClusterAwareConfiguration.ClusterAwareCondition.reset();
}
@Test
public void matchRegistersApplicationListenerCallsDoMatch() {
public void matchesCallsIsMatchThenDoCachedMatchCorrectly() {
ConditionContext mockConditionContext = mock(ConditionContext.class);
doReturn(false).when(this.condition).isMatch(eq(mockConditionContext));
doReturn(true).when(this.condition).doCachedMatch(eq(mockConditionContext));
assertThat(this.condition.matches(mockConditionContext, null)).isTrue();
InOrder order = inOrder(this.condition);
order.verify(this.condition, times(1)).isMatch(eq(mockConditionContext));
order.verify(this.condition, times(1)).doCachedMatch(eq(mockConditionContext));
verifyNoInteractions(mockConditionContext);
}
@Test
public void matchesCallsIsMatchReturningTrueWillNotCallDoCachedMatchCorrectly() {
ConditionContext mockConditionContext = mock(ConditionContext.class);
doReturn(true).when(this.condition).isMatch(eq(mockConditionContext));
assertThat(this.condition.matches(mockConditionContext, null)).isTrue();
InOrder order = inOrder(this.condition);
order.verify(this.condition, times(1)).isMatch(eq(mockConditionContext));
order.verify(this.condition, never()).doCachedMatch(any());
verifyNoInteractions(mockConditionContext);
}
@Test
public void isMatchQueriesEnvironmentReturnsTrue() {
ConditionContext mockConditionContext = mock(ConditionContext.class);
Environment mockEnvironment = mock(Environment.class);
doReturn(mockEnvironment).when(mockConditionContext).getEnvironment();
doReturn(true).when(mockEnvironment)
.getProperty(eq(ClusterAwareConfiguration.SPRING_BOOT_DATA_GEMFIRE_CLUSTER_CONDITION_MATCH_PROPERTY),
eq(Boolean.class), eq(ClusterAwareConfiguration.DEFAULT_CLUSTER_CONDITION_MATCH));
assertThat(this.condition.isMatch(mockConditionContext)).isTrue();
verify(mockConditionContext, times(1)).getEnvironment();
verify(mockEnvironment, times(1))
.getProperty(eq(ClusterAwareConfiguration.SPRING_BOOT_DATA_GEMFIRE_CLUSTER_CONDITION_MATCH_PROPERTY),
eq(Boolean.class), eq(ClusterAwareConfiguration.DEFAULT_CLUSTER_CONDITION_MATCH));
verifyNoMoreInteractions(mockConditionContext, mockEnvironment);
}
@Test
public void isMatchIsNullSafe() {
assertThat(this.condition.isMatch(null)).isFalse();
}
@Test
public void doCachedMatchRegistersApplicationListenerCallsDoMatchAndCachesTheResultCorrectly() {
ConditionContext mockConditionContext = mock(ConditionContext.class);
doReturn(mockConditionContext).when(this.condition).registerApplicationListener(eq(mockConditionContext));
doReturn(true).when(this.condition).doMatch(eq(mockConditionContext));
assertThat(this.condition.doCachedMatch(mockConditionContext)).isTrue();
assertThat(this.condition.doCachedMatch(mockConditionContext)).isTrue();
InOrder order = inOrder(this.condition);
order.verify(this.condition, times(1)).registerApplicationListener(eq(mockConditionContext));
order.verify(this.condition, times(1)).doMatch(eq(mockConditionContext));
verifyNoInteractions(mockConditionContext);
}
@Test
public void registersApplicationListenerCorrectly() {
ConditionContext mockConditionContext = mock(ConditionContext.class);
ConfigurableApplicationContext mockApplicationContext = mock(ConfigurableApplicationContext.class);
Environment mockEnvironment = mock(Environment.class);
doReturn(mockApplicationContext).when(mockConditionContext).getResourceLoader();
when(mockConditionContext.getEnvironment()).thenReturn(mockEnvironment);
when(mockConditionContext.getResourceLoader()).thenReturn(mockApplicationContext);
ClusterAwareConfiguration.ClusterAwareCondition condition =
spy(new ClusterAwareConfiguration.ClusterAwareCondition());
assertThat(condition.matches(mockConditionContext, null)).isFalse();
assertThat(this.condition.registerApplicationListener(mockConditionContext)).isEqualTo(mockConditionContext);
verify(mockConditionContext, times(1)).getResourceLoader();
verify(mockApplicationContext, times(1))
.addApplicationListener(isA(ApplicationListener.class));
verify(condition, times(1)).doMatch(eq(mockConditionContext));
verifyNoMoreInteractions(mockConditionContext, mockApplicationContext);
}
@Test
public void doMatchCallsGetConnectionEndpointsThenCountConnectionsThenConfigureTopologyAndIsMatchCorrectly() {
ConditionContext mockConditionContext = mock(ConditionContext.class);
ConnectionEndpointList mockConnectionEndpointList = mock(ConnectionEndpointList.class);
Environment mockEnvironment = mock(Environment.class);
doReturn(mockEnvironment).when(mockConditionContext).getEnvironment();
doReturn(mockConnectionEndpointList).when(this.condition).getConnectionEndpoints(eq(mockEnvironment));
doReturn(1).when(this.condition).countConnections(eq(mockConnectionEndpointList));
assertThat(this.condition.doMatch(mockConditionContext)).isTrue();
InOrder order = inOrder(this.condition, mockConditionContext, mockConnectionEndpointList, mockEnvironment);
order.verify(this.condition, times(1)).doMatch(eq(mockConditionContext));
order.verify(mockConditionContext, times(1)).getEnvironment();
order.verify(this.condition, times(1)).getConnectionEndpoints(eq(mockEnvironment));
order.verify(this.condition, times(1)).countConnections(eq(mockConnectionEndpointList));
order.verify(this.condition, times(1))
.configureTopology(eq(mockEnvironment), eq(mockConnectionEndpointList), eq(1));
order.verify(this.condition, times(1)).isMatch(eq(mockConnectionEndpointList), eq(1));
verify(this.condition, atLeastOnce()).getLogger();
verifyNoMoreInteractions(this.condition, mockConditionContext);
verifyNoInteractions(mockConnectionEndpointList, mockEnvironment);
}
@Test
public void isMatchReturnsTrue() {
assertThat(this.condition.isMatch(null, 1)).isTrue();
}
@Test
public void isNotMatchReturnsFalse() {
ConnectionEndpointList connectionEndpoints =
ConnectionEndpointList.from(new ConnectionEndpoint("localhost", 1234));
assertThat(this.condition.isMatch(connectionEndpoints, 0)).isFalse();
}
@Test
public void getConnectionEndpointsCollectsDefaultsConfiguredAndPooledConnections() {
ConnectionEndpoint defaultConnectionEndpoint = ConnectionEndpoint.parse("default[1234]");
ConnectionEndpoint configuredConnectionEndpoint = ConnectionEndpoint.parse("configured[5678]");
ConnectionEndpoint pooledConnectionEndpoint = ConnectionEndpoint.parse("pooled[9012]");
Environment mockEnvironment = mock(Environment.class);
doReturn(Collections.singletonList(defaultConnectionEndpoint))
.when(this.condition).getDefaultConnectionEndpoints(eq(mockEnvironment));
doReturn(Collections.singletonList(configuredConnectionEndpoint))
.when(this.condition).getConfiguredConnectionEndpoints(eq(mockEnvironment));
doReturn(Collections.singletonList(pooledConnectionEndpoint))
.when(this.condition).getPooledConnectionEndpoints(eq(mockEnvironment));
assertThat(this.condition.getConnectionEndpoints(mockEnvironment))
.containsExactly(defaultConnectionEndpoint, configuredConnectionEndpoint, pooledConnectionEndpoint);
verify(this.condition, times(1)).getDefaultConnectionEndpoints(eq(mockEnvironment));
verify(this.condition, times(1)).getConfiguredConnectionEndpoints(eq(mockEnvironment));
verify(this.condition, times(1)).getPooledConnectionEndpoints(eq(mockEnvironment));
verifyNoInteractions(mockEnvironment);
}
@Test
public void getDefaultConnectionEndpointsIncludesDefaultLocatorAndDefaultServer() {
assertThat(this.condition.getDefaultConnectionEndpoints().stream()
Environment mockEnvironment = mock(Environment.class);
assertThat(this.condition.getDefaultConnectionEndpoints(mockEnvironment).stream()
.map(ConnectionEndpoint::toString)
.collect(Collectors.toList()))
.containsExactly("localhost[40404]", "localhost[10334]");
verifyNoInteractions(mockEnvironment);
}
@Test
@@ -127,9 +307,11 @@ public class ClusterAwareConfigurationUnitTests extends IntegrationTestsSupport
Properties locatorProperties = new Properties();
Properties serverProperties = new Properties();
locatorProperties.setProperty("spring.data.gemfire.other-property", "junk");
locatorProperties.setProperty("spring.data.gemfire.pool.locators", "boombox[1234], cardboardbox[5678], mailbox[9012]");
locatorProperties.setProperty("spring.data.gemfire.pool.car.locators", "skullbox[11235]");
locatorProperties.setProperty("spring.data.gemfire.pool.other-property", "junk");
serverProperties.setProperty("spring.data.gemfire.other-property", "junk");
serverProperties.setProperty("spring.data.gemfire.pool.servers", "mars[41414]");
serverProperties.setProperty("spring.data.gemfire.pool.swimming.servers", "jupiter[42424], saturn[43434]");
serverProperties.setProperty("spring.data.gemfire.pool.other-property", "junk");
@@ -162,6 +344,86 @@ public class ClusterAwareConfigurationUnitTests extends IntegrationTestsSupport
verify(mockEnvironment, times(4)).getProperty(anyString());
}
@Test
public void getPooledConnectionEndpointsIsCorrect() {
Pool mockPoolOne = mock(Pool.class);
Pool mockPoolTwo = mock(Pool.class);
doReturn(Arrays.asList(mockPoolOne, null, mockPoolTwo, null, null))
.when(this.condition).getPoolsFromApacheGeode();
List<InetSocketAddress> locatorSocketAddressesForPoolOne =
Arrays.asList(null, newSocketAddress("boombox", 1234),
newSocketAddress("skullbox", 56789), null);
doReturn(locatorSocketAddressesForPoolOne).when(mockPoolOne).getLocators();
List<InetSocketAddress> onlineLocatorsSocketAddressesForPoolOne =
Arrays.asList(newSocketAddress("cardboardbox", 9012), null);
doReturn(onlineLocatorsSocketAddressesForPoolOne).when(mockPoolOne).getOnlineLocators();
List<InetSocketAddress> serverSocketAddressesForPoolOne =
Arrays.asList(newSocketAddress("mailbox", 10334), newSocketAddress("pobox", 4321));
doReturn(serverSocketAddressesForPoolOne).when(mockPoolOne).getServers();
List<InetSocketAddress> locatorSocketAddressesForPoolTwo =
Arrays.asList(null, newSocketAddress("mars", 1234), null, null);
doReturn(locatorSocketAddressesForPoolTwo).when(mockPoolTwo).getLocators();
doReturn(null).when(mockPoolTwo).getOnlineLocators();
List<InetSocketAddress> serverSocketAddressesForPoolTwo =
Arrays.asList(newSocketAddress("saturn", 1234), newSocketAddress("neptune", 56789));
doReturn(serverSocketAddressesForPoolTwo).when(mockPoolTwo).getServers();
Environment mockEnvironment = mock(Environment.class);
List<ConnectionEndpoint> pooledConnectionEndpoints =
this.condition.getPooledConnectionEndpoints(mockEnvironment);
assertThat(pooledConnectionEndpoints).isNotNull();
assertThat(pooledConnectionEndpoints).hasSize(8);
assertThat(pooledConnectionEndpoints).containsExactlyInAnyOrder(
new PoolConnectionEndpoint("boombox", 1234).with(mockPoolOne),
new PoolConnectionEndpoint("skullbox", 56789).with(mockPoolOne),
new PoolConnectionEndpoint("cardboardbox", 9012).with(mockPoolOne),
new PoolConnectionEndpoint("mailbox", 10334).with(mockPoolOne),
new PoolConnectionEndpoint("pobox", 4321).with(mockPoolOne),
new PoolConnectionEndpoint("mars", 1234).with(mockPoolTwo),
new PoolConnectionEndpoint("saturn", 1234).with(mockPoolTwo),
new PoolConnectionEndpoint("neptune", 56789).with(mockPoolTwo)
);
verify(this.condition, times(1)).getPooledConnectionEndpoints(eq(mockEnvironment));
verify(this.condition, times(1)).getPoolsFromApacheGeode();
verifyNoMoreInteractions(this.condition);
verifyNoInteractions(mockEnvironment);
}
@Test
public void getPoolsFromApacheGeodeIsCorrect() {
Pool mockPoolOne = mock(Pool.class);
Pool mockPoolTwo = mock(Pool.class);
doReturn(Collections.singleton(mockPoolOne)).when(this.condition).getPoolsFromClientCache();
doReturn(Collections.singletonList(mockPoolTwo)).when(this.condition).getPoolsFromPoolManager();
assertThat(this.condition.getPoolsFromApacheGeode()).containsExactlyInAnyOrder(mockPoolOne, mockPoolTwo);
verify(this.condition, times(1)).getPoolsFromApacheGeode();
verify(this.condition, times(1)).getPoolsFromClientCache();
verify(this.condition, times(1)).getPoolsFromPoolManager();
verifyNoMoreInteractions(this.condition);
}
@Test
public void countConnectionsIsCorrect() throws Exception {
@@ -169,6 +431,7 @@ public class ClusterAwareConfigurationUnitTests extends IntegrationTestsSupport
new ConnectionEndpoint("boombox", 1234),
new ConnectionEndpoint("cardboardbox", 5678),
new ConnectionEndpoint("mailbox", 9012),
new ConnectionEndpoint("pobox", 40404),
new ConnectionEndpoint("skullbox", 10334)
);
@@ -180,7 +443,12 @@ public class ClusterAwareConfigurationUnitTests extends IntegrationTestsSupport
ConnectionEndpoint connectionEndpoint = invocation.getArgument(0);
if (Arrays.asList("boombox", "skullbox").contains(connectionEndpoint.getHost())) {
if (Arrays.asList("mailbox", "pobox").contains(connectionEndpoint.getHost())) {
Socket mockSocket = mock(Socket.class);
doReturn(true).when(mockSocket).isConnected();
return mockSocket;
}
else if (Arrays.asList("boombox", "skullbox").contains(connectionEndpoint.getHost())) {
return mock(Socket.class);
}
else {
@@ -192,6 +460,263 @@ public class ClusterAwareConfigurationUnitTests extends IntegrationTestsSupport
assertThat(this.condition.countConnections(list)).isEqualTo(2);
}
@Test
public void isConnectedSocketReturnsTrue() {
Socket mockSocket = mock(Socket.class);
doReturn(true).when(mockSocket).isConnected();
assertThat(this.condition.isConnected(mockSocket)).isTrue();
verify(mockSocket, times(1)).isConnected();
verifyNoMoreInteractions(mockSocket);
}
@Test
public void isConnectedSocketReturnsFalse() {
Socket mockSocket = mock(Socket.class);
doReturn(false).when(mockSocket).isConnected();
assertThat(this.condition.isConnected(mockSocket)).isFalse();
verify(mockSocket, times(1)).isConnected();
verifyNoMoreInteractions(mockSocket);
}
@Test
public void isConnectedWithNullIsNullSafe() {
assertThat(this.condition.isConnected(null)).isFalse();
}
@Test
public void connectNonPooledSocket() throws IOException {
ConnectionEndpoint mockConnectionEndpoint = mock(ConnectionEndpoint.class);
InetSocketAddress mockSocketAddress = mock(InetSocketAddress.class);
Socket mockSocket = mock(Socket.class);
doReturn(mockSocketAddress).when(mockConnectionEndpoint).toInetSocketAddress();
doReturn(mockSocket).when(this.condition).newSocket(isA(ConnectionEndpoint.class));
assertThat(this.condition.connect(mockConnectionEndpoint)).isEqualTo(mockSocket);
verify(mockConnectionEndpoint).toInetSocketAddress();
verify(this.condition, times(1)).newSocket(eq(mockConnectionEndpoint));
verify(mockSocket, times(1))
.connect(eq(mockSocketAddress), eq(ClusterAwareConfiguration.DEFAULT_TIMEOUT_IN_MILLISECONDS));
verifyNoMoreInteractions(mockConnectionEndpoint, mockSocket);
verifyNoInteractions(mockSocketAddress);
}
@Test
@SuppressWarnings("all")
public void connectPooledSocket() throws IOException {
ConnectionEndpoint mockConnectionEndpoint = mock(PoolConnectionEndpoint.class);
InetSocketAddress mockSocketAddress = mock(InetSocketAddress.class);
Socket mockSocket = mock(Socket.class);
doReturn(mockSocketAddress).when(mockConnectionEndpoint).toInetSocketAddress();
doReturn(mockSocket).when(this.condition).newSocket(isA(PoolConnectionEndpoint.class));
//doNothing().when(mockSocket).connect(isA(InetSocketAddress.class), anyInt());
assertThat(this.condition.connect(mockConnectionEndpoint)).isEqualTo(mockSocket);
verify(mockConnectionEndpoint, times(1)).toInetSocketAddress();
verify(this.condition, times(1)).newSocket(eq((PoolConnectionEndpoint) mockConnectionEndpoint));
verify(mockSocket, times(1))
.connect(isA(InetSocketAddress.class), eq(ClusterAwareConfiguration.DEFAULT_TIMEOUT_IN_MILLISECONDS));
verifyNoMoreInteractions(mockConnectionEndpoint, mockSocket);
verifyNoInteractions(mockSocketAddress);
}
@Test
public void newSocketFromConnectionEndpoint() throws IOException {
ConnectionEndpoint mockConnectionEndpoint = mock(ConnectionEndpoint.class);
Socket socket = this.condition.newSocket(mockConnectionEndpoint);
assertThat(socket).isNotNull();
assertThat(socket.getKeepAlive()).isFalse();
assertThat(socket.getReuseAddress()).isTrue();
assertThat(socket.getSoLinger()).isEqualTo(-1);
verifyNoInteractions(mockConnectionEndpoint);
}
@Test
public void newSocketFromPoolConnectionEndpoint() throws IOException {
Pool mockPool = mock(Pool.class);
PoolConnectionEndpoint mockPoolConnectionEndpoint = mock(PoolConnectionEndpoint.class);
Socket mockSocket = mock(Socket.class);
SocketFactory mockSocketFactory = mock(SocketFactory.class);
doReturn(Optional.of(mockPool)).when(mockPoolConnectionEndpoint).getPool();
doReturn(mockSocketFactory).when(mockPool).getSocketFactory();
doReturn(mockSocket).when(mockSocketFactory).createSocket();
assertThat(this.condition.newSocket(mockPoolConnectionEndpoint)).isEqualTo(mockSocket);
verify(mockPoolConnectionEndpoint, times(1)).getPool();
verify(mockPool, times(1)).getSocketFactory();
verify(mockSocketFactory, times(1)).createSocket();
verifyNoMoreInteractions(mockPool, mockPoolConnectionEndpoint, mockSocketFactory);
verifyNoInteractions(mockSocket);
}
@Test(expected = SocketCreationException.class)
public void newSocketFromPoolConnectionEndpointHandlesIOException() throws IOException {
Pool mockPool = mock(Pool.class);
PoolConnectionEndpoint mockPoolConnectionEndpoint = mock(PoolConnectionEndpoint.class);
SocketFactory mockSocketFactory = mock(SocketFactory.class);
doReturn(Optional.of(mockPool)).when(mockPoolConnectionEndpoint).getPool();
doReturn(mockSocketFactory).when(mockPool).getSocketFactory();
doThrow(new IOException("TEST")).when(mockSocketFactory).createSocket();
try {
this.condition.newSocket(mockPoolConnectionEndpoint);
}
catch (SocketCreationException expected) {
assertThat(expected).hasMessage("Failed to create Socket from PoolConnectionEndpoint [%s]",
mockPoolConnectionEndpoint);
assertThat(expected).hasCauseInstanceOf(IOException.class);
assertThat(expected.getCause()).hasMessage("TEST");
assertThat(expected.getCause()).hasNoCause();
throw expected;
}
finally {
verify(mockPoolConnectionEndpoint, times(1)).getPool();
verify(mockPool, times(1)).getSocketFactory();
verify(mockSocketFactory, times(1)).createSocket();
verify(this.condition, never()).newSocket(isA(ConnectionEndpoint.class));
verifyNoMoreInteractions(mockPool, mockPoolConnectionEndpoint, mockSocketFactory);
}
}
@Test
public void newSocketFromPoolConnectionEndpointReturnsNonPooledSocket() throws IOException {
Pool mockPool = mock(Pool.class);
PoolConnectionEndpoint mockPoolConnectionEndpoint = mock(PoolConnectionEndpoint.class);
Socket mockSocket = mock(Socket.class);
SocketFactory mockSocketFactory = mock(SocketFactory.class);
doReturn(Optional.of(mockPool)).when(mockPoolConnectionEndpoint).getPool();
doReturn(mockSocketFactory).when(mockPool).getSocketFactory();
doReturn(null).when(mockSocketFactory).createSocket();
doReturn(mockSocket).when(this.condition).newSocket(ArgumentMatchers.<ConnectionEndpoint>any());
assertThat(this.condition.newSocket(mockPoolConnectionEndpoint)).isEqualTo(mockSocket);
verify(mockPoolConnectionEndpoint, times(1)).getPool();
verify(mockPool, times(1)).getSocketFactory();
verify(mockSocketFactory, times(1)).createSocket();
verify(this.condition, times(1))
.newSocket(eq((ConnectionEndpoint) mockPoolConnectionEndpoint));
verifyNoMoreInteractions(mockPool, mockPoolConnectionEndpoint, mockSocketFactory);
verifyNoInteractions(mockSocket);
}
@Test(expected = SocketCreationException.class)
public void newSocketFromPoolConnectionEndpointReturningNonPooledSocketHandlesIOException() throws IOException {
Pool mockPool = mock(Pool.class);
PoolConnectionEndpoint mockPoolConnectionEndpoint = mock(PoolConnectionEndpoint.class);
SocketFactory mockSocketFactory = mock(SocketFactory.class);
doReturn(Optional.of(mockPool)).when(mockPoolConnectionEndpoint).getPool();
doReturn(mockSocketFactory).when(mockPool).getSocketFactory();
doReturn(null).when(mockSocketFactory).createSocket();
doThrow(new IOException("TEST")).when(this.condition).newSocket(ArgumentMatchers.<ConnectionEndpoint>any());
try {
this.condition.newSocket(mockPoolConnectionEndpoint);
}
catch (SocketCreationException expected) {
assertThat(expected).hasMessage("Failed to create Socket from PoolConnectionEndpoint [%s]",
mockPoolConnectionEndpoint);
assertThat(expected).hasCauseInstanceOf(IOException.class);
assertThat(expected.getCause()).hasMessage("TEST");
assertThat(expected.getCause()).hasNoCause();
throw expected;
}
finally {
verify(mockPoolConnectionEndpoint, times(1)).getPool();
verify(mockPool, times(1)).getSocketFactory();
verify(mockSocketFactory, times(1)).createSocket();
verify(this.condition, times(1))
.newSocket(eq((ConnectionEndpoint) mockPoolConnectionEndpoint));
verifyNoMoreInteractions(mockPool, mockPoolConnectionEndpoint, mockSocketFactory);
}
}
@Test
public void closeSocketReturnsTrue() throws IOException {
Socket mockSocket = mock(Socket.class);
assertThat(this.condition.close(mockSocket)).isTrue();
verify(mockSocket, times(1)).close();
verifyNoMoreInteractions(mockSocket);
}
@Test
public void closeSocketReturnsFalse() throws IOException {
Socket mockSocket = mock(Socket.class);
doThrow(new IOException("TEST")).when(mockSocket).close();
assertThat(this.condition.close(mockSocket)).isFalse();
verify(mockSocket, times(1)).close();
verifyNoMoreInteractions(mockSocket);
}
@Test
public void closeNullSocketIsNullSafe() {
assertThat(this.condition.close(null)).isFalse();
}
@Test
public void configureTopologySetsLocalWhenConnectionCountIsLessThanOneAndEnvironmentDoesNotContainTargetProperty() {
@@ -248,18 +773,4 @@ public class ClusterAwareConfigurationUnitTests extends IntegrationTestsSupport
verify(mockEnvironment, times(1))
.containsProperty(eq(ClusterAwareConfiguration.SPRING_DATA_GEMFIRE_CACHE_CLIENT_REGION_SHORTCUT_PROPERTY));
}
@Test
public void isMatchReturnsTrue() {
assertThat(this.condition.isMatch(null, 1)).isTrue();
}
@Test
public void isNotMatchReturnsFalse() {
ConnectionEndpointList connectionEndpoints =
ConnectionEndpointList.from(new ConnectionEndpoint("localhost", 1234));
assertThat(this.condition.isMatch(connectionEndpoints, 0)).isFalse();
}
}