Spring/spring-data-geode
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAGEODE-38 - Add Annotation configuration support for Continuous Queries.

Browse Source
This commit is contained in:
John Blum
2017-08-21 15:50:26 -07:00
parent d9c985bdf7
commit 35df6902fa
10 changed files with 995 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

295
src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationIntegrationTests.java Normal file
View File

@@ -0,0 +1,295 @@
/*
* 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 static org.springframework.data.gemfire.util.ArrayUtils.asArray;
import java.io.Serializable;
import java.util.Collections;
import java.util.concurrent.atomic.AtomicInteger;
import javax.annotation.Resource;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.CacheLoader;
import org.apache.geode.cache.CacheLoaderException;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.GemFireCache;
import org.apache.geode.cache.LoaderHelper;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.geode.cache.query.CqEvent;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
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.PartitionedRegionFactoryBean;
import org.springframework.data.gemfire.client.ClientRegionFactoryBean;
import org.springframework.data.gemfire.listener.annotation.ContinuousQuery;
import org.springframework.data.gemfire.process.ProcessWrapper;
import org.springframework.data.gemfire.support.ConnectionEndpoint;
import org.springframework.data.gemfire.test.support.ClientServerIntegrationTestsSupport;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
import lombok.Data;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
/**
* Integration tests for {@link EnableContinuousQueries}, {@link ContinuousQueryConfiguration}
* and {@link ContinuousQuery}.
*
* @author John Blum
* @see org.junit.Test
* @see org.apache.geode.cache.GemFireCache
* @see org.apache.geode.cache.Region
* @see org.apache.geode.cache.query.CqEvent
* @see org.springframework.data.gemfire.config.annotation.ContinuousQueryConfiguration
* @see org.springframework.data.gemfire.config.annotation.EnableContinuousQueries
* @see org.springframework.data.gemfire.listener.annotation.ContinuousQuery
* @see org.springframework.data.gemfire.process.ProcessWrapper
* @see org.springframework.data.gemfire.test.support.ClientServerIntegrationTestsSupport
* @see org.springframework.test.context.ContextConfiguration
* @see org.springframework.test.context.junit4.SpringRunner
* @since 2.0.0
*/
@RunWith(SpringRunner.class)
@ContextConfiguration(classes = EnableContinuousQueriesConfigurationIntegrationTests.TestConfiguration.class)
@SuppressWarnings("unused")
public class EnableContinuousQueriesConfigurationIntegrationTests extends ClientServerIntegrationTestsSupport {
private static AtomicInteger boilingTemperatureReadingsCounter = new AtomicInteger(0);
private static AtomicInteger freezingTemperatureReadingsCounter = new AtomicInteger(0);
private static AtomicInteger totalTemperatureReadingsCounter = new AtomicInteger(0);
private static ProcessWrapper gemfireServer;
@BeforeClass
public static void startGemFireServer() throws Exception {
int availablePort = findAvailablePort();
gemfireServer = run(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);
}
@Resource(name = "TemperatureReadings")
private Region<Long, TemperatureReading> temperatureReadings;
@Before
public void setup() {
TemperatureReading temperatureReading = this.temperatureReadings.get(1L);
assertThat(temperatureReading).isNotNull();
assertThat(temperatureReading.getTemperature()).isEqualTo(99);
//System.err.printf("Number of Temperature Readings [%d] on Server [%d]%n",
// this.temperatureReadings.size(), this.temperatureReadings.sizeOnServer());
assertThat(this.temperatureReadings.sizeOnServer()).isEqualTo(10);
//waitOn(() -> totalTemperatureReadingsCounter.get() >= 5, 100L);
//assertThat(totalTemperatureReadingsCounter.get()).isNotZero();
//System.err.printf("Total Temperature Readings [%d]%n", totalTemperatureReadingsCounter.get());
}
@Test
public void boilingTemperatureReadingsEqualsThree() {
assertThat(boilingTemperatureReadingsCounter.get()).isEqualTo(3);
}
@Test
public void freezingTemperatureReadingsEqualsTwo() {
assertThat(freezingTemperatureReadingsCounter.get()).isEqualTo(2);
}
@Configuration
@EnableContinuousQueries(poolName = "DEFAULT")
@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")
public void boilingTemperatures(CqEvent event) {
boilingTemperatureReadingsCounter.incrementAndGet();
}
@ContinuousQuery(name = "FreezingTemperatures", query = "SELECT * FROM /TemperatureReadings r WHERE r.temperature <= 32")
public void freezingTemperatures(CqEvent event) {
freezingTemperatureReadingsCounter.incrementAndGet();
}
}
@ClientCacheApplication(logLevel = "warning", subscriptionEnabled = true)
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)));
}
@Bean(name = "TemperatureReadings")
ClientRegionFactoryBean<Long, TemperatureReading> temperatureReadingsRegion(GemFireCache gemfireCache) {
ClientRegionFactoryBean<Long, TemperatureReading> temperatureReadings =
new ClientRegionFactoryBean<>();
temperatureReadings.setCache(gemfireCache);
temperatureReadings.setCacheListeners(asArray(temperatureReadingCounterListener()));
temperatureReadings.setClose(false);
temperatureReadings.setShortcut(ClientRegionShortcut.PROXY);
return temperatureReadings;
}
private CacheListener<Long, TemperatureReading> temperatureReadingCounterListener() {
return new CacheListenerAdapter<Long, TemperatureReading>() {
@Override
public void afterCreate(EntryEvent<Long, TemperatureReading> event) {
if (Operation.LOCAL_LOAD_CREATE.equals(event.getOperation())) {
totalTemperatureReadingsCounter.incrementAndGet();
}
}
};
}
}
@CacheServerApplication(name = "EnableContinuousQueriesConfigurationIntegrationTests", logLevel = "warning")
static class GemFireServerConfiguration {
public static void main(String[] args) {
AnnotationConfigApplicationContext applicationContext =
new AnnotationConfigApplicationContext(GemFireServerConfiguration.class);
applicationContext.registerShutdownHook();
}
@Bean
CacheServerConfigurer cacheServerPortConfigurer(
@Value("${" + GEMFIRE_CACHE_SERVER_PORT_PROPERTY + ":40404}") int port) {
return (bean, cacheServerFactoryBean) -> cacheServerFactoryBean.setPort(port);
}
@Bean(name = "TemperatureReadings")
PartitionedRegionFactoryBean<Long, TemperatureReading> temperatureReadingsRegion(GemFireCache gemfireCache) {
PartitionedRegionFactoryBean<Long, TemperatureReading> temperatureReadings =
new PartitionedRegionFactoryBean<>();
temperatureReadings.setCache(gemfireCache);
temperatureReadings.setCacheLoader(temperatureReadingsLoader());
temperatureReadings.setClose(false);
temperatureReadings.setPersistent(false);
return temperatureReadings;
}
private CacheLoader<Long, TemperatureReading> temperatureReadingsLoader() {
return new CacheLoader<Long, TemperatureReading>() {
@Override
public TemperatureReading load(LoaderHelper<Long, TemperatureReading> helper) throws CacheLoaderException {
long key = helper.getKey();
Region<Long, TemperatureReading> temperatureReadings = helper.getRegion();
recordTemperature(temperatureReadings, ++key, 213);
recordTemperature(temperatureReadings, ++key, 72);
recordTemperature(temperatureReadings, ++key, 400);
recordTemperature(temperatureReadings, ++key, 1024);
recordTemperature(temperatureReadings, ++key, 43);
recordTemperature(temperatureReadings, ++key, 0);
recordTemperature(temperatureReadings, ++key, 33);
recordTemperature(temperatureReadings, ++key, -45);
recordTemperature(temperatureReadings, ++key, 67);
return TemperatureReading.newTemperatureReading(99);
}
private void recordTemperature(Region<Long, TemperatureReading> temperatureReadings,
long key, int temperature) {
sleep(50);
temperatureReadings.put(key, TemperatureReading.newTemperatureReading(temperature));
}
private void sleep(long milliseconds) {
try {
Thread.sleep(milliseconds);
}
catch (InterruptedException ignore) {
}
}
@Override
public void close() {
}
};
}
}
@Data
@RequiredArgsConstructor(staticName = "newTemperatureReading")
public static class TemperatureReading implements Serializable {
@NonNull
private Integer temperature;
@Override
public String toString() {
return String.format("%d °F", getTemperature());
}
}
}

25
src/test/java/org/springframework/data/gemfire/config/xml/ContinuousQueryListenerContainerNamespaceTest.java
View File

@@ -50,19 +50,18 @@ import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.util.ErrorHandler;
/**
* The ContinuousQueryListenerContainerNamespaceTest class is a test suite of test cases testing the SDG XML namespace
* for proper configuration and initialization of a ContinuousQueryListenerContainer bean component
* in the Spring context.
* Integration tests for the configuration and initialization of the SDG {@link ContinuousQueryListenerContainer}
* using the SDG XML namespace.
*
* @author John Blum
* @see org.junit.Test
* @see org.junit.runner.RunWith
* @see org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer
* @see org.springframework.test.context.ContextConfiguration
* @see org.springframework.test.context.junit4.SpringRunner
* @see org.apache.geode.cache.client.ClientCache
* @see org.apache.geode.cache.query.CqListener
* @see org.apache.geode.cache.query.CqQuery
* @see org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer
* @see org.springframework.test.context.ContextConfiguration
* @see org.springframework.test.context.junit4.SpringRunner
* @since 1.4.0
*/
@RunWith(SpringRunner.class)
@@ -74,6 +73,7 @@ public class ContinuousQueryListenerContainerNamespaceTest extends ClientServerI
@BeforeClass
public static void startGemFireServer() throws Exception {
int availablePort = findAvailablePort();
gemfireServer = run(CqCacheServerProcess.class,
@@ -104,6 +104,7 @@ public class ContinuousQueryListenerContainerNamespaceTest extends ClientServerI
@Test
public void testContainerConfiguration() throws Exception {
assertNotNull("The ContinuousQueryListenerContainer was not properly configured!", container);
assertTrue("The CQ Listener Container should be active (initialized)!", container.isActive());
assertFalse("The CQ Listener container should not be configured to auto-start!", container.isAutoStartup());
@@ -119,9 +120,10 @@ public class ContinuousQueryListenerContainerNamespaceTest extends ClientServerI
assertNotNull(queries);
assertEquals(3, queries.length);
List<String> actualNames = new ArrayList<String>(3);
List<String> actualNames = new ArrayList<>(3);
for (CqQuery query : queries) {
actualNames.add(query.getName());
assertEquals("SELECT * FROM /test-cq", query.getQueryString());
@@ -129,12 +131,9 @@ public class ContinuousQueryListenerContainerNamespaceTest extends ClientServerI
CqListener cqListener = query.getCqAttributes().getCqListener();
assertNotNull(cqListener);
// the CqListener object should be an instance of...
// org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer.EventDispatcherAdapter
// So, get the SDG "ContinuousQueryListener"...
ContinuousQueryListener listener = TestUtils.readField("delegate", cqListener);
// The CqListener should be an instance of o.s.d.g.listener.ContinuousQueryListenerContainer.EventDispatcherAdapter
// So, get the SDG "ContinuousQueryListener"
ContinuousQueryListener listener = TestUtils.readField("listener", cqListener);
assertTrue(listener instanceof ContinuousQueryListenerAdapter);
assertTrue(((ContinuousQueryListenerAdapter) listener).getDelegate() instanceof GemfireMDP);

22
src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerTests.java
View File

@@ -125,7 +125,7 @@ public class ContinuousQueryListenerContainerTests {
assertThat(cqListenerContainer.getQueryService()).isEqualTo(mockQueryService);
assertThat(cqListenerContainer.getTaskExecutor()).isInstanceOf(Executor.class);
verify(mockBeanFactory, times(1)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME));
verify(mockBeanFactory, times(2)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME));
verify(mockBeanFactory, times(1)).isTypeMatch(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class));
verify(mockBeanFactory, times(1)).getBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class));
verify(mockPool, times(2)).getName();
@@ -143,6 +143,7 @@ public class ContinuousQueryListenerContainerTests {
QueryService mockQueryService = mock(QueryService.class);
when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(true);
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
cqListenerContainer.setAutoStartup(false);
@@ -160,7 +161,7 @@ public class ContinuousQueryListenerContainerTests {
assertThat(cqListenerContainer.getQueryService()).isSameAs(mockQueryService);
assertThat(cqListenerContainer.getTaskExecutor()).isSameAs(mockExecutor);
verify(mockBeanFactory, never()).containsBean(eq("TestPool"));
verify(mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
verify(mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class));
verify(poolManagerSpy, never()).find(anyString());
@@ -170,6 +171,7 @@ public class ContinuousQueryListenerContainerTests {
@Test(expected = IllegalStateException.class)
public void afterPropertiesSetThrowsIllegalStateExceptionWhenQueryServiceIsUninitialized() {
when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(true);
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
try {
@@ -189,7 +191,7 @@ public class ContinuousQueryListenerContainerTests {
assertThat(cqListenerContainer.isAutoStartup()).isTrue();
assertThat(cqListenerContainer.isRunning()).isFalse();
verify(mockBeanFactory, never()).containsBean(eq("TestPool"));
verify(mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
verify(mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class));
}
@@ -240,12 +242,14 @@ public class ContinuousQueryListenerContainerTests {
@Test
public void eagerlyInitializePoolWithGivenPoolName() {
when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(true);
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
cqListenerContainer.setBeanFactory(mockBeanFactory);
assertThat(cqListenerContainer.eagerlyInitializePool("TestPool")).isEqualTo("TestPool");
verify(mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
verify(mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class));
}
@@ -255,7 +259,7 @@ public class ContinuousQueryListenerContainerTests {
Pool mockPool = mock(Pool.class);
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(false);
when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(false);
when(mockPool.getName()).thenReturn("TestPool");
try {
@@ -268,7 +272,8 @@ public class ContinuousQueryListenerContainerTests {
finally {
assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue();
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
verify(mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(mockBeanFactory, never()).isTypeMatch(eq("TestPool"), eq(Pool.class));
verify(mockBeanFactory, never()).getBean(eq("TestPool"), eq(Pool.class));
verify(mockPool, atLeast(1)).getName();
}
@@ -279,6 +284,7 @@ public class ContinuousQueryListenerContainerTests {
Pool mockPool = mock(Pool.class);
when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(true);
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
when(mockBeanFactory.getBean(eq("TestPool"), eq(Pool.class)))
.thenThrow(new NoSuchBeanDefinitionException("TEST"));
@@ -294,6 +300,7 @@ public class ContinuousQueryListenerContainerTests {
finally {
assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue();
verify(mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
verify(mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class));
verify(mockPool, atLeast(1)).getName();
@@ -303,7 +310,7 @@ public class ContinuousQueryListenerContainerTests {
@Test(expected = IllegalArgumentException.class)
public void eagerlyInitializePoolThrowsIllegalArgumentExceptionCausedByNoPoolWithGivenName() {
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(false);
when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(false);
try {
cqListenerContainer.setBeanFactory(mockBeanFactory);
@@ -317,7 +324,8 @@ public class ContinuousQueryListenerContainerTests {
throw expected;
}
finally {
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
verify(mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(mockBeanFactory, never()).isTypeMatch(anyString(), eq(Pool.class));
verify(mockBeanFactory, never()).getBean(eq("TestPool"), eq(Pool.class));
}
}