DATAGEODE-37 - Apply Java 8 types and functionality to the Continuous Query support.
This commit is contained in:
John Blum
@@ -17,14 +17,16 @@
|
||||
|
||||
package org.springframework.data.gemfire.listener;
|
||||
|
||||
import static org.hamcrest.Matchers.equalTo;
|
||||
import static org.hamcrest.Matchers.instanceOf;
|
||||
import static org.hamcrest.Matchers.is;
|
||||
import static org.hamcrest.Matchers.nullValue;
|
||||
import static org.hamcrest.Matchers.sameInstance;
|
||||
import static org.junit.Assert.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyBoolean;
|
||||
import static org.mockito.ArgumentMatchers.isA;
|
||||
import static org.mockito.Matchers.anyString;
|
||||
import static org.mockito.Matchers.eq;
|
||||
import static org.mockito.Mockito.atLeast;
|
||||
import static org.mockito.Mockito.doAnswer;
|
||||
import static org.mockito.Mockito.doReturn;
|
||||
import static org.mockito.Mockito.doThrow;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.spy;
|
||||
@@ -33,21 +35,32 @@ import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.verifyZeroInteractions;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.Executor;
|
||||
|
||||
import org.apache.geode.cache.RegionService;
|
||||
import org.apache.geode.cache.client.Pool;
|
||||
import org.apache.geode.cache.query.CqAttributes;
|
||||
import org.apache.geode.cache.query.CqEvent;
|
||||
import org.apache.geode.cache.query.CqException;
|
||||
import org.apache.geode.cache.query.CqQuery;
|
||||
import org.apache.geode.cache.query.CqState;
|
||||
import org.apache.geode.cache.query.QueryException;
|
||||
import org.apache.geode.cache.query.QueryService;
|
||||
import org.apache.geode.internal.cache.PoolManagerImpl;
|
||||
import org.junit.Before;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
import org.junit.rules.ExpectedException;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.junit.MockitoJUnitRunner;
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.beans.factory.DisposableBean;
|
||||
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
|
||||
import org.springframework.data.gemfire.GemfireQueryException;
|
||||
import org.springframework.data.gemfire.GemfireUtils;
|
||||
import org.springframework.data.gemfire.TestUtils;
|
||||
import org.springframework.data.gemfire.config.xml.GemfireConstants;
|
||||
import org.springframework.util.ErrorHandler;
|
||||
|
||||
/**
|
||||
* Unit tests for {@link ContinuousQueryListenerContainer}.
|
||||
@@ -60,43 +73,57 @@ import org.springframework.data.gemfire.config.xml.GemfireConstants;
|
||||
* @see org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer
|
||||
* @since 1.8.0
|
||||
*/
|
||||
@RunWith(MockitoJUnitRunner.class)
|
||||
public class ContinuousQueryListenerContainerTests {
|
||||
|
||||
@Rule
|
||||
public ExpectedException exception = ExpectedException.none();
|
||||
@Mock
|
||||
private BeanFactory mockBeanFactory;
|
||||
|
||||
private ContinuousQueryListenerContainer listenerContainer;
|
||||
private ContinuousQueryListenerContainer cqListenerContainer;
|
||||
|
||||
private CqQuery mockCqQuery(String name, String query, CqAttributes attributes, boolean durable) {
|
||||
|
||||
CqQuery mockQuery = mock(CqQuery.class);
|
||||
|
||||
when(mockQuery.getName()).thenReturn(name);
|
||||
when(mockQuery.getQueryString()).thenReturn(query);
|
||||
when(mockQuery.getCqAttributes()).thenReturn(attributes);
|
||||
when(mockQuery.isDurable()).thenReturn(durable);
|
||||
|
||||
return mockQuery;
|
||||
}
|
||||
@Before
|
||||
public void setup() {
|
||||
listenerContainer = new ContinuousQueryListenerContainer();
|
||||
cqListenerContainer = spy(new ContinuousQueryListenerContainer());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void afterPropertiesSetIsAutoStart() throws Exception {
|
||||
BeanFactory mockBeanFactory = mock(BeanFactory.class);
|
||||
|
||||
Pool mockPool = mock(Pool.class);
|
||||
|
||||
QueryService mockQueryService = mock(QueryService.class);
|
||||
|
||||
when(mockBeanFactory.containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME))).thenReturn(true);
|
||||
when(mockBeanFactory.isTypeMatch(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class))).thenReturn(true);
|
||||
when(mockBeanFactory.getBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class))).thenReturn(mockPool);
|
||||
when(mockPool.getName()).thenReturn(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME);
|
||||
when(mockPool.getQueryService()).thenReturn(mockQueryService);
|
||||
|
||||
try {
|
||||
PoolManagerImpl.getPMI().register(mockPool);
|
||||
|
||||
listenerContainer.setAutoStartup(true);
|
||||
listenerContainer.setBeanFactory(mockBeanFactory);
|
||||
listenerContainer.afterPropertiesSet();
|
||||
cqListenerContainer.setAutoStartup(true);
|
||||
cqListenerContainer.setBeanFactory(mockBeanFactory);
|
||||
cqListenerContainer.afterPropertiesSet();
|
||||
}
|
||||
finally {
|
||||
assertThat(PoolManagerImpl.getPMI().unregister(mockPool), is(true));
|
||||
assertThat(listenerContainer.isActive(), is(true));
|
||||
assertThat(listenerContainer.isAutoStartup(), is(true));
|
||||
assertThat(listenerContainer.isRunning(), is(false));
|
||||
assertThat(TestUtils.readField("queryService", listenerContainer), is(equalTo(mockQueryService)));
|
||||
assertThat(TestUtils.readField("taskExecutor", listenerContainer), is(instanceOf(Executor.class)));
|
||||
assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue();
|
||||
assertThat(cqListenerContainer.isActive()).isTrue();
|
||||
assertThat(cqListenerContainer.isAutoStartup()).isTrue();
|
||||
assertThat(cqListenerContainer.isRunning()).isFalse();
|
||||
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(1)).isTypeMatch(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class));
|
||||
@@ -109,22 +136,29 @@ public class ContinuousQueryListenerContainerTests {
|
||||
|
||||
@Test
|
||||
public void afterPropertiesSetIsManualStart() {
|
||||
BeanFactory mockBeanFactory = mock(BeanFactory.class);
|
||||
QueryService mockQueryService = mock(QueryService.class);
|
||||
|
||||
Executor mockExecutor = mock(Executor.class);
|
||||
|
||||
PoolManagerImpl poolManagerSpy = spy(PoolManagerImpl.getPMI());
|
||||
|
||||
QueryService mockQueryService = mock(QueryService.class);
|
||||
|
||||
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
|
||||
|
||||
listenerContainer.setAutoStartup(false);
|
||||
listenerContainer.setBeanFactory(mockBeanFactory);
|
||||
listenerContainer.setPoolName("TestPool");
|
||||
listenerContainer.setQueryService(mockQueryService);
|
||||
listenerContainer.afterPropertiesSet();
|
||||
cqListenerContainer.setAutoStartup(false);
|
||||
cqListenerContainer.setBeanFactory(mockBeanFactory);
|
||||
cqListenerContainer.setPoolName("TestPool");
|
||||
cqListenerContainer.setQueryService(mockQueryService);
|
||||
cqListenerContainer.setTaskExecutor(mockExecutor);
|
||||
cqListenerContainer.afterPropertiesSet();
|
||||
|
||||
assertThat(listenerContainer.isActive(), is(true));
|
||||
assertThat(listenerContainer.isAutoStartup(), is(false));
|
||||
assertThat(listenerContainer.isRunning(), is(false));
|
||||
assertThat(cqListenerContainer.isActive()).isTrue();
|
||||
assertThat(cqListenerContainer.isAutoStartup()).isFalse();
|
||||
assertThat(cqListenerContainer.isRunning()).isFalse();
|
||||
assertThat(cqListenerContainer.getBeanFactory()).isSameAs(mockBeanFactory);
|
||||
assertThat(cqListenerContainer.getPoolName()).isEqualTo("TestPool");
|
||||
assertThat(cqListenerContainer.getQueryService()).isSameAs(mockQueryService);
|
||||
assertThat(cqListenerContainer.getTaskExecutor()).isSameAs(mockExecutor);
|
||||
|
||||
verify(mockBeanFactory, never()).containsBean(eq("TestPool"));
|
||||
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
|
||||
@@ -133,146 +167,183 @@ public class ContinuousQueryListenerContainerTests {
|
||||
verifyZeroInteractions(mockQueryService);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void afterPropertiesSetThrowsIllegalStateExceptionWhenQueryServicesIsUninitialized() {
|
||||
exception.expect(IllegalStateException.class);
|
||||
exception.expectCause(is(nullValue(Throwable.class)));
|
||||
exception.expectMessage("QueryService was not properly initialized");
|
||||
|
||||
try {
|
||||
listenerContainer.setPoolName("TestPool");
|
||||
listenerContainer.afterPropertiesSet();
|
||||
}
|
||||
finally {
|
||||
assertThat(listenerContainer.isActive(), is(false));
|
||||
assertThat(listenerContainer.isAutoStartup(), is(true));
|
||||
assertThat(listenerContainer.isRunning(), is(false));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void resolvesToProvidedPoolName() {
|
||||
listenerContainer.setPoolName("TestPool");
|
||||
assertThat(listenerContainer.resolvePoolName(), is(equalTo("TestPool")));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void resolvesToGemFireDefaultPoolName() {
|
||||
listenerContainer.setPoolName(null);
|
||||
assertThat(listenerContainer.resolvePoolName(), is(equalTo(GemfireUtils.DEFAULT_POOL_NAME)));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void resolvesToGemFireDefaultPoolNameWhenBeanFactoryDoesNotContainNamedPool() {
|
||||
BeanFactory mockBeanFactory = mock(BeanFactory.class);
|
||||
|
||||
when(mockBeanFactory.containsBean(anyString())).thenReturn(false);
|
||||
|
||||
listenerContainer.setBeanFactory(mockBeanFactory);
|
||||
listenerContainer.setPoolName(null);
|
||||
|
||||
assertThat(listenerContainer.resolvePoolName(), is(equalTo(GemfireUtils.DEFAULT_POOL_NAME)));
|
||||
|
||||
verify(mockBeanFactory, times(1)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void resolvesToSpringDataGemFireDefaultPoolName() {
|
||||
BeanFactory mockBeanFactory = mock(BeanFactory.class);
|
||||
|
||||
when(mockBeanFactory.containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME))).thenReturn(true);
|
||||
|
||||
listenerContainer.setBeanFactory(mockBeanFactory);
|
||||
listenerContainer.setPoolName(null);
|
||||
|
||||
assertThat(listenerContainer.resolvePoolName(), is(equalTo(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME)));
|
||||
|
||||
verify(mockBeanFactory, times(1)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void eagerlyInitializesNamedPool() {
|
||||
BeanFactory mockBeanFactory = mock(BeanFactory.class);
|
||||
@Test(expected = IllegalStateException.class)
|
||||
public void afterPropertiesSetThrowsIllegalStateExceptionWhenQueryServiceIsUninitialized() {
|
||||
|
||||
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
|
||||
|
||||
listenerContainer.setBeanFactory(mockBeanFactory);
|
||||
|
||||
assertThat(listenerContainer.eagerlyInitializePool("TestPool"), is(equalTo("TestPool")));
|
||||
|
||||
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
|
||||
verify(mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void eagerlyInitializePoolFindsRegisteredPoolByNameInGemFire() {
|
||||
BeanFactory mockBeanFactory = mock(BeanFactory.class);
|
||||
Pool mockPool = mock(Pool.class);
|
||||
|
||||
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
|
||||
when(mockBeanFactory.getBean(eq("TestPool"), eq(Pool.class))).thenThrow(
|
||||
new NoSuchBeanDefinitionException("test"));
|
||||
when(mockPool.getName()).thenReturn("TestPool");
|
||||
|
||||
listenerContainer.setBeanFactory(mockBeanFactory);
|
||||
|
||||
try {
|
||||
PoolManagerImpl.getPMI().register(mockPool);
|
||||
assertThat(listenerContainer.eagerlyInitializePool("TestPool"), is(equalTo("TestPool")));
|
||||
cqListenerContainer.setBeanFactory(mockBeanFactory);
|
||||
cqListenerContainer.setPoolName("TestPool");
|
||||
cqListenerContainer.afterPropertiesSet();
|
||||
}
|
||||
catch (IllegalStateException expected) {
|
||||
|
||||
assertThat(expected).hasMessage("QueryService was not properly initialized");
|
||||
assertThat(expected).hasNoCause();
|
||||
|
||||
throw expected;
|
||||
}
|
||||
finally {
|
||||
assertThat(PoolManagerImpl.getPMI().unregister(mockPool), is(true));
|
||||
verify(mockPool, times(2)).getName();
|
||||
}
|
||||
}
|
||||
assertThat(cqListenerContainer.isActive()).isFalse();
|
||||
assertThat(cqListenerContainer.isAutoStartup()).isTrue();
|
||||
assertThat(cqListenerContainer.isRunning()).isFalse();
|
||||
|
||||
@Test
|
||||
public void eagerlyInitializePoolThrowsIllegalArgumentExceptionCausedByNoSuchBeanDefinitionException() {
|
||||
BeanFactory mockBeanFactory = mock(BeanFactory.class);
|
||||
|
||||
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
|
||||
when(mockBeanFactory.getBean(eq("TestPool"), eq(Pool.class))).thenThrow(
|
||||
new NoSuchBeanDefinitionException("test"));
|
||||
|
||||
try {
|
||||
exception.expect(IllegalArgumentException.class);
|
||||
exception.expectCause(is(nullValue(Throwable.class)));
|
||||
exception.expectMessage("No GemFire Pool with name [TestPool] was found");
|
||||
|
||||
listenerContainer.setBeanFactory(mockBeanFactory);
|
||||
listenerContainer.eagerlyInitializePool("TestPool");
|
||||
}
|
||||
finally {
|
||||
verify(mockBeanFactory, never()).containsBean(eq("TestPool"));
|
||||
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
|
||||
verify(mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void initQueryServiceReturnsProvidedQueryService() {
|
||||
public void resolvePoolNameReturnsConfiguredPoolName() {
|
||||
|
||||
cqListenerContainer.setPoolName("TestPool");
|
||||
|
||||
assertThat(cqListenerContainer.resolvePoolName()).isEqualTo("TestPool");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void resolvePoolNameReturnsGemFireDefaultPoolName() {
|
||||
|
||||
cqListenerContainer.setPoolName(null);
|
||||
|
||||
assertThat(cqListenerContainer.resolvePoolName()).isEqualTo(GemfireUtils.DEFAULT_POOL_NAME);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void resolvePoolNameReturnsGemFireDefaultPoolNameWhenBeanFactoryDoesNotContainNamedPool() {
|
||||
|
||||
when(mockBeanFactory.containsBean(anyString())).thenReturn(false);
|
||||
|
||||
cqListenerContainer.setBeanFactory(mockBeanFactory);
|
||||
cqListenerContainer.setPoolName("");
|
||||
|
||||
assertThat(cqListenerContainer.resolvePoolName()).isEqualTo(GemfireUtils.DEFAULT_POOL_NAME);
|
||||
|
||||
verify(mockBeanFactory, times(1)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void resolvePoolNameReturnsSpringDataGemFireDefaultPoolName() {
|
||||
|
||||
when(mockBeanFactory.containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME))).thenReturn(true);
|
||||
|
||||
cqListenerContainer.setBeanFactory(mockBeanFactory);
|
||||
cqListenerContainer.setPoolName(" ");
|
||||
|
||||
assertThat(cqListenerContainer.resolvePoolName()).isEqualTo(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME);
|
||||
|
||||
verify(mockBeanFactory, times(1)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void eagerlyInitializePoolWithGivenPoolName() {
|
||||
|
||||
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
|
||||
|
||||
cqListenerContainer.setBeanFactory(mockBeanFactory);
|
||||
|
||||
assertThat(cqListenerContainer.eagerlyInitializePool("TestPool")).isEqualTo("TestPool");
|
||||
|
||||
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
|
||||
verify(mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void eagerlyInitializePoolFindsRegisteredPoolWithTheGivenName() {
|
||||
|
||||
Pool mockPool = mock(Pool.class);
|
||||
|
||||
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(false);
|
||||
when(mockPool.getName()).thenReturn("TestPool");
|
||||
|
||||
try {
|
||||
PoolManagerImpl.getPMI().register(mockPool);
|
||||
|
||||
cqListenerContainer.setBeanFactory(mockBeanFactory);
|
||||
|
||||
assertThat(cqListenerContainer.eagerlyInitializePool("TestPool")).isEqualTo("TestPool");
|
||||
}
|
||||
finally {
|
||||
assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue();
|
||||
|
||||
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
|
||||
verify(mockBeanFactory, never()).getBean(eq("TestPool"), eq(Pool.class));
|
||||
verify(mockPool, atLeast(1)).getName();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void eagerlyInitializePoolFindsRegisteredPoolWithTheGivenNameWhenBeanFactoryGetBeanThrowsBeansException() {
|
||||
|
||||
Pool mockPool = mock(Pool.class);
|
||||
|
||||
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
|
||||
when(mockBeanFactory.getBean(eq("TestPool"), eq(Pool.class)))
|
||||
.thenThrow(new NoSuchBeanDefinitionException("TEST"));
|
||||
when(mockPool.getName()).thenReturn("TestPool");
|
||||
|
||||
cqListenerContainer.setBeanFactory(mockBeanFactory);
|
||||
|
||||
try {
|
||||
PoolManagerImpl.getPMI().register(mockPool);
|
||||
|
||||
assertThat(cqListenerContainer.eagerlyInitializePool("TestPool")).isEqualTo("TestPool");
|
||||
}
|
||||
finally {
|
||||
assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue();
|
||||
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
@Test(expected = IllegalArgumentException.class)
|
||||
public void eagerlyInitializePoolThrowsIllegalArgumentExceptionCausedByNoPoolWithGivenName() {
|
||||
|
||||
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(false);
|
||||
|
||||
try {
|
||||
cqListenerContainer.setBeanFactory(mockBeanFactory);
|
||||
cqListenerContainer.eagerlyInitializePool("TestPool");
|
||||
}
|
||||
catch (IllegalArgumentException expected) {
|
||||
|
||||
assertThat(expected).hasMessage("No Pool with name [TestPool] was found");
|
||||
assertThat(expected).hasNoCause();
|
||||
|
||||
throw expected;
|
||||
}
|
||||
finally {
|
||||
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
|
||||
verify(mockBeanFactory, never()).getBean(eq("TestPool"), eq(Pool.class));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void initQueryServiceReturnsConfiguredQueryService() {
|
||||
|
||||
QueryService mockQueryService = mock(QueryService.class);
|
||||
|
||||
listenerContainer.setQueryService(mockQueryService);
|
||||
cqListenerContainer.setQueryService(mockQueryService);
|
||||
|
||||
assertThat(listenerContainer.initQueryService(null), is(sameInstance(mockQueryService)));
|
||||
assertThat(cqListenerContainer.initQueryService("TestPool")).isSameAs(mockQueryService);
|
||||
|
||||
verifyZeroInteractions(mockQueryService);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void initializesQueryServiceFromContainer() {
|
||||
QueryService mockQueryService = mock(QueryService.class);
|
||||
|
||||
listenerContainer.setQueryService(mockQueryService);
|
||||
|
||||
assertThat(listenerContainer.initQueryService("TestPool"), is(equalTo(mockQueryService)));
|
||||
|
||||
verifyZeroInteractions(mockQueryService);
|
||||
public void initQueryServiceReturnsNull() {
|
||||
assertThat(cqListenerContainer.initQueryService(null)).isNull();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void initializesQueryServiceFromPool() {
|
||||
|
||||
Pool mockPool = mock(Pool.class);
|
||||
|
||||
QueryService mockQueryService = mock(QueryService.class);
|
||||
|
||||
when(mockPool.getName()).thenReturn("TestPool");
|
||||
@@ -280,19 +351,22 @@ public class ContinuousQueryListenerContainerTests {
|
||||
|
||||
try {
|
||||
PoolManagerImpl.getPMI().register(mockPool);
|
||||
listenerContainer.setQueryService(null);
|
||||
assertThat(listenerContainer.initQueryService("TestPool"), is(equalTo(mockQueryService)));
|
||||
|
||||
assertThat(cqListenerContainer.getQueryService()).isNull();
|
||||
assertThat(cqListenerContainer.initQueryService("TestPool")).isEqualTo(mockQueryService);
|
||||
}
|
||||
finally {
|
||||
assertThat(PoolManagerImpl.getPMI().unregister(mockPool), is(true));
|
||||
verify(mockPool, times(2)).getName();
|
||||
assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue();
|
||||
|
||||
verify(mockPool, atLeast(1)).getName();
|
||||
verify(mockPool, times(1)).getQueryService();
|
||||
verifyZeroInteractions(mockQueryService);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void initializesQueryServiceFromPoolInThePresenceOfAProvidedQueryService() {
|
||||
public void initializesQueryServiceFromPoolIgnoresConfiguredQueryService() {
|
||||
|
||||
Pool mockPool = mock(Pool.class);
|
||||
|
||||
QueryService mockQueryServiceOne = mock(QueryService.class);
|
||||
@@ -303,12 +377,16 @@ public class ContinuousQueryListenerContainerTests {
|
||||
|
||||
try {
|
||||
PoolManagerImpl.getPMI().register(mockPool);
|
||||
listenerContainer.setQueryService(mockQueryServiceTwo);
|
||||
assertThat(listenerContainer.initQueryService("TestPool"), is(equalTo(mockQueryServiceOne)));
|
||||
|
||||
cqListenerContainer.setQueryService(mockQueryServiceTwo);
|
||||
|
||||
assertThat(cqListenerContainer.getQueryService()).isSameAs(mockQueryServiceTwo);
|
||||
assertThat(cqListenerContainer.initQueryService("TestPool")).isEqualTo(mockQueryServiceOne);
|
||||
}
|
||||
finally {
|
||||
assertThat(PoolManagerImpl.getPMI().unregister(mockPool), is(true));
|
||||
verify(mockPool, times(2)).getName();
|
||||
assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue();
|
||||
|
||||
verify(mockPool, atLeast(1)).getName();
|
||||
verify(mockPool, times(1)).getQueryService();
|
||||
verifyZeroInteractions(mockQueryServiceOne);
|
||||
verifyZeroInteractions(mockQueryServiceTwo);
|
||||
@@ -317,45 +395,374 @@ public class ContinuousQueryListenerContainerTests {
|
||||
|
||||
@Test
|
||||
public void initExecutorReturnsProvidedExecutor() {
|
||||
|
||||
Executor mockExecutor = mock(Executor.class);
|
||||
|
||||
listenerContainer.setTaskExecutor(mockExecutor);
|
||||
cqListenerContainer.setTaskExecutor(mockExecutor);
|
||||
|
||||
assertThat(listenerContainer.initExecutor(), is(sameInstance(mockExecutor)));
|
||||
assertThat(cqListenerContainer.initExecutor()).isSameAs(mockExecutor);
|
||||
|
||||
verifyZeroInteractions(mockExecutor);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void initializesDefaultTaskExecutor() {
|
||||
assertThat(listenerContainer.initExecutor(), is(instanceOf(Executor.class)));
|
||||
assertThat(cqListenerContainer.getTaskExecutor()).isNull();
|
||||
assertThat(cqListenerContainer.initExecutor()).isInstanceOf(Executor.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void setAndGetAutoStartup() {
|
||||
|
||||
assertThat(cqListenerContainer.isAutoStartup()).isTrue();
|
||||
|
||||
cqListenerContainer.setAutoStartup(false);
|
||||
|
||||
assertThat(cqListenerContainer.isAutoStartup()).isFalse();
|
||||
|
||||
cqListenerContainer.setAutoStartup(true);
|
||||
|
||||
assertThat(cqListenerContainer.isAutoStartup()).isTrue();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void setCacheSetsQueryService() {
|
||||
|
||||
QueryService mockQueryService = mock(QueryService.class);
|
||||
|
||||
RegionService mockRegionService = mock(RegionService.class);
|
||||
|
||||
when(mockRegionService.getQueryService()).thenReturn(mockQueryService);
|
||||
|
||||
listenerContainer.setCache(mockRegionService);
|
||||
cqListenerContainer.setCache(mockRegionService);
|
||||
|
||||
assertThat(listenerContainer.initQueryService(null), is(equalTo(mockQueryService)));
|
||||
assertThat(cqListenerContainer.initQueryService(null)).isEqualTo(mockQueryService);
|
||||
|
||||
verify(mockRegionService, times(1)).getQueryService();
|
||||
verifyZeroInteractions(mockQueryService);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void setAndGetAutoStartup() {
|
||||
assertThat(listenerContainer.isAutoStartup(), is(true));
|
||||
public void addListenerCreatesCqQueryAndExecutesQueryWhenRunning() throws Exception {
|
||||
|
||||
listenerContainer.setAutoStartup(false);
|
||||
ContinuousQueryListener mockListener = mock(ContinuousQueryListener.class);
|
||||
|
||||
assertThat(listenerContainer.isAutoStartup(), is(false));
|
||||
ContinuousQueryDefinition definition =
|
||||
new ContinuousQueryDefinition("TestQuery", "SELECT * FROM /Utilization u WHERE u.value > 100", mockListener);
|
||||
|
||||
listenerContainer.setAutoStartup(true);
|
||||
CqQuery mockQuery = mock(CqQuery.class);
|
||||
|
||||
assertThat(listenerContainer.isAutoStartup(), is(true));
|
||||
doReturn(mockQuery).when(cqListenerContainer).addContinuousQuery(eq(definition));
|
||||
when(cqListenerContainer.isRunning()).thenReturn(true);
|
||||
|
||||
cqListenerContainer.addListener(definition);
|
||||
|
||||
verify(cqListenerContainer, times(1)).addContinuousQuery(eq(definition));
|
||||
verify(cqListenerContainer, times(1)).isRunning();
|
||||
verify(mockQuery, times(1)).execute();
|
||||
}
|
||||
|
||||
@Test(expected = GemfireQueryException.class)
|
||||
public void addContinuousQueryThrowsQueryException() throws Exception {
|
||||
|
||||
QueryService mockQueryService = mock(QueryService.class);
|
||||
|
||||
when(mockQueryService.newCq(anyString(), any(CqAttributes.class), anyBoolean()))
|
||||
.thenThrow(new CqException("TEST"));
|
||||
|
||||
ContinuousQueryListener mockListener = mock(ContinuousQueryListener.class);
|
||||
|
||||
ContinuousQueryDefinition definition =
|
||||
new ContinuousQueryDefinition("SELECT * FROM /Utilization u WHERE u.value > 100",
|
||||
mockListener, false);
|
||||
|
||||
cqListenerContainer.setQueryService(mockQueryService);
|
||||
|
||||
try {
|
||||
cqListenerContainer.addContinuousQuery(definition);
|
||||
}
|
||||
catch (GemfireQueryException expected) {
|
||||
|
||||
assertThat(expected).hasMessageStartingWith("Unable to create query [SELECT * FROM /Utilization u WHERE u.value > 100]");
|
||||
assertThat(expected).hasCauseInstanceOf(QueryException.class);
|
||||
assertThat(expected.getCause()).hasMessage("TEST");
|
||||
assertThat(expected.getCause()).hasNoCause();
|
||||
|
||||
throw expected;
|
||||
|
||||
}
|
||||
finally {
|
||||
assertThat(cqListenerContainer.getContinuousQueries()).isEmpty();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void addManagedNamedContinuousQuery() throws Exception {
|
||||
|
||||
QueryService mockQueryService = mock(QueryService.class);
|
||||
|
||||
when(mockQueryService.newCq(anyString(), anyString(), any(CqAttributes.class), anyBoolean()))
|
||||
.thenAnswer(invocation -> mockCqQuery(invocation.getArgument(0), invocation.getArgument(1),
|
||||
invocation.getArgument(2), invocation.getArgument(3)));
|
||||
|
||||
ContinuousQueryListener mockListener = mock(ContinuousQueryListener.class);
|
||||
|
||||
ContinuousQueryDefinition definition =
|
||||
new ContinuousQueryDefinition("TestQuery", "SELECT * FROM /Utilization u WHERE u.value > 100",
|
||||
mockListener, true);
|
||||
|
||||
cqListenerContainer.setQueryService(mockQueryService);
|
||||
|
||||
CqQuery query = cqListenerContainer.addContinuousQuery(definition);
|
||||
|
||||
assertThat(query).isNotNull();
|
||||
assertThat(query.isDurable()).isTrue();
|
||||
assertThat(query.getName()).isEqualTo("TestQuery");
|
||||
assertThat(query.getQueryString()).isEqualTo("SELECT * FROM /Utilization u WHERE u.value > 100");
|
||||
assertThat(query.isRunning()).isFalse();
|
||||
|
||||
CqAttributes attributes = query.getCqAttributes();
|
||||
|
||||
assertThat(attributes).isNotNull();
|
||||
assertThat(attributes.getCqListener()).isInstanceOf(ContinuousQueryListenerContainer.EventDispatcherAdapter.class);
|
||||
|
||||
ContinuousQueryListenerContainer.EventDispatcherAdapter eventDispatcherAdapter =
|
||||
(ContinuousQueryListenerContainer.EventDispatcherAdapter) attributes.getCqListener();
|
||||
|
||||
assertThat(eventDispatcherAdapter.getListener()).isEqualTo(mockListener);
|
||||
assertThat(cqListenerContainer.getContinuousQueries().peek()).isEqualTo(query);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void addManagedUnnamedContinuousQuery() throws Exception {
|
||||
|
||||
QueryService mockQueryService = mock(QueryService.class);
|
||||
|
||||
when(mockQueryService.newCq(anyString(), any(CqAttributes.class), anyBoolean()))
|
||||
.thenAnswer(invocation -> mockCqQuery(null, invocation.getArgument(0), invocation.getArgument(1),
|
||||
invocation.getArgument(2)));
|
||||
|
||||
ContinuousQueryListener mockListener = mock(ContinuousQueryListener.class);
|
||||
|
||||
ContinuousQueryDefinition definition =
|
||||
new ContinuousQueryDefinition("SELECT * FROM /Utilization u WHERE u.value > 100",
|
||||
mockListener, false);
|
||||
|
||||
cqListenerContainer.setQueryService(mockQueryService);
|
||||
|
||||
CqQuery query = cqListenerContainer.addContinuousQuery(definition);
|
||||
|
||||
assertThat(query).isNotNull();
|
||||
assertThat(query.isDurable()).isFalse();
|
||||
assertThat(query.getName()).isNull();
|
||||
assertThat(query.getQueryString()).isEqualTo("SELECT * FROM /Utilization u WHERE u.value > 100");
|
||||
assertThat(query.isRunning()).isFalse();
|
||||
|
||||
CqAttributes attributes = query.getCqAttributes();
|
||||
|
||||
assertThat(attributes).isNotNull();
|
||||
assertThat(attributes.getCqListener()).isInstanceOf(ContinuousQueryListenerContainer.EventDispatcherAdapter.class);
|
||||
|
||||
ContinuousQueryListenerContainer.EventDispatcherAdapter eventDispatcherAdapter =
|
||||
(ContinuousQueryListenerContainer.EventDispatcherAdapter) attributes.getCqListener();
|
||||
|
||||
assertThat(eventDispatcherAdapter.getListener()).isEqualTo(mockListener);
|
||||
assertThat(cqListenerContainer.getContinuousQueries().peek()).isEqualTo(query);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void cqListenerContainerStartsWhenNotRunning() throws Exception {
|
||||
|
||||
CqQuery mockQueryOne = mock(CqQuery.class);
|
||||
CqQuery mockQueryTwo = mock(CqQuery.class);
|
||||
|
||||
cqListenerContainer.getContinuousQueries().add(mockQueryOne);
|
||||
cqListenerContainer.getContinuousQueries().add(mockQueryTwo);
|
||||
|
||||
assertThat(cqListenerContainer.isRunning()).isFalse();
|
||||
|
||||
cqListenerContainer.start();
|
||||
|
||||
assertThat(cqListenerContainer.isRunning()).isTrue();
|
||||
|
||||
verify(mockQueryOne, times(1)).execute();
|
||||
verify(mockQueryTwo, times(1)).execute();
|
||||
}
|
||||
|
||||
@Test(expected = GemfireQueryException.class)
|
||||
public void cqListenerContainerStartHandlesCqException() throws Exception {
|
||||
|
||||
CqQuery mockQueryOne = mock(CqQuery.class);
|
||||
CqQuery mockQueryTwo = mock(CqQuery.class);
|
||||
|
||||
CqState mockQueryState = mock(CqState.class);
|
||||
|
||||
cqListenerContainer.getContinuousQueries().add(mockQueryOne);
|
||||
cqListenerContainer.getContinuousQueries().add(mockQueryTwo);
|
||||
|
||||
when(mockQueryOne.getName()).thenReturn("ONE");
|
||||
when(mockQueryOne.getState()).thenReturn(mockQueryState);
|
||||
when(mockQueryState.toString()).thenReturn("FAILED");
|
||||
doThrow(new CqException("ONE")).when(mockQueryOne).execute();
|
||||
|
||||
try {
|
||||
cqListenerContainer.start();
|
||||
}
|
||||
catch (GemfireQueryException cause) {
|
||||
|
||||
assertThat(cause).hasMessageStartingWith("Could not execute query [ONE]; state is [FAILED]");
|
||||
assertThat(cause).hasCauseInstanceOf(CqException.class);
|
||||
assertThat(cause.getCause()).hasMessage("ONE");
|
||||
assertThat(cause.getCause()).hasNoCause();
|
||||
|
||||
throw cause;
|
||||
}
|
||||
finally {
|
||||
assertThat(cqListenerContainer.isRunning()).isFalse();
|
||||
|
||||
verify(mockQueryOne, times(1)).execute();
|
||||
verify(mockQueryOne, times(1)).getName();
|
||||
verify(mockQueryOne, times(1)).getState();
|
||||
verify(mockQueryTwo, never()).execute();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void cqListenerContainerDoesNotStartWhenAlreadyRunning() {
|
||||
|
||||
when(cqListenerContainer.isRunning()).thenReturn(true);
|
||||
|
||||
cqListenerContainer.start();
|
||||
|
||||
verify(cqListenerContainer, never()).doStart();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void dispatchEventNotifiesListenerOfCqEvent() {
|
||||
|
||||
Executor mockExecutor = mock(Executor.class);
|
||||
|
||||
doAnswer(invocation -> {
|
||||
invocation.<Runnable>getArgument(0).run();
|
||||
return null;
|
||||
}).when(mockExecutor).execute(any());
|
||||
|
||||
ContinuousQueryListener mockListener = mock(ContinuousQueryListener.class);
|
||||
|
||||
CqEvent mockEvent = mock(CqEvent.class);
|
||||
|
||||
cqListenerContainer.setTaskExecutor(mockExecutor);
|
||||
cqListenerContainer.dispatchEvent(mockListener, mockEvent);
|
||||
|
||||
verify(mockExecutor, times(1)).execute(isA(Runnable.class));
|
||||
verify(mockListener, times(1)).onEvent(eq(mockEvent));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void dispatchEventInvokesConfiguredErrorHandlerOnListenerException() {
|
||||
|
||||
RuntimeException expectedCause = new RuntimeException("TEST");
|
||||
|
||||
Executor mockExecutor = mock(Executor.class);
|
||||
|
||||
doAnswer(invocation -> {
|
||||
invocation.<Runnable>getArgument(0).run();
|
||||
return null;
|
||||
}).when(mockExecutor).execute(any());
|
||||
|
||||
ErrorHandler mockErrorHandler = mock(ErrorHandler.class);
|
||||
|
||||
ContinuousQueryListener mockListener = mock(ContinuousQueryListener.class);
|
||||
|
||||
doThrow(expectedCause).when(mockListener).onEvent(any());
|
||||
|
||||
CqEvent mockEvent = mock(CqEvent.class);
|
||||
|
||||
cqListenerContainer.setErrorHandler(mockErrorHandler);
|
||||
cqListenerContainer.setTaskExecutor(mockExecutor);
|
||||
|
||||
doReturn(true).when(cqListenerContainer).isActive();
|
||||
|
||||
cqListenerContainer.dispatchEvent(mockListener, mockEvent);
|
||||
|
||||
verify(mockExecutor, times(1)).execute(isA(Runnable.class));
|
||||
verify(mockListener, times(1)).onEvent(eq(mockEvent));
|
||||
verify(mockErrorHandler, times(1)).handleError(eq(expectedCause));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void stopStopsCqsCallsRunnableHandlesExceptionsOnCqQueryStopWhenRunning() throws Exception {
|
||||
|
||||
CqQuery mockQueryOne = mock(CqQuery.class);
|
||||
CqQuery mockQueryTwo = mock(CqQuery.class);
|
||||
CqQuery mockQueryThree = mock(CqQuery.class);
|
||||
|
||||
doThrow(new CqException("TWO")).when(mockQueryTwo).stop();
|
||||
|
||||
Runnable mockRunnable = mock(Runnable.class);
|
||||
|
||||
cqListenerContainer.getContinuousQueries().add(mockQueryOne);
|
||||
cqListenerContainer.getContinuousQueries().add(mockQueryTwo);
|
||||
cqListenerContainer.getContinuousQueries().add(mockQueryThree);
|
||||
|
||||
doReturn(true).when(cqListenerContainer).isRunning();
|
||||
|
||||
cqListenerContainer.stop(mockRunnable);
|
||||
|
||||
verify(cqListenerContainer, times(1)).stop();
|
||||
verify(mockQueryOne, times(1)).stop();
|
||||
verify(mockQueryTwo, times(1)).stop();
|
||||
verify(mockQueryThree, times(1)).stop();
|
||||
verify(mockRunnable, times(1)).run();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void stopDoesNothingWhenContainerIsNotRunning() {
|
||||
|
||||
when(cqListenerContainer.isRunning()).thenReturn(false);
|
||||
|
||||
cqListenerContainer.stop();
|
||||
|
||||
verify(cqListenerContainer, never()).doStop();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void destroyIsSuccessful() throws Exception {
|
||||
|
||||
CqQuery mockQueryOne = mock(CqQuery.class);
|
||||
CqQuery mockQueryTwo = mock(CqQuery.class);
|
||||
CqQuery mockQueryThree = mock(CqQuery.class);
|
||||
CqQuery mockQueryFour = mock(CqQuery.class);
|
||||
|
||||
when(mockQueryTwo.isClosed()).thenReturn(true);
|
||||
doThrow(new RuntimeException("THREE")).when(mockQueryThree).close();
|
||||
|
||||
List<CqQuery> queries = Arrays.asList(mockQueryOne, mockQueryTwo, mockQueryThree, mockQueryFour);
|
||||
|
||||
cqListenerContainer.getContinuousQueries().addAll(queries);
|
||||
|
||||
DisposableExecutorBean mockDisposableExecutorBean = mock(DisposableExecutorBean.class);
|
||||
|
||||
doReturn(mockDisposableExecutorBean).when(cqListenerContainer).createDefaultTaskExecutor();
|
||||
|
||||
assertThat(cqListenerContainer.getContinuousQueries()).hasSize(queries.size());
|
||||
|
||||
cqListenerContainer.initExecutor();
|
||||
cqListenerContainer.destroy();
|
||||
|
||||
assertThat(cqListenerContainer.getContinuousQueries()).isEmpty();
|
||||
|
||||
verify(mockQueryOne, times(1)).isClosed();
|
||||
verify(mockQueryOne, times(1)).close();
|
||||
verify(mockQueryTwo, times(1)).isClosed();
|
||||
verify(mockQueryTwo, never()).close();
|
||||
verify(mockQueryThree, times(1)).isClosed();
|
||||
verify(mockQueryThree, times(1)).close();
|
||||
verify(mockQueryFour, times(1)).isClosed();
|
||||
verify(mockQueryFour, times(1)).close();
|
||||
verify(mockDisposableExecutorBean, times(1)).destroy();
|
||||
}
|
||||
|
||||
interface DisposableExecutorBean extends Executor, DisposableBean { }
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user