diff --git a/src/main/java/org/springframework/data/gemfire/config/annotation/ContinuousQueryConfiguration.java b/src/main/java/org/springframework/data/gemfire/config/annotation/ContinuousQueryConfiguration.java index dfc4acf4..109ba30a 100644 --- a/src/main/java/org/springframework/data/gemfire/config/annotation/ContinuousQueryConfiguration.java +++ b/src/main/java/org/springframework/data/gemfire/config/annotation/ContinuousQueryConfiguration.java @@ -19,6 +19,7 @@ package org.springframework.data.gemfire.config.annotation; import static java.util.Arrays.stream; import static org.springframework.data.gemfire.util.CollectionUtils.nullSafeMap; +import java.lang.annotation.Annotation; import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -28,7 +29,8 @@ import java.util.concurrent.Executor; import java.util.stream.Collectors; import org.apache.geode.cache.GemFireCache; -import org.apache.shiro.util.Assert; +import org.springframework.aop.framework.AopProxyUtils; +import org.springframework.aop.support.AopUtils; import org.springframework.beans.BeansException; import org.springframework.beans.factory.ListableBeanFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -44,6 +46,7 @@ import org.springframework.data.gemfire.listener.ContinuousQueryListenerContaine import org.springframework.data.gemfire.listener.annotation.ContinuousQuery; import org.springframework.data.gemfire.util.CacheUtils; import org.springframework.lang.Nullable; +import org.springframework.util.Assert; import org.springframework.util.ErrorHandler; import org.springframework.util.StringUtils; @@ -63,6 +66,7 @@ import org.springframework.util.StringUtils; * @see org.springframework.data.gemfire.listener.ContinuousQueryDefinition * @see org.springframework.data.gemfire.listener.ContinuousQueryListener * @see org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer + * @see org.springframework.data.gemfire.listener.annotation.ContinuousQuery * @since 2.0.0 */ @Configuration @@ -82,7 +86,7 @@ public class ContinuousQueryConfiguration extends AbstractAnnotationConfigSuppor private String taskExecutorBeanName; @Override - protected Class getAnnotationType() { + protected Class getAnnotationType() { return EnableContinuousQueries.class; } @@ -118,11 +122,14 @@ public class ContinuousQueryConfiguration extends AbstractAnnotationConfigSuppor this.continuousQueryDefinitions.forEach(this.container::addListener); this.continuousQueryDefinitions.clear(); } - else if (isApplicationBean(bean)) { + else if (isApplicationBean(bean, beanName)) { - List definitions = stream(bean.getClass().getMethods()) + //Object resolvedBean = resolveTargetObject(bean); + Object resolvedBean = bean; + + List definitions = stream(resolvedBean.getClass().getMethods()) .filter(method -> method.isAnnotationPresent(ContinuousQuery.class)) - .map(method -> ContinuousQueryDefinition.from(bean, method)) + .map(method -> ContinuousQueryDefinition.from(resolvedBean, method)) .collect(Collectors.toList()); Optional.ofNullable(this.container).map(container -> { @@ -140,15 +147,36 @@ public class ContinuousQueryConfiguration extends AbstractAnnotationConfigSuppor }; } - private boolean isApplicationBean(Object bean) { + private boolean isApplicationBean(Object bean, String beanName) { return Optional.ofNullable(bean) + //.filter(this::isNotProxy) + //.map(this::resolveTargetObject) .map(Object::getClass) - .filter(type -> type.getPackage().getName().startsWith(ORG_SPRINGFRAMEWORK_DATA_GEMFIRE_PACKAGE_NAME) - || !type.getPackage().getName().startsWith(ORG_SPRINGFRAMEWORK_PACKAGE_NAME)) + .map(Class::getPackage) + .map(Package::getName) + .filter(StringUtils::hasText) + .filter(packageName -> packageName.startsWith(ORG_SPRINGFRAMEWORK_DATA_GEMFIRE_PACKAGE_NAME) + || !packageName.startsWith(ORG_SPRINGFRAMEWORK_PACKAGE_NAME)) .isPresent(); } + private boolean isNotProxy(Object bean) { + return !isProxy(bean); + } + + private boolean isProxy(Object bean) { + return AopUtils.isAopProxy(bean); + } + + private Object resolveTargetObject(Object bean) { + + return Optional.ofNullable(bean) + .filter(this::isProxy) + .map(proxy -> AopProxyUtils.getSingletonTarget(proxy)) + .orElse(bean); + } + @Bean public ContinuousQueryListenerContainer continuousQueryListenerContainer(GemFireCache gemfireCache) { diff --git a/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationUnitTests.java b/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationUnitTests.java index 8dc9ecd5..bf4ba301 100644 --- a/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationUnitTests.java +++ b/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationUnitTests.java @@ -22,8 +22,13 @@ import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.springframework.data.gemfire.util.ArrayUtils.asArray; + +import java.lang.reflect.Proxy; import org.apache.geode.cache.GemFireCache; +import org.apache.geode.cache.Region; +import org.apache.geode.cache.RegionAttributes; import org.apache.geode.cache.client.ClientCache; import org.apache.geode.cache.client.Pool; import org.apache.geode.cache.query.CqAttributes; @@ -31,23 +36,49 @@ import org.apache.geode.cache.query.CqEvent; import org.apache.geode.cache.query.CqQuery; import org.apache.geode.cache.query.QueryService; import org.junit.Test; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.config.BeanFactoryPostProcessor; +import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.context.ConfigurableApplicationContext; 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.core.Ordered; import org.springframework.core.annotation.Order; +import org.springframework.data.annotation.Id; +import org.springframework.data.gemfire.RegionAttributesFactoryBean; import org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer; import org.springframework.data.gemfire.listener.annotation.ContinuousQuery; +import org.springframework.data.gemfire.mapping.GemfireMappingContext; +import org.springframework.data.gemfire.repository.config.EnableGemfireRepositories; +import org.springframework.data.gemfire.repository.support.GemfireRepositoryFactoryBean; +import org.springframework.data.gemfire.test.mock.MockGemFireObjectsSupport; import org.springframework.data.gemfire.test.mock.annotation.EnableGemFireMockObjects; +import org.springframework.data.gemfire.test.model.Person; +import org.springframework.data.gemfire.test.repo.PersonRepository; +import org.springframework.data.gemfire.test.support.IOUtils; +import org.springframework.data.repository.CrudRepository; +import org.springframework.lang.Nullable; import org.springframework.stereotype.Component; +import lombok.Data; + /** * Unit tests for {@link EnableContinuousQueries}, {@link ContinuousQueryConfiguration}, {@link ContinuousQuery} * and {@link ContinuousQueryListenerContainer}. * * @author John Blum + * @see java.lang.reflect.Proxy * @see org.junit.Test + * @see org.apache.geode.cache.GemFireCache + * @see org.apache.geode.cache.Region + * @see org.apache.geode.cache.client.ClientCache + * @see org.apache.geode.cache.query.CqQuery + * @see org.apache.geode.cache.query.QueryService + * @see org.springframework.aop.framework.ProxyFactory * @see org.springframework.data.gemfire.config.annotation.ContinuousQueryConfiguration * @see org.springframework.data.gemfire.config.annotation.EnableContinuousQueries * @see org.springframework.data.gemfire.listener.annotation.ContinuousQuery @@ -59,35 +90,48 @@ public class EnableContinuousQueriesConfigurationUnitTests { return new AnnotationConfigApplicationContext(annotatedClasses); } + private void testRegisterAndExecuteContinuousQuery(Class... annotatedClasses) throws Exception { + + ConfigurableApplicationContext applicationContext = newApplicationContext(annotatedClasses); + + try { + assertThat(applicationContext).isNotNull(); + assertThat(applicationContext.containsBean("DEFAULT")).isTrue(); + + GemFireCache gemfireCache = applicationContext.getBean(ClientCache.class); + + assertThat(gemfireCache).isNotNull(); + + QueryService mockQueryService = gemfireCache.getQueryService(); + + assertThat(mockQueryService).isNotNull(); + assertThat(mockQueryService.getCqs()).hasSize(1); + + CqQuery mockCqQuery = mockQueryService.getCqs()[0]; + + assertThat(mockCqQuery).isNotNull(); + assertThat(mockCqQuery.getName()).isEqualTo("TestQuery"); + assertThat(mockCqQuery.getQueryString()).isEqualTo("SELECT * FROM /Example"); + assertThat(mockCqQuery.isRunning()).isTrue(); + + verify(mockQueryService, times(1)).newCq(eq("TestQuery"), + eq("SELECT * FROM /Example"), any(CqAttributes.class), eq(false)); + + verify(mockCqQuery, times(1)).execute(); + } + finally { + IOUtils.close(applicationContext); + } + } + @Test public void registersAndExecutesContinuousQueries() throws Exception { + testRegisterAndExecuteContinuousQuery(TestContinuousQueryRegistrationAndExecutionConfiguration.class); + } - ConfigurableApplicationContext applicationContext = - newApplicationContext(TestContinuousQueryRegistrationAndExecutionConfiguration.class); - - assertThat(applicationContext).isNotNull(); - assertThat(applicationContext.containsBean("DEFAULT")).isTrue(); - - GemFireCache gemfireCache = applicationContext.getBean(ClientCache.class); - - assertThat(gemfireCache).isNotNull(); - - QueryService mockQueryService = gemfireCache.getQueryService(); - - assertThat(mockQueryService).isNotNull(); - assertThat(mockQueryService.getCqs()).hasSize(1); - - CqQuery mockCqQuery = mockQueryService.getCqs()[0]; - - assertThat(mockCqQuery).isNotNull(); - assertThat(mockCqQuery.getName()).isEqualTo("TestQuery"); - assertThat(mockCqQuery.getQueryString()).isEqualTo("SELECT * FROM /Example"); - assertThat(mockCqQuery.isRunning()).isTrue(); - - verify(mockQueryService, times(1)).newCq(eq("TestQuery"), - eq("SELECT * FROM /Example"), any(CqAttributes.class), eq(false)); - - verify(mockCqQuery, times(1)).execute(); + @Test + public void registersAndExecutesContinuousQueriesOnProxiedBean() throws Exception { + testRegisterAndExecuteContinuousQuery(TestContinuousQueryRegistrationAndExecutionOnProxiedBeanConfiguration.class); } @ClientCacheApplication @@ -99,10 +143,9 @@ public class EnableContinuousQueriesConfigurationUnitTests { @Bean BeanFactoryPostProcessor dependsOnBeanFactoryPostProcessor() { - return configurableListableBeanFactory -> { + return configurableListableBeanFactory -> configurableListableBeanFactory.getBeanDefinition("continuousQueryListenerContainer") - .setDependsOn("testComponent"); - }; + .setDependsOn("continuousQueryComponent"); } @Bean("DEFAULT") @@ -112,11 +155,108 @@ public class EnableContinuousQueriesConfigurationUnitTests { @Bean @Order(Ordered.HIGHEST_PRECEDENCE) - TestContinuousQueryComponent testComponent() { + TestContinuousQueryComponent continuousQueryComponent() { return new TestContinuousQueryComponent(); } } + @Configuration + @EnableGemfireRepositories(basePackageClasses = PersonRepository.class) + @Import(TestContinuousQueryRegistrationAndExecutionConfiguration.class) + @SuppressWarnings("unused") + static class TestContinuousQueryRegistrationAndExecutionOnProxiedBeanConfiguration { + + /* + @Bean + BeanPostProcessor proxyTestContinuousQueryComponentPostProcessor() { + + return new BeanPostProcessor() { + + @Nullable @Override + public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + + if ("continuousQueryComponent".equals(beanName)) { + return new ProxyFactory(bean).getProxy(bean.getClass().getClassLoader()); + } + + return bean; + } + }; + } + */ + + @Bean("People") + Region mockPeopleRegion(GemFireCache gemfireCache, + @Qualifier("peopleRegionAttributes") RegionAttributes peopleRegionAttributes) { + + return MockGemFireObjectsSupport.mockRegion(gemfireCache, "People", peopleRegionAttributes); + } + + @Bean + @SuppressWarnings("unchecked") + RegionAttributesFactoryBean peopleRegionAttributes() { + + RegionAttributesFactoryBean peopleRegionAttributes = new RegionAttributesFactoryBean(); + + peopleRegionAttributes.setKeyConstraint(Long.class); + peopleRegionAttributes.setValueConstraint(Person.class); + + return peopleRegionAttributes; + } + + @Bean("Examples") + Region mockExamplesRegion(GemFireCache gemfireCache, + @Qualifier("examplesRegionAttributes") RegionAttributes mockExamplesRegionAttributes) { + + return MockGemFireObjectsSupport.mockRegion(gemfireCache, "Examples", + mockExamplesRegionAttributes); + } + + @Bean + @SuppressWarnings("unchecked") + RegionAttributesFactoryBean examplesRegionAttributes() { + + RegionAttributesFactoryBean examplesRegionAttributes = new RegionAttributesFactoryBean(); + + examplesRegionAttributes.setKeyConstraint(Long.class); + examplesRegionAttributes.setValueConstraint(Example.class); + + return examplesRegionAttributes; + } + + @Bean + ExampleDataAccessObject exampleDao() { + + return (ExampleDataAccessObject) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), + asArray(ExampleDataAccessObject.class), (proxy, method, args) -> null); + } + + @Bean + GemfireRepositoryFactoryBean exampleRepository() { + + GemfireRepositoryFactoryBean exampleRepositoryFactory = + new GemfireRepositoryFactoryBean<>(ExampleRepository.class); + + exampleRepositoryFactory.setGemfireMappingContext(new GemfireMappingContext()); + exampleRepositoryFactory.setLazyInit(false); + + return exampleRepositoryFactory; + } + + @Bean + Object exampleService(ExampleRepository exampleRepository, ExampleDataAccessObject exampleDao) { + assertThat(exampleRepository).isNotNull(); + assertThat(exampleDao).isNotNull(); + return null; + } + + @Bean + Object personService(PersonRepository personRepository) { + assertThat(personRepository).isNotNull(); + return null; + } + } + @Component @SuppressWarnings("unused") static class TestContinuousQueryComponent { @@ -125,4 +265,16 @@ public class EnableContinuousQueriesConfigurationUnitTests { public void handle(CqEvent event) { } } + + @Data + @org.springframework.data.gemfire.mapping.annotation.Region("Examples") + static class Example { + @Id Long id; + } + + interface ExampleDataAccessObject { + } + + interface ExampleRepository extends CrudRepository { + } } diff --git a/src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerTests.java b/src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerTests.java index 8fb00d0b..7b31d2e1 100644 --- a/src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerTests.java +++ b/src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerTests.java @@ -181,7 +181,7 @@ public class ContinuousQueryListenerContainerTests { } catch (IllegalStateException expected) { - assertThat(expected).hasMessage("QueryService was not properly initialized"); + assertThat(expected).hasMessage("QueryService is required"); assertThat(expected).hasNoCause(); throw expected; diff --git a/src/test/java/org/springframework/data/gemfire/test/model/Person.java b/src/test/java/org/springframework/data/gemfire/test/model/Person.java index 48e25d7d..58aa4104 100644 --- a/src/test/java/org/springframework/data/gemfire/test/model/Person.java +++ b/src/test/java/org/springframework/data/gemfire/test/model/Person.java @@ -31,14 +31,21 @@ import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; /** - * The Person class... + * The {@link Person} class is an Abstract Data Type (ADT) modeling a person. * * @author John Blum - * @since 1.0.0 + * @see java.lang.Comparable + * @see java.util.Calendar + * @see java.util.Date + * @see org.springframework.data.annotation.Id + * @see org.springframework.data.annotation.PersistenceConstructor + * @see org.springframework.data.gemfire.mapping.annotation.Region + * @see org.springframework.data.gemfire.test.support.IdentifierSequence + * @since 2.0.0 */ @Region("People") @SuppressWarnings("unused") -public class Person implements Serializable { +public class Person implements Comparable, Serializable { protected static final String BIRTH_DATE_PATTERN = "yyyy/MM/dd"; @@ -53,11 +60,14 @@ public class Person implements Serializable { private final String lastName; public static Date newBirthDate(int year, int month, int dayOfMonth) { + Calendar birthDate = Calendar.getInstance(); + birthDate.clear(); birthDate.set(Calendar.YEAR, year); birthDate.set(Calendar.MONTH, month); birthDate.set(Calendar.DAY_OF_MONTH, dayOfMonth); + return birthDate.getTime(); } @@ -71,8 +81,9 @@ public class Person implements Serializable { @PersistenceConstructor public Person(String firstName, String lastName, Date birthDate, Gender gender) { - Assert.hasText(firstName, "firstName must be specified"); - Assert.hasText(lastName, "lastName must be specified"); + + Assert.hasText(firstName, "firstName is required"); + Assert.hasText(lastName, "lastName is required"); this.firstName = firstName; this.lastName = lastName; @@ -122,8 +133,24 @@ public class Person implements Serializable { } @Override - public boolean equals(final Object obj) { - if (obj == this) { + public int compareTo(Person that) { + + int result = nullSafeCompareTo(this.getLastName(), that.getLastName()); + + result = (result != 0 ? result : nullSafeCompareTo(this.getFirstName(), that.getLastName())); + result = (result != 0 ? result : nullSafeCompareTo(this.getBirthDate(), that.getBirthDate())); + + return result; + } + + private > int nullSafeCompareTo(T comparableOne, T comparableTwo) { + return (comparableOne == null ? 1 : (comparableTwo == null ? -1 : comparableOne.compareTo(comparableTwo))); + } + + @Override + public boolean equals(Object obj) { + + if (this == obj) { return true; } @@ -142,11 +169,14 @@ public class Person implements Serializable { @Override public int hashCode() { + int hashValue = 17; + hashValue = 37 * hashValue + ObjectUtils.nullSafeHashCode(getId()); hashValue = 37 * hashValue + ObjectUtils.nullSafeHashCode(getBirthDate()); hashValue = 37 * hashValue + ObjectUtils.nullSafeHashCode(getFirstName()); hashValue = 37 * hashValue + ObjectUtils.nullSafeHashCode(getLastName()); + return hashValue; } diff --git a/src/test/java/org/springframework/data/gemfire/test/repo/PersonRepository.java b/src/test/java/org/springframework/data/gemfire/test/repo/PersonRepository.java new file mode 100644 index 00000000..8d6e05d6 --- /dev/null +++ b/src/test/java/org/springframework/data/gemfire/test/repo/PersonRepository.java @@ -0,0 +1,34 @@ +/* + * 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.test.repo; + +import org.springframework.data.gemfire.test.model.Person; +import org.springframework.data.repository.CrudRepository; + +/** + * The {@link PersonRepository} interface defines a Spring Data {@link CrudRepository} used by applications + * to perform basic CRUD and querying data access operations on {@link Person people}. + * + * @author John Blum + * @see java.lang.Long + * @see org.springframework.data.gemfire.test.model.Person + * @see org.springframework.data.repository.CrudRepository + * @since 2.0.0 + */ +public interface PersonRepository extends CrudRepository { + +}