From 8b1d862fb5265c237d254c5b6148b180ae5dcb8a Mon Sep 17 00:00:00 2001 From: John Blum Date: Fri, 17 Apr 2020 12:42:24 -0700 Subject: [PATCH] Improve Multi-Site (WAN) Caching Sample Integration Tests resiliency. Include an additional Spring @Configuration class exclusively for testing that registers a Geode CacheListener to wait for the cache Region EntryEvent to occur. When the ClientCache Registers Interest (all keys) and receives the cache Region (CustomersByName) EntryEvent, then it signifies the WAN Gateways (Sender and Receiver) has replicated the Region EntryEvent between disparate sites. Client 1 creates the Region EntryEvent that Client 2 expects to see in its cache. Resolves gh-80. --- .../MultiSiteCachingIntegrationTests.java | 99 ++++++++++++++++++- 1 file changed, 98 insertions(+), 1 deletion(-) diff --git a/spring-geode-samples/caching/multi-site/src/test/java/example/app/caching/multisite/MultiSiteCachingIntegrationTests.java b/spring-geode-samples/caching/multi-site/src/test/java/example/app/caching/multisite/MultiSiteCachingIntegrationTests.java index 7ba88ab6..dc4c9c7f 100644 --- a/spring-geode-samples/caching/multi-site/src/test/java/example/app/caching/multisite/MultiSiteCachingIntegrationTests.java +++ b/spring-geode-samples/caching/multi-site/src/test/java/example/app/caching/multisite/MultiSiteCachingIntegrationTests.java @@ -19,18 +19,36 @@ import static org.assertj.core.api.Assertions.assertThat; import java.io.IOException; import java.time.Duration; +import java.util.Collections; import java.util.Optional; import java.util.Properties; +import java.util.Set; import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; +import org.apache.geode.cache.CacheListener; +import org.apache.geode.cache.EntryEvent; +import org.apache.geode.cache.InterestResultPolicy; +import org.apache.geode.cache.util.CacheListenerAdapter; + +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.SpringApplication; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.gemfire.client.ClientRegionFactoryBean; +import org.springframework.data.gemfire.client.Interest; +import org.springframework.data.gemfire.client.RegexInterest; +import org.springframework.data.gemfire.config.annotation.ClientCacheConfigurer; +import org.springframework.data.gemfire.config.annotation.RegionConfigurer; import org.springframework.data.gemfire.tests.integration.ForkingClientServerIntegrationTestsSupport; import org.springframework.data.gemfire.tests.process.ProcessWrapper; +import org.springframework.data.gemfire.util.ArrayUtils; +import org.springframework.data.gemfire.util.CollectionUtils; +import org.springframework.lang.NonNull; import example.app.caching.multisite.client.BootGeodeMultiSiteCachingClientApplication; import example.app.caching.multisite.client.model.Customer; @@ -64,6 +82,7 @@ public class MultiSiteCachingIntegrationTests extends ForkingClientServerIntegra private static ProcessWrapper geodeClusterTwo; private static final String HOSTNAME = "localhost"; + private static final String TEST_CUSTOMER_NAME = "Jon Doe"; @BeforeClass public static void startGeodeClusters() throws IOException { @@ -95,6 +114,11 @@ public class MultiSiteCachingIntegrationTests extends ForkingClientServerIntegra } private ConfigurableApplicationContext newApplicationContext(int locatorPort, String... springProfiles) { + return newApplicationContext(locatorPort, Collections.emptySet(), springProfiles); + } + + private ConfigurableApplicationContext newApplicationContext(int locatorPort, Set> sources, + String... springProfiles) { Properties configuration = new Properties(); @@ -109,6 +133,7 @@ public class MultiSiteCachingIntegrationTests extends ForkingClientServerIntegra .headless(true) .profiles(springProfiles) //.properties(configuration) + .sources(CollectionUtils.nullSafeSet(sources).toArray(new Class[0])) .registerShutdownHook(true) .build(); @@ -135,6 +160,7 @@ public class MultiSiteCachingIntegrationTests extends ForkingClientServerIntegra assertThat(applicationContext).isNotNull(); assertThat(applicationContext.isActive()).isTrue(); + assertThat(applicationContext.isRunning()).isTrue(); CustomerService customerService = applicationContext.getBean(CustomerService.class); @@ -150,10 +176,19 @@ public class MultiSiteCachingIntegrationTests extends ForkingClientServerIntegra close(applicationContext); - applicationContext = newApplicationContext(locatorPortClusterTwo, "client-site2"); + applicationContext = newApplicationContext(locatorPortClusterTwo, + Collections.singleton(TestGeodeClientConfiguration.class), "client-site2"); assertThat(applicationContext).isNotNull(); assertThat(applicationContext.isActive()).isTrue(); + assertThat(applicationContext.isRunning()).isTrue(); + + CustomersByNameCacheListener customersByNameCacheListener = + applicationContext.getBean("customersByNameCacheListener", CustomersByNameCacheListener.class); + + assertThat(customersByNameCacheListener).isNotNull(); + + ThreadUtils.waitFor(Duration.ofSeconds(10), 500L, customersByNameCacheListener::isEntryEventArrived); customerService = applicationContext.getBean(CustomerService.class); customerService.setSleepInSeconds(2L); @@ -169,4 +204,66 @@ public class MultiSiteCachingIntegrationTests extends ForkingClientServerIntegra close(applicationContext); } } + + @SuppressWarnings("unused") + static class CustomersByNameCacheListener extends CacheListenerAdapter { + + private boolean entryEventArrived = false; + + @Override + public synchronized void afterCreate(EntryEvent event) { + + this.entryEventArrived |= Optional.ofNullable(event) + //.map(this::log) + .map(EntryEvent::getKey) + .filter(TEST_CUSTOMER_NAME::equals) + .isPresent(); + } + + synchronized boolean isEntryEventArrived() { + return this.entryEventArrived; + } + + @NonNull EntryEvent log(@NonNull EntryEvent entryEvent) { + + System.err.printf("EntryEvent with key [%s] and value [%s] arrived for Region [%s]%n", + entryEvent.getKey(), entryEvent.getNewValue(), entryEvent.getRegion().getName()); + + return entryEvent; + } + } + + @Configuration + @SuppressWarnings("unused") + static class TestGeodeClientConfiguration { + + @Bean + ClientCacheConfigurer clientCacheSubscriptionsEnabledConfigurer() { + return (beanName, bean) -> bean.setSubscriptionEnabled(true); + } + + @Bean + @SuppressWarnings({ "rawtypes", "unchecked" }) + RegionConfigurer customersByNameCacheListenerConfigurer(@Qualifier("customersByNameCacheListener") + CacheListener customersByNameCacheListener) { + + return new RegionConfigurer() { + + @Override + public void configure(String beanName, ClientRegionFactoryBean bean) { + + if ("CustomersByName".equals(beanName)) { + bean.setCacheListeners(ArrayUtils.asArray(customersByNameCacheListener)); + bean.setInterests(ArrayUtils.asArray(new RegexInterest(".*", + InterestResultPolicy.KEYS, false, false))); + } + } + }; + } + + @Bean + CacheListener customersByNameCacheListener() { + return new CustomersByNameCacheListener(); + } + } }