DATAGEODE-52 - Using SD Repositories and @EnableContinuousQueries causes errors with Spring container initialization.

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<? extends Annotation> 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<ContinuousQueryDefinition> definitions = stream(bean.getClass().getMethods())
+					//Object resolvedBean = resolveTargetObject(bean);
+					Object resolvedBean = bean;
+
+					List<ContinuousQueryDefinition> 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)
+			.map(Class::getPackage)
-				|| !type.getPackage().getName().startsWith(ORG_SPRINGFRAMEWORK_PACKAGE_NAME))
+			.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) {
 

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 =
+	@Test
-			newApplicationContext(TestContinuousQueryRegistrationAndExecutionConfiguration.class);
+	public void registersAndExecutesContinuousQueriesOnProxiedBean() throws Exception {
-
+		testRegisterAndExecuteContinuousQuery(TestContinuousQueryRegistrationAndExecutionOnProxiedBeanConfiguration.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();
 	}
 
 	@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<Long, Person> mockPeopleRegion(GemFireCache gemfireCache,
+				@Qualifier("peopleRegionAttributes") RegionAttributes<Long, Person> 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<Long, Example> mockExamplesRegion(GemFireCache gemfireCache,
+				@Qualifier("examplesRegionAttributes") RegionAttributes<Long, Example> 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<ExampleRepository, Example, Long> 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<Example, Long> {
+	}
 }

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;

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<Person>, 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) {
+	public int compareTo(Person that) {
-		if (obj == this) {
+
+		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 <T extends Comparable<T>> 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;
 	}
 

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<Person, Long> {
+
+}