diff --git a/pom.xml b/pom.xml index c7ed39c9..fe2abaa9 100644 --- a/pom.xml +++ b/pom.xml @@ -272,6 +272,7 @@ true ${basedir}/src/test/resources/trusted.keystore + error apache-geode diff --git a/src/main/java/org/springframework/data/gemfire/config/annotation/ClusterConfigurationConfiguration.java b/src/main/java/org/springframework/data/gemfire/config/annotation/ClusterConfigurationConfiguration.java index 71bdf850..ef271731 100644 --- a/src/main/java/org/springframework/data/gemfire/config/annotation/ClusterConfigurationConfiguration.java +++ b/src/main/java/org/springframework/data/gemfire/config/annotation/ClusterConfigurationConfiguration.java @@ -18,6 +18,7 @@ package org.springframework.data.gemfire.config.annotation; import static java.util.stream.StreamSupport.stream; import static org.springframework.data.gemfire.util.CacheUtils.isClient; +import static org.springframework.data.gemfire.util.CacheUtils.isPeer; import java.lang.annotation.Annotation; import java.util.Optional; @@ -27,17 +28,12 @@ import org.apache.geode.cache.Region; import org.apache.geode.cache.RegionShortcut; import org.apache.geode.cache.client.ClientCache; import org.apache.geode.cache.query.Index; -import org.springframework.beans.BeansException; -import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.ImportAware; -import org.springframework.context.event.ApplicationContextEvent; -import org.springframework.context.event.ContextRefreshedEvent; -import org.springframework.context.event.EventListener; import org.springframework.core.OrderComparator; import org.springframework.core.annotation.AnnotationAttributes; import org.springframework.core.type.AnnotationMetadata; -import org.springframework.data.gemfire.GemfireUtils; import org.springframework.data.gemfire.config.admin.GemfireAdminOperations; import org.springframework.data.gemfire.config.admin.remote.FunctionGemfireAdminTemplate; import org.springframework.data.gemfire.config.admin.remote.RestHttpGemfireAdminTemplate; @@ -51,6 +47,9 @@ import org.springframework.data.gemfire.config.schema.support.ComposableSchemaOb import org.springframework.data.gemfire.config.schema.support.IndexCollector; import org.springframework.data.gemfire.config.schema.support.IndexDefiner; import org.springframework.data.gemfire.config.schema.support.RegionDefiner; +import org.springframework.data.gemfire.config.support.AbstractSmartLifecycle; +import org.springframework.data.gemfire.util.CacheUtils; +import org.springframework.util.Assert; /** * Spring {@link Configuration @Configuration} class defining Spring beans that will record the creation of @@ -162,70 +161,22 @@ public class ClusterConfigurationConfiguration extends AbstractAnnotationConfigS } } - @EventListener - public void gemfireClusterSchemaCreationHandler(ContextRefreshedEvent event) { + @Bean + public ClusterSchemaObjectInitializer gemfireClusterSchemaObjectInitializer(GemFireCache gemfireCache) { - GemFireCache gemfireCache = resolveGemFireCache(event); + return Optional.ofNullable(gemfireCache) + .filter(CacheUtils::isClient) + .map(clientCache -> { - if (isClient(gemfireCache)) { + SchemaObjectContext schemaObjectContext = SchemaObjectContext.from(gemfireCache) + .with(newGemfireAdminOperations((ClientCache) clientCache)) + .with(newSchemaObjectCollector()) + .with(newSchemaObjectDefiner()); - GemfireAdminOperations gemfireAdminOperations = newGemfireAdminOperations((ClientCache) gemfireCache); + return new ClusterSchemaObjectInitializer(schemaObjectContext); - SchemaObjectDefiner schemaObjectDefiner = newSchemaObjectDefiner(); - - Iterable schemaObjects = newSchemaObjectCollector().collectFrom(resolveApplicationContext(event)); - - stream(schemaObjects.spliterator(), false) - .map(schemaObjectDefiner::define) - .sorted(OrderComparator.INSTANCE) - .forEach(schemaObjectDefinition -> - schemaObjectDefinition.ifPresent(it -> it.create(gemfireAdminOperations))); - } - /* - else if (isPeer(gemfireCache)) { - - GemfireFunctionUtils.registerFunctionForPojoMethod(new CreateRegionFunction(), - CreateRegionFunction.CREATE_REGION_FUNCTION_ID); - - GemfireFunctionUtils.registerFunctionForPojoMethod(new CreateIndexFunction(), - CreateIndexFunction.CREATE_INDEX_FUNCTION_ID); - } - */ - } - - /** - * Resolves a reference to the Spring {@link ApplicationContext} from the given {@link ApplicationContextEvent}. - * - * @param event {@link ApplicationContextEvent} from which to resolve the Spring {@link ApplicationContext}. - * @return the resolved Spring {@link ApplicationContext}. - * @see org.springframework.context.event.ApplicationContextEvent - * @see org.springframework.context.ApplicationContext - */ - private ApplicationContext resolveApplicationContext(ApplicationContextEvent event) { - return event.getApplicationContext(); - } - - /** - * Tries to resolve the {@link GemFireCache} from the Spring {@link ApplicationContext}. - * The {@link GemFireCache} will be resolvable from the Spring {@link ApplicationContext} if the cache - * was registered a managed bean in the Spring container. - * - * If the {@link GemFireCache} cannot be resolved from the {@link ApplicationContext}, this method will attempt - * to resolve the cache reference from GemFire's global context using the GemFire API. - * - * @param event Spring {@link ApplicationContextEvent} encapsulating the details of the Spring container event. - * @return the resolved {@link GemFireCache} if available. - * @see org.springframework.context.event.ApplicationContextEvent - * @see org.apache.geode.cache.GemFireCache - */ - private GemFireCache resolveGemFireCache(ApplicationContextEvent event) { - - try { - return resolveApplicationContext(event).getBean(GemFireCache.class); - } - catch (BeansException ignore) { - return GemfireUtils.resolveGemFireCache(); - } + }) + .orElse(null); } /** @@ -283,4 +234,147 @@ public class ClusterConfigurationConfiguration extends AbstractAnnotationConfigS new IndexDefiner() ); } + + public static class ClusterSchemaObjectInitializer extends AbstractSmartLifecycle { + + private final SchemaObjectContext schemaObjectContext; + + protected ClusterSchemaObjectInitializer(SchemaObjectContext schemaObjectContext) { + Assert.notNull(schemaObjectContext, "SchemaObjectContext is required"); + this.schemaObjectContext = schemaObjectContext; + } + + @Override + public boolean isAutoStartup() { + return true; + } + + @Override + public int getPhase() { + return Integer.MIN_VALUE; + } + + protected SchemaObjectContext getSchemaObjectContext() { + return this.schemaObjectContext; + } + + @Override + public void start() { + + SchemaObjectContext schemaObjectContext = getSchemaObjectContext(); + + if (schemaObjectContext.isClientCache()) { + + Iterable schemaObjects = schemaObjectContext.getSchemaObjectCollector() + .collectFrom(requireApplicationContext()); + + stream(schemaObjects.spliterator(), false) + .map(schemaObjectContext.getSchemaObjectDefiner()::define) + .sorted(OrderComparator.INSTANCE) + .forEach(schemaObjectDefinition -> schemaObjectDefinition + .ifPresent(it -> it.create(schemaObjectContext.getGemfireAdminOperations()))); + + setRunning(true); + + } + /* + else if (schemaObjectContext.isPeerCache()) { + + GemfireFunctionUtils.registerFunctionForPojoMethod(new CreateRegionFunction(), + CreateRegionFunction.CREATE_REGION_FUNCTION_ID); + + GemfireFunctionUtils.registerFunctionForPojoMethod(new CreateIndexFunction(), + CreateIndexFunction.CREATE_INDEX_FUNCTION_ID); + + } + */ + } + + @Override + public void stop() { + setRunning(false); + } + + @Override + public void stop(Runnable callback) { + setRunning(false); + callback.run(); + } + } + + public static class SchemaObjectContext { + + private final GemFireCache gemfireCache; + + private GemfireAdminOperations gemfireAdminOperations; + + private SchemaObjectCollector schemaObjectCollector; + + private SchemaObjectDefiner schemaObjectDefiner; + + protected static SchemaObjectContext from(GemFireCache gemfireCache) { + return new SchemaObjectContext(gemfireCache); + } + + private SchemaObjectContext(GemFireCache gemfireCache) { + Assert.notNull(gemfireCache, "GemFireCache is required"); + this.gemfireCache = gemfireCache; + } + + public GemfireAdminOperations getGemfireAdminOperations() { + + Assert.state(this.gemfireAdminOperations != null, + "GemfireAdminOperations was not initialized"); + + return this.gemfireAdminOperations; + } + + public boolean isClientCache() { + return isClient(getGemfireCache()); + } + + public boolean isPeerCache() { + return isPeer(getGemfireCache()); + } + + @SuppressWarnings("unchecked") + public T getGemfireCache() { + return (T) this.gemfireCache; + } + + public SchemaObjectCollector getSchemaObjectCollector() { + + Assert.state(this.schemaObjectCollector != null, + "SchemaObjectCollector was not initialized"); + + return this.schemaObjectCollector; + } + + + public SchemaObjectDefiner getSchemaObjectDefiner() { + + Assert.state(this.schemaObjectDefiner != null, + "SchemaObjectDefiner was not initialized"); + + return this.schemaObjectDefiner; + } + + protected SchemaObjectContext with(GemfireAdminOperations gemfireAdminOperations) { + Assert.notNull(gemfireAdminOperations, "GemfireAdminOperations are required"); + this.gemfireAdminOperations = gemfireAdminOperations; + return this; + } + + protected SchemaObjectContext with(SchemaObjectCollector schemaObjectCollector) { + Assert.notNull(schemaObjectCollector, "SchemaObjectCollector is required"); + this.schemaObjectCollector = schemaObjectCollector; + return this; + } + + protected SchemaObjectContext with(SchemaObjectDefiner schemaObjectDefiner) { + Assert.notNull(schemaObjectDefiner, "SchemaObjectDefiner is required"); + this.schemaObjectDefiner = schemaObjectDefiner; + return this; + } + } } diff --git a/src/main/java/org/springframework/data/gemfire/config/support/AbstractSmartLifecycle.java b/src/main/java/org/springframework/data/gemfire/config/support/AbstractSmartLifecycle.java new file mode 100644 index 00000000..be9d3c09 --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/config/support/AbstractSmartLifecycle.java @@ -0,0 +1,90 @@ +/* + * Copyright 2017 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 + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.data.gemfire.config.support; + +import static org.springframework.data.gemfire.util.RuntimeExceptionFactory.newIllegalStateException; + +import java.util.Optional; + +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.context.SmartLifecycle; + +/** + * {@link AbstractSmartLifecycle} is an abstract base class implementing the Spring{@link SmartLifecycle} interface + * to support custom implementations. + * + * @author John Blum + * @see org.springframework.context.SmartLifecycle + * @since 2.0.2 + */ +@SuppressWarnings("unused") +public abstract class AbstractSmartLifecycle implements ApplicationContextAware, SmartLifecycle { + + protected static final int DEFAULT_PHASE = 0; + + private volatile boolean running = false; + + private ApplicationContext applicationContext; + + @Override + public boolean isAutoStartup() { + return false; + } + + @Override + public boolean isRunning() { + return this.running; + } + + protected void setRunning(boolean running) { + this.running = running; + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; + } + + protected Optional getApplicationContext() { + return Optional.ofNullable(this.applicationContext); + } + + protected ApplicationContext requireApplicationContext() { + + return getApplicationContext() + .orElseThrow(() -> newIllegalStateException("ApplicationContext could not be resolved")); + } + + @Override + public int getPhase() { + return DEFAULT_PHASE; + } + + @Override + public void start() { + } + + @Override + public void stop() { + } + + @Override + public void stop(Runnable callback) { + } +} diff --git a/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationIntegrationTests.java b/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationIntegrationTests.java index 08d71ead..523da9a3 100644 --- a/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationIntegrationTests.java +++ b/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationIntegrationTests.java @@ -142,30 +142,24 @@ public class EnableContinuousQueriesConfigurationIntegrationTests extends Client } @Configuration - @EnableContinuousQueries(poolName = "DEFAULT") + @EnableContinuousQueries @Import(GemFireClientConfiguration.class) static class TestConfiguration { - @Bean - TemperatureReadingQueryListeners temperatureReadingQueryListeners() { - return new TemperatureReadingQueryListeners(); - } - } - - static class TemperatureReadingQueryListeners { - - @ContinuousQuery(name = "BoilingTemperatures", query = "SELECT * FROM /TemperatureReadings r WHERE r.temperature >= 212") + @ContinuousQuery(name = "BoilingTemperatures", + query = "SELECT * FROM /TemperatureReadings r WHERE r.temperature >= 212") public void boilingTemperatures(CqEvent event) { boilingTemperatureReadingsCounter.incrementAndGet(); } - @ContinuousQuery(name = "FreezingTemperatures", query = "SELECT * FROM /TemperatureReadings r WHERE r.temperature <= 32") + @ContinuousQuery(name = "FreezingTemperatures", + query = "SELECT * FROM /TemperatureReadings r WHERE r.temperature <= 32") public void freezingTemperatures(CqEvent event) { freezingTemperatureReadingsCounter.incrementAndGet(); } } - @ClientCacheApplication(logLevel = "warning", subscriptionEnabled = true) + @ClientCacheApplication(logLevel = "error", subscriptionEnabled = true) static class GemFireClientConfiguration { @Bean @@ -204,7 +198,7 @@ public class EnableContinuousQueriesConfigurationIntegrationTests extends Client } } - @CacheServerApplication(name = "EnableContinuousQueriesConfigurationIntegrationTests", logLevel = "warning") + @CacheServerApplication(name = "EnableContinuousQueriesConfigurationIntegrationTests", logLevel = "error") static class GemFireServerConfiguration { public static void main(String[] args) { @@ -215,13 +209,6 @@ public class EnableContinuousQueriesConfigurationIntegrationTests extends Client applicationContext.registerShutdownHook(); } - @Bean - CacheServerConfigurer cacheServerPortConfigurer( - @Value("${" + GEMFIRE_CACHE_SERVER_PORT_PROPERTY + ":40404}") int port) { - - return (bean, cacheServerFactoryBean) -> cacheServerFactoryBean.setPort(port); - } - @Bean(name = "TemperatureReadings") PartitionedRegionFactoryBean temperatureReadingsRegion(GemFireCache gemfireCache) { diff --git a/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesWithClusterConfigurationIntegrationTests.java b/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesWithClusterConfigurationIntegrationTests.java new file mode 100644 index 00000000..3e2ef7f7 --- /dev/null +++ b/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesWithClusterConfigurationIntegrationTests.java @@ -0,0 +1,155 @@ +/* + * Copyright 2017 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 + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.data.gemfire.config.annotation; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.Collections; +import java.util.Optional; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; + +import org.apache.geode.cache.query.CqEvent; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.data.gemfire.listener.annotation.ContinuousQuery; +import org.springframework.data.gemfire.process.ProcessWrapper; +import org.springframework.data.gemfire.repository.config.EnableGemfireRepositories; +import org.springframework.data.gemfire.support.ConnectionEndpoint; +import org.springframework.data.gemfire.test.model.Gender; +import org.springframework.data.gemfire.test.model.Person; +import org.springframework.data.gemfire.test.repo.PersonRepository; +import org.springframework.data.gemfire.test.support.ClientServerIntegrationTestsSupport; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * Integration tests testing the combination of {@link EnableContinuousQueries} with {@link EnableClusterConfiguration}. + * + * @author John Blum + * @see org.junit.Test + * @see org.apache.geode.cache.query.CqEvent + * @see org.springframework.data.gemfire.config.annotation.EnableClusterConfiguration + * @see org.springframework.data.gemfire.config.annotation.EnableContinuousQueries + * @see org.springframework.data.gemfire.listener.annotation.ContinuousQuery + * @see org.springframework.test.context.ContextConfiguration + * @see org.springframework.test.context.junit4.SpringRunner + * @see Fix race condition between ContinuousQuery registration and EnableClusterConfiguration Region creation. + * @since 2.0.3 + */ +@RunWith(SpringRunner.class) +@ContextConfiguration(classes = EnableContinuousQueriesWithClusterConfigurationIntegrationTests.TestConfiguration.class) +@SuppressWarnings("unused") +public class EnableContinuousQueriesWithClusterConfigurationIntegrationTests + extends ClientServerIntegrationTestsSupport { + + private static final BlockingQueue events = new ArrayBlockingQueue<>(2); + + private static ProcessWrapper gemfireServer; + + @BeforeClass + public static void startGemFireServer() throws Exception { + + int availablePort = findAvailablePort(); + + gemfireServer = run(EnableContinuousQueriesWithClusterConfigurationIntegrationTests.GemFireServerConfiguration.class, + String.format("-D%s=%d", GEMFIRE_CACHE_SERVER_PORT_PROPERTY, availablePort)); + + waitForServerToStart("localhost", availablePort); + + System.setProperty(GEMFIRE_CACHE_SERVER_PORT_PROPERTY, String.valueOf(availablePort)); + } + + @AfterClass + public static void stopGemFireServer() { + System.clearProperty(GEMFIRE_CACHE_SERVER_PORT_PROPERTY); + stop(gemfireServer); + } + + @Autowired + private PersonRepository personRepository; + + @Test + public void personEventsFired() throws Exception { + + Person jonDoe = Person.newPerson(1L, "Jon", "Doe", null, Gender.MALE); + + jonDoe = this.personRepository.save(jonDoe); + + assertThat(this.personRepository.findById(jonDoe.getId()).orElse(null)).isEqualTo(jonDoe); + assertThat(events.poll(5L, TimeUnit.SECONDS)).isEqualTo(jonDoe); + + Person janeDoe = Person.newPerson(2L, "Jane", "Doe", null, Gender.FEMALE); + + janeDoe = this.personRepository.save(janeDoe); + + assertThat(this.personRepository.findById(janeDoe.getId()).orElse(null)).isEqualTo(janeDoe); + assertThat(events.poll(5L, TimeUnit.SECONDS)).isEqualTo(janeDoe); + } + + @Configuration + @EnableContinuousQueries + @Import(GemFireClientConfiguration.class) + static class TestConfiguration { + + @ContinuousQuery(name = "PersonEvents", query = "SELECT * FROM /People") + public void peopleEventHandler(CqEvent event) { + + Optional.ofNullable(event) + .map(CqEvent::getNewValue) + .filter(newValue -> newValue instanceof Person) + .map(newValue -> (Person) newValue) + .ifPresent(events::offer); + } + } + + @ClientCacheApplication(logLevel = "error", subscriptionEnabled = true) + @EnableClusterConfiguration + @EnableEntityDefinedRegions(basePackageClasses = Person.class) + @EnableGemfireRepositories(basePackageClasses = PersonRepository.class) + static class GemFireClientConfiguration { + + @Bean + ClientCacheConfigurer clientCachePoolPortConfigurer( + @Value("${" + GEMFIRE_CACHE_SERVER_PORT_PROPERTY + ":40404}") int port) { + + return (bean, clientCacheFactoryBean) -> clientCacheFactoryBean.setServers( + Collections.singletonList(new ConnectionEndpoint("localhost", port))); + } + } + + @CacheServerApplication(name = "EnableContinuousQueriesWithClusterConfigurationIntegrationTests", logLevel = "error") + static class GemFireServerConfiguration { + + public static void main(String[] args) { + + AnnotationConfigApplicationContext applicationContext = + new AnnotationConfigApplicationContext(GemFireServerConfiguration.class); + + applicationContext.registerShutdownHook(); + } + } +} diff --git a/src/test/resources/logback.xml b/src/test/resources/logback.xml index 05ced12f..63ea4f79 100644 --- a/src/test/resources/logback.xml +++ b/src/test/resources/logback.xml @@ -1,5 +1,5 @@ - + @@ -13,14 +13,16 @@ - + + + - - + +