From 4a20f7be61fc3828ea9cb088129b9e79c8bc10b8 Mon Sep 17 00:00:00 2001 From: John Blum Date: Fri, 15 Jan 2021 01:32:11 -0800 Subject: [PATCH] Support awareness and discovery of secure (Auth and SSL eanbled) Apache Geode clusters using @EnableClusterAware annotation functionality. Improves log details about the runtime environment (e.g. cloud, standalone/self-managed). Resolves gh-57. Resolves gh-99. --- ...terAwareConfigurationIntegrationTests.java | 224 +++++++ .../annotation/ClusterAwareConfiguration.java | 349 +++++++++-- .../ClusterNotAvailableConfiguration.java | 6 +- .../ClusterAwareConfigurationUnitTests.java | 565 +++++++++++++++++- 4 files changed, 1059 insertions(+), 85 deletions(-) create mode 100644 spring-geode-autoconfigure/src/test/java/org/springframework/geode/boot/autoconfigure/cluster/aware/SecureClusterAwareConfigurationIntegrationTests.java diff --git a/spring-geode-autoconfigure/src/test/java/org/springframework/geode/boot/autoconfigure/cluster/aware/SecureClusterAwareConfigurationIntegrationTests.java b/spring-geode-autoconfigure/src/test/java/org/springframework/geode/boot/autoconfigure/cluster/aware/SecureClusterAwareConfigurationIntegrationTests.java new file mode 100644 index 00000000..e056b255 --- /dev/null +++ b/spring-geode-autoconfigure/src/test/java/org/springframework/geode/boot/autoconfigure/cluster/aware/SecureClusterAwareConfigurationIntegrationTests.java @@ -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 regionNames = cache.rootRegions().stream() + .map(Region::getName) + .collect(Collectors.toList()); + + assertThat(regionNames) + .describedAs("Expected no Regions; but was [%s]", regionNames) + .isEmpty(); + }; + } + } +} diff --git a/spring-geode/src/main/java/org/springframework/geode/config/annotation/ClusterAwareConfiguration.java b/spring-geode/src/main/java/org/springframework/geode/config/annotation/ClusterAwareConfiguration.java index 61465769..38e3cede 100644 --- a/spring-geode/src/main/java/org/springframework/geode/config/annotation/ClusterAwareConfiguration.java +++ b/spring-geode/src/main/java/org/springframework/geode/config/annotation/ClusterAwareConfiguration.java @@ -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 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 getAnnotationType() { return EnableClusterAware.class; @@ -116,7 +145,7 @@ public class ClusterAwareConfiguration extends AbstractAnnotationConfigSupport { private static final AtomicReference clusterAvailable = new AtomicReference<>(null); - private static ApplicationListener clusterAwareConditionResetApplicationListener() { + private static ApplicationListener 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 evaluateConditionMatch = () -> { + registerApplicationListener(conditionContext); + return doMatch(conditionContext); + }; + + UnaryOperator 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 getDefaultConnectionEndpoints() { + protected ConnectionEndpointList getConnectionEndpoints(@NonNull Environment environment) { + + return new ConnectionEndpointList(getDefaultConnectionEndpoints(environment)) + .add(getConfiguredConnectionEndpoints(environment)) + .add(getPooledConnectionEndpoints(environment)); + } + + protected List getDefaultConnectionEndpoints(@NonNull Environment environment) { return Arrays.asList( new ConnectionEndpoint(LOCALHOST, DEFAULT_CACHE_SERVER_PORT), @@ -201,7 +258,7 @@ public class ClusterAwareConfiguration extends AbstractAnnotationConfigSupport { ); } - List getConfiguredConnectionEndpoints(@NonNull Environment environment) { + protected List getConfiguredConnectionEndpoints(@NonNull Environment environment) { List connectionEndpoints = new ArrayList<>(); @@ -252,55 +309,122 @@ public class ClusterAwareConfiguration extends AbstractAnnotationConfigSupport { return connectionEndpoints; } - int countConnections(@NonNull ConnectionEndpointList connectionEndpoints) { + protected List getPooledConnectionEndpoints(@NonNull Environment environment) { + + List pooledConnectionEndpoints = new ArrayList<>(); + + getPoolsFromApacheGeode().stream() + .filter(Objects::nonNull) + .map(ConnectionEndpointListBuilder::from) + .forEach(pooledConnectionEndpoints::addAll); + + return pooledConnectionEndpoints; + } + + protected Collection getPoolsFromApacheGeode() { + + Set pools = new HashSet<>(); + + pools.addAll(getPoolsFromClientCache()); + pools.addAll(getPoolsFromPoolManager()); + + return pools; + } + + // Technically, should be registered with the PoolManager, but... + Collection getPoolsFromClientCache() { + + return SimpleCacheResolver.getInstance().resolveClientCache() + .map(ClientCache::getDefaultPool) + .map(Collections::singleton) + .orElseGet(Collections::emptySet); + } + + Collection getPoolsFromPoolManager() { + + Map 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 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.doOperationSafely(socketFactory::createSocket, + ioExceptionHandlingFunction)) + .orElseGet(() -> ObjectUtils.doOperationSafely(() -> + newSocket((ConnectionEndpoint) poolConnectionEndpoint), ioExceptionHandlingFunction)); + } + + protected boolean close(@Nullable Socket socket) { return ObjectUtils.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 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 collect(@NonNull T collection, + @NonNull Collection 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 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); } } } diff --git a/spring-geode/src/main/java/org/springframework/geode/config/annotation/ClusterNotAvailableConfiguration.java b/spring-geode/src/main/java/org/springframework/geode/config/annotation/ClusterNotAvailableConfiguration.java index 02611657..685bd327 100644 --- a/spring-geode/src/main/java/org/springframework/geode/config/annotation/ClusterNotAvailableConfiguration.java +++ b/spring-geode/src/main/java/org/springframework/geode/config/annotation/ClusterNotAvailableConfiguration.java @@ -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); } } diff --git a/spring-geode/src/test/java/org/springframework/geode/config/annotation/ClusterAwareConfigurationUnitTests.java b/spring-geode/src/test/java/org/springframework/geode/config/annotation/ClusterAwareConfigurationUnitTests.java index f9255afd..93b7c4cd 100644 --- a/spring-geode/src/test/java/org/springframework/geode/config/annotation/ClusterAwareConfigurationUnitTests.java +++ b/spring-geode/src/test/java/org/springframework/geode/config/annotation/ClusterAwareConfigurationUnitTests.java @@ -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 locatorSocketAddressesForPoolOne = + Arrays.asList(null, newSocketAddress("boombox", 1234), + newSocketAddress("skullbox", 56789), null); + + doReturn(locatorSocketAddressesForPoolOne).when(mockPoolOne).getLocators(); + + List onlineLocatorsSocketAddressesForPoolOne = + Arrays.asList(newSocketAddress("cardboardbox", 9012), null); + + doReturn(onlineLocatorsSocketAddressesForPoolOne).when(mockPoolOne).getOnlineLocators(); + + List serverSocketAddressesForPoolOne = + Arrays.asList(newSocketAddress("mailbox", 10334), newSocketAddress("pobox", 4321)); + + doReturn(serverSocketAddressesForPoolOne).when(mockPoolOne).getServers(); + + List locatorSocketAddressesForPoolTwo = + Arrays.asList(null, newSocketAddress("mars", 1234), null, null); + + doReturn(locatorSocketAddressesForPoolTwo).when(mockPoolTwo).getLocators(); + doReturn(null).when(mockPoolTwo).getOnlineLocators(); + + List serverSocketAddressesForPoolTwo = + Arrays.asList(newSocketAddress("saturn", 1234), newSocketAddress("neptune", 56789)); + + doReturn(serverSocketAddressesForPoolTwo).when(mockPoolTwo).getServers(); + + Environment mockEnvironment = mock(Environment.class); + + List 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.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.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(); - } }