diff --git a/src/main/java/org/springframework/data/gemfire/client/support/DefaultableDelegatingPoolAdapter.java b/src/main/java/org/springframework/data/gemfire/client/support/DefaultableDelegatingPoolAdapter.java index 7789c669..a189e516 100644 --- a/src/main/java/org/springframework/data/gemfire/client/support/DefaultableDelegatingPoolAdapter.java +++ b/src/main/java/org/springframework/data/gemfire/client/support/DefaultableDelegatingPoolAdapter.java @@ -28,10 +28,12 @@ import com.gemstone.gemfire.cache.client.Pool; import com.gemstone.gemfire.cache.query.QueryService; /** - * The DefaultableDelegatingPoolAdapter class... + * The DefaultableDelegatingPoolAdapter class is a wrapper class around Pool allowing default configuration property + * values to be providing in the case that the Pool's setting were null. * * @author John Blum - * @since 1.0.0 + * @see com.gemstone.gemfire.cache.client.Pool + * @since 1.8.0 */ @SuppressWarnings("unused") public abstract class DefaultableDelegatingPoolAdapter { @@ -201,8 +203,12 @@ public abstract class DefaultableDelegatingPoolAdapter { } /* (non-Javadoc) */ - public QueryService getQueryService() { - return getDelegate().getQueryService(); + public QueryService getQueryService(QueryService defaultQueryService) { + return defaultIfNull(defaultQueryService, new ValueProvider() { + @Override public QueryService getValue() { + return getDelegate().getQueryService(); + } + }); } /* (non-Javadoc) */ diff --git a/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java b/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java index 9fc30044..1619c832 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java +++ b/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java @@ -116,7 +116,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN } /* (non-Javadoc) */ - private String resolvePoolName() { + String resolvePoolName() { String poolName = this.poolName; if (!StringUtils.hasText(poolName)) { @@ -129,29 +129,37 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN } /* (non-Javadoc) */ - private String eagerlyInitializePool(String poolName) { + String eagerlyInitializePool(String poolName) { try { if (beanFactory != null && beanFactory.isTypeMatch(poolName, Pool.class)) { beanFactory.getBean(poolName, Pool.class); } } catch (BeansException ignore) { - Assert.notNull(PoolManager.find(poolName), String.format("No GemFire Pool with name [%1$s] was found.", + Assert.notNull(PoolManager.find(poolName), String.format("No GemFire Pool with name [%1$s] was found", poolName)); } return poolName; } - private void initQueryService(String poolName) { - queryService = PoolManager.find(poolName).getQueryService(); + /* (non-Javadoc) */ + QueryService initQueryService(String poolName) { + if (queryService == null) { + queryService = PoolManager.find(poolName).getQueryService(); + } + + return queryService; } - private void initExecutor() { + /* (non-Javadoc) */ + Executor initExecutor() { if (taskExecutor == null) { - manageExecutor = true; taskExecutor = createDefaultTaskExecutor(); + manageExecutor = true; } + + return taskExecutor; } /** @@ -204,11 +212,11 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN cq.execute(); } catch (QueryException ex) { - throw new GemfireQueryException(String.format("Could not execute query '%1$s'; state is '%2$s'.", + throw new GemfireQueryException(String.format("Could not execute query [%1$s]; state is [%2$s].", cq.getName(), cq.getState()), ex); } catch (RuntimeException ex) { - throw new GemfireQueryException(String.format("Could not execute query '%1$s'; state is '%2$s'.", + throw new GemfireQueryException(String.format("Could not execute query [%1$s]; state is [%2$s].", cq.getName(), cq.getState()), ex); } } @@ -286,12 +294,12 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN } /** - * Determines whether the container has be started and is currently running. + * Sets whether the CQ listener container should automatically start on startup. * - * @return a boolean value indicating whether the container has been started and is currently running. + * @param autoStartup a boolean value indicating whether this CQ listener container should automatically start. */ - public synchronized boolean isRunning() { - return running; + public void setAutoStartup(final boolean autoStartup) { + this.autoStartup = autoStartup; } /** @@ -305,12 +313,12 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN } /** - * Sets whether the CQ listener container should automatically start on startup. + * Determines whether the container has be started and is currently running. * - * @param autoStartup a boolean value indicating whether this CQ listener container should automatically start. + * @return a boolean value indicating whether the container has been started and is currently running. */ - public void setAutoStartup(final boolean autoStartup) { - this.autoStartup = autoStartup; + public synchronized boolean isRunning() { + return running; } /** @@ -357,6 +365,15 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN this.errorHandler = errorHandler; } + /** + * Sets the phase in which this CQ listener container will start in the Spring container. + * + * @param phase the phase value of this CQ listener container. + */ + public void setPhase(final int phase) { + this.phase = phase; + } + /** * Gets the phase in which this CQ listener container will start in the Spring container. * @@ -367,15 +384,6 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN return phase; } - /** - * Sets the phase in which this CQ listener container will start in the Spring container. - * - * @param phase the phase value of this CQ listener container. - */ - public void setPhase(final int phase) { - this.phase = phase; - } - /** * Set the name of the {@link Pool} used for performing the queries by this container. * diff --git a/src/main/java/org/springframework/data/gemfire/listener/GemfireListenerExecutionFailedException.java b/src/main/java/org/springframework/data/gemfire/listener/GemfireListenerExecutionFailedException.java index bd867463..ddee91a2 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/GemfireListenerExecutionFailedException.java +++ b/src/main/java/org/springframework/data/gemfire/listener/GemfireListenerExecutionFailedException.java @@ -25,7 +25,7 @@ import org.springframework.data.gemfire.listener.adapter.ContinuousQueryListener * @author Costin Leau * @see ContinuousQueryListenerAdapter */ -@SuppressWarnings("serial") +@SuppressWarnings({ "serial", "unused" }) public class GemfireListenerExecutionFailedException extends InvalidDataAccessApiUsageException { /** diff --git a/src/main/java/org/springframework/data/gemfire/listener/adapter/ContinuousQueryListenerAdapter.java b/src/main/java/org/springframework/data/gemfire/listener/adapter/ContinuousQueryListenerAdapter.java index 54c06651..907d3074 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/adapter/ContinuousQueryListenerAdapter.java +++ b/src/main/java/org/springframework/data/gemfire/listener/adapter/ContinuousQueryListenerAdapter.java @@ -74,7 +74,7 @@ import com.gemstone.gemfire.cache.query.CqQuery; public class ContinuousQueryListenerAdapter implements ContinuousQueryListener { // Out-of-the-box value for the default listener handler method "handleEvent". - public static final String ORIGINAL_DEFAULT_LISTENER_METHOD = "handleEvent"; + public static final String DEFAULT_LISTENER_METHOD_NAME = "handleEvent"; protected final Log logger = LogFactory.getLog(getClass()); @@ -82,7 +82,7 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener { private Object delegate; - private String defaultListenerMethod = ORIGINAL_DEFAULT_LISTENER_METHOD; + private String defaultListenerMethod = DEFAULT_LISTENER_METHOD_NAME; /** * Create a new {@link ContinuousQueryListenerAdapter} with default settings. @@ -110,7 +110,7 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener { * @param delegate delegate object */ public void setDelegate(Object delegate) { - Assert.notNull(delegate, "The delegate must not be null."); + Assert.notNull(delegate, "'delegate' must not be null"); this.delegate = delegate; this.invoker = null; } @@ -126,7 +126,7 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener { /** * Specify the name of the default listener method to delegate to in the case where no specific listener method - * has been determined. Out-of-the-box value is {@link #ORIGINAL_DEFAULT_LISTENER_METHOD "handleEvent}. + * has been determined. Out-of-the-box value is {@link #DEFAULT_LISTENER_METHOD_NAME "handleEvent}. * * @param defaultListenerMethod the name of the default listener method to invoke. * @see #getListenerMethodName @@ -157,30 +157,28 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener { public void onEvent(CqEvent event) { try { // Check whether the delegate is a ContinuousQueryListener implementation itself. - // In that case, the adapter will simply act as a pass-through. - if (delegate != this) { - if (delegate instanceof ContinuousQueryListener) { - ((ContinuousQueryListener) delegate).onEvent(event); - return; + // If so, this adapter will simply act as a pass-through. + if (delegate != this && delegate instanceof ContinuousQueryListener) { + ((ContinuousQueryListener) delegate).onEvent(event); + } + // Else... find the listener handler method reflectively. + else { + String methodName = getListenerMethodName(event); + + if (methodName == null) { + throw new InvalidDataAccessApiUsageException("No default listener method specified." + + " Either specify a non-null value for the 'defaultListenerMethod' property" + + " or override the 'getListenerMethodName' method."); } + + invoker = (invoker != null ? invoker : new MethodInvoker(delegate, methodName)); + + invokeListenerMethod(event, methodName); } - // Other case... find the listener handler method reflectively. - String methodName = getListenerMethodName(event); - - if (methodName == null) { - throw new InvalidDataAccessApiUsageException("No default listener method specified." - + " Either specify a non-null value for the 'defaultListenerMethod' property" - + " or override the 'getListenerMethodName' method."); - } - - if (invoker == null) { - invoker = new MethodInvoker(delegate, methodName); - } - - invokeListenerMethod(event, methodName); - } catch (Throwable th) { - handleListenerException(th); + } + catch (Throwable cause) { + handleListenerException(cause); } } @@ -201,10 +199,10 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener { /** * Handle the given exception that arose during listener execution. * The default implementation logs the exception at error level. - * @param ex the exception to handle + * @param cause the exception to handle */ - protected void handleListenerException(Throwable ex) { - logger.error("Listener execution failed...", ex); + protected void handleListenerException(Throwable cause) { + logger.error("Listener execution failed...", cause); } /** @@ -223,117 +221,123 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener { } else { throw new GemfireListenerExecutionFailedException( - String.format("Listener method '%1$s' threw exception...", methodName), e.getTargetException()); + String.format("Listener method [%1$s] threw Exception...", methodName), e.getTargetException()); } } catch (Throwable e) { throw new GemfireListenerExecutionFailedException( - String.format("Failed to invoke target listener method '%1$s'", methodName), e); + String.format("Failed to invoke the target listener method [%1$s]", methodName), e); } } private class MethodInvoker { + private final Object delegate; + List methods; MethodInvoker(Object delegate, final String methodName) { - this.delegate = delegate; - Class c = delegate.getClass(); + this.delegate = delegate; methods = new ArrayList(); ReflectionUtils.doWithMethods(c, new MethodCallback() { - public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { ReflectionUtils.makeAccessible(method); methods.add(method); } - }, new MethodFilter() { public boolean matches(Method method) { - if (Modifier.isPublic(method.getModifiers()) && methodName.equals(method.getName())) { - - // check out the arguments - Class[] parameterTypes = method.getParameterTypes(); - int objects = 0; - int operations = 0; - - if (parameterTypes.length > 0) { - for (Class paramType : parameterTypes) { - - if (Object.class.equals(paramType)) { - objects++; - if (objects > 2) { - return false; - } - } - else if (Operation.class.equals(paramType)) { - operations++; - if (operations > 2) { - return false; - } - } - else if (CqEvent.class.equals(paramType)) { - } - else if (Throwable.class.equals(paramType)) { - } - else if (byte[].class.equals(paramType)) { - } - else if (CqQuery.class.equals(paramType)) { - } - else { - return false; - } - } - return true; - } - } - return false; + return isValidEventMethodSignature(method, methodName); } }); - Assert.isTrue(!methods.isEmpty(), "Cannot find a suitable method named [" + c.getName() + "#" + methodName - + "] - is the method public and has the proper arguments?"); + Assert.isTrue(!methods.isEmpty(), String.format( + "Cannot find a suitable method named [%1$s#%2$s] - is the method public and does it have the proper arguments?", + c.getName(), methodName)); + } + + @SuppressWarnings("all") + boolean isValidEventMethodSignature(Method method, String methodName) { + if (Modifier.isPublic(method.getModifiers()) && methodName.equals(method.getName())) { + Class[] parameterTypes = method.getParameterTypes(); + + int objects = 0; + int operations = 0; + + if (parameterTypes.length > 0) { + for (Class parameterType : parameterTypes) { + if (Object.class.equals(parameterType)) { + if (++objects > 2) { + return false; + } + } + else if (Operation.class.equals(parameterType)) { + if (++operations > 2) { + return false; + } + } + else if (byte[].class.equals(parameterType)) { + } + else if (CqEvent.class.equals(parameterType)) { + } + else if (CqQuery.class.equals(parameterType)) { + } + else if (Throwable.class.equals(parameterType)) { + } + else { + return false; + } + } + + return true; + } + } + + return false; } void invoke(CqEvent event) throws InvocationTargetException, IllegalAccessException { - - for (Method m : methods) { - Class[] types = m.getParameterTypes(); - Object[] args = new Object[types.length]; - - boolean value = false; - boolean query = false; - - for (int i = 0; i < types.length; i++) { - Class paramType = types[i]; - - if (Object.class.equals(paramType)) { - args[i] = (!value ? event.getKey() : event.getNewValue()); - value = true; - } - else if (Operation.class.equals(paramType)) { - args[i] = (!query ? event.getBaseOperation() : event.getQueryOperation()); - query = true; - } - else if (CqEvent.class.equals(paramType)) { - args[i] = event; - } - else if (Throwable.class.equals(paramType)) { - args[i] = event.getThrowable(); - } - else if (byte[].class.equals(paramType)) { - args[i] = event.getDeltaValue(); - } - else if (CqQuery.class.equals(paramType)) { - args[i] = event.getCq(); - } - } - - m.invoke(delegate, args); + for (Method method : methods) { + method.invoke(delegate, getMethodArguments(method, event)); } } + + Object[] getMethodArguments(Method method, CqEvent event) { + Class[] parameterTypes = method.getParameterTypes(); + Object[] args = new Object[parameterTypes.length]; + + boolean query = false; + boolean value = false; + + for (int index = 0; index < parameterTypes.length; index++) { + Class parameterType = parameterTypes[index]; + + if (Object.class.equals(parameterType)) { + args[index] = (value ? event.getNewValue() : event.getKey()); + value = true; + } + else if (Operation.class.equals(parameterType)) { + args[index] = (query ? event.getQueryOperation() : event.getBaseOperation()); + query = true; + } + else if (byte[].class.equals(parameterType)) { + args[index] = event.getDeltaValue(); + } + else if (CqEvent.class.equals(parameterType)) { + args[index] = event; + } + else if (CqQuery.class.equals(parameterType)) { + args[index] = event.getCq(); + } + else if (Throwable.class.equals(parameterType)) { + args[index] = event.getThrowable(); + } + } + + return args; + } } } diff --git a/src/test/java/org/springframework/data/gemfire/client/support/DefaultableDelegatingPoolAdapterTest.java b/src/test/java/org/springframework/data/gemfire/client/support/DefaultableDelegatingPoolAdapterTest.java new file mode 100644 index 00000000..c25acf46 --- /dev/null +++ b/src/test/java/org/springframework/data/gemfire/client/support/DefaultableDelegatingPoolAdapterTest.java @@ -0,0 +1,526 @@ +/* + * Copyright 2012 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.client.support; + +import static org.hamcrest.Matchers.equalTo; +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.mockito.Matchers.anyBoolean; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; +import static org.mockito.Mockito.verifyZeroInteractions; +import static org.mockito.Mockito.when; + +import java.net.InetSocketAddress; +import java.util.Collections; +import java.util.List; + +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.runners.MockitoJUnitRunner; +import org.springframework.data.gemfire.GemfireUtils; + +import com.gemstone.gemfire.cache.client.Pool; +import com.gemstone.gemfire.cache.query.QueryService; + +/** + * The DefaultableDelegatingPoolAdapterTest class is a default suite of default cases testing the contract and functionality + * of the {@link DefaultableDelegatingPoolAdapter} class. + * + * @author John Blum + * @see org.junit.Rule + * @see org.junit.Test + * @see org.junit.rules.ExpectedException + * @see org.mockito.Mock + * @see org.mockito.Mockito + * @see org.mockito.runners.MockitoJUnitRunner + * @see org.springframework.data.gemfire.client.support.DefaultableDelegatingPoolAdapter + * @since 1.8.0 + */ +@RunWith(MockitoJUnitRunner.class) +public class DefaultableDelegatingPoolAdapterTest { + + @Rule + public ExpectedException exception = ExpectedException.none(); + + private DefaultableDelegatingPoolAdapter poolAdapter; + + @Mock + private Pool mockPool; + + @Mock + private QueryService mockQueryService; + + @Mock + private DefaultableDelegatingPoolAdapter.ValueProvider mockValueProvider; + + protected static InetSocketAddress newSocketAddress(String host, int port) { + return new InetSocketAddress(host, port); + } + + @Before + public void setupMockPool() { + when(mockPool.isDestroyed()).thenReturn(false); + when(mockPool.getFreeConnectionTimeout()).thenReturn(5000); + when(mockPool.getIdleTimeout()).thenReturn(120000l); + when(mockPool.getLoadConditioningInterval()).thenReturn(300000); + when(mockPool.getLocators()).thenReturn(Collections.emptyList()); + when(mockPool.getMaxConnections()).thenReturn(500); + when(mockPool.getMinConnections()).thenReturn(50); + when(mockPool.getMultiuserAuthentication()).thenReturn(true); + when(mockPool.getName()).thenReturn("TestPool"); + when(mockPool.getPendingEventCount()).thenReturn(2); + when(mockPool.getPingInterval()).thenReturn(15000l); + when(mockPool.getPRSingleHopEnabled()).thenReturn(true); + when(mockPool.getQueryService()).thenReturn(null); + when(mockPool.getReadTimeout()).thenReturn(30000); + when(mockPool.getRetryAttempts()).thenReturn(1); + when(mockPool.getServerGroup()).thenReturn("TestGroup"); + when(mockPool.getServers()).thenReturn(Collections.singletonList( + newSocketAddress("localhost", GemfireUtils.DEFAULT_CACHE_SERVER_PORT))); + when(mockPool.getSocketBufferSize()).thenReturn(16384); + when(mockPool.getStatisticInterval()).thenReturn(1000); + when(mockPool.getSubscriptionAckInterval()).thenReturn(200); + when(mockPool.getSubscriptionEnabled()).thenReturn(true); + when(mockPool.getSubscriptionMessageTrackingTimeout()).thenReturn(20000); + when(mockPool.getSubscriptionRedundancy()).thenReturn(2); + when(mockPool.getThreadLocalConnections()).thenReturn(false); + + setupPoolAdapter(); + } + + //@Before + public void setupPoolAdapter() { + poolAdapter = DefaultableDelegatingPoolAdapter.from(mockPool); + } + + @Test + public void fromMockPoolAsDelegate() { + assertThat(poolAdapter.getDelegate(), is(sameInstance(mockPool))); + } + + @Test + public void fromNullAsDelegate() { + exception.expect(IllegalArgumentException.class); + exception.expectCause(is(nullValue(Throwable.class))); + exception.expectMessage("'delegate' must not be null"); + + DefaultableDelegatingPoolAdapter.from(null); + } + + @Test + public void preferenceDefaultsToPoolDelegateAndIsMutable() { + assertThat(poolAdapter.getPreference(), is(equalTo(DefaultableDelegatingPoolAdapter.Preference.PREFER_POOL))); + assertThat(poolAdapter.prefersPool(), is(true)); + + poolAdapter.preferDefault(); + + assertThat(poolAdapter.getPreference(), is(equalTo(DefaultableDelegatingPoolAdapter.Preference.PREFER_DEFAULT))); + assertThat(poolAdapter.prefersDefault(), is(true)); + + poolAdapter.preferPool(); + + assertThat(poolAdapter.getPreference(), is(equalTo(DefaultableDelegatingPoolAdapter.Preference.PREFER_POOL))); + assertThat(poolAdapter.prefersPool(), is(true)); + } + + @Test + public void defaultIfNullWhenPrefersDefaultUsesDefault() { + poolAdapter = poolAdapter.preferDefault(); + + assertThat(poolAdapter.prefersDefault(), is(true)); + assertThat((String) poolAdapter.defaultIfNull("default", mockValueProvider), is(equalTo("default"))); + + verifyZeroInteractions(mockValueProvider); + } + + @Test + public void defaultIfNullWhenPrefersDefaultUsesPoolValueWhenDefaultIsNull() { + poolAdapter = poolAdapter.preferDefault(); + + when(mockValueProvider.getValue()).thenReturn("pool"); + + assertThat(poolAdapter.prefersDefault(), is(true)); + assertThat((String) poolAdapter.defaultIfNull(null, mockValueProvider), is(equalTo("pool"))); + + verify(mockValueProvider, times(1)).getValue(); + } + + @Test + public void defaultIfNullWhenPrefersPoolUsesPoolValue() { + poolAdapter = poolAdapter.preferPool(); + + when(mockValueProvider.getValue()).thenReturn("pool"); + + assertThat(poolAdapter.prefersPool(), is(true)); + assertThat((String) poolAdapter.defaultIfNull("default", mockValueProvider), is(equalTo("pool"))); + + verify(mockValueProvider, times(1)).getValue(); + } + + @Test + public void defaultIfNullWhenPrefersPoolUsesDefaultWhenPoolValueIsNull() { + poolAdapter = poolAdapter.preferPool(); + + when(mockValueProvider.getValue()).thenReturn(null); + + assertThat(poolAdapter.prefersPool(), is(true)); + assertThat((String) poolAdapter.defaultIfNull("default", mockValueProvider), is(equalTo("default"))); + + verify(mockValueProvider, times(1)).getValue(); + } + + @Test + @SuppressWarnings("unchecked") + public void defaultIfEmptyWhenPrefersDefaultUsesDefault() { + poolAdapter = poolAdapter.preferDefault(); + + List defaultList = Collections.singletonList("default"); + + assertThat(poolAdapter.prefersDefault(), is(true)); + assertThat((List) poolAdapter.defaultIfEmpty(defaultList, mockValueProvider), is(equalTo(defaultList))); + + verifyZeroInteractions(mockValueProvider); + } + + @Test + @SuppressWarnings("unchecked") + public void defaultIfEmptyWhenPrefersDefaultUsesPoolValueWhenDefaultIsNull() { + poolAdapter = poolAdapter.preferDefault(); + + List poolList = Collections.singletonList("pool"); + + when(mockValueProvider.getValue()).thenReturn(poolList); + + assertThat(poolAdapter.prefersDefault(), is(true)); + assertThat((List) poolAdapter.defaultIfEmpty(null, mockValueProvider), is(equalTo(poolList))); + + verify(mockValueProvider, times(1)).getValue(); + } + + @Test + @SuppressWarnings("unchecked") + public void defaultIfEmptyWhenPrefersDefaultUsesPoolValueWhenDefaultIsEmpty() { + poolAdapter = poolAdapter.preferDefault(); + + List poolList = Collections.singletonList("pool"); + + when(mockValueProvider.getValue()).thenReturn(poolList); + + assertThat(poolAdapter.prefersDefault(), is(true)); + assertThat((List) poolAdapter.defaultIfEmpty(Collections.emptyList(), mockValueProvider), + is(equalTo(poolList))); + + verify(mockValueProvider, times(1)).getValue(); + } + + @Test + @SuppressWarnings("unchecked") + public void defaultIfEmptyWhenPrefersPoolUsesPoolValue() { + poolAdapter = poolAdapter.preferPool(); + + List poolList = Collections.singletonList("pool"); + + when(mockValueProvider.getValue()).thenReturn(poolList); + + assertThat(poolAdapter.prefersPool(), is(true)); + assertThat((List) poolAdapter.defaultIfEmpty(Collections.singletonList("default"), mockValueProvider), + is(equalTo(poolList))); + + verify(mockValueProvider, times(1)).getValue(); + } + + @Test + @SuppressWarnings("unchecked") + public void defaultIfEmptyWhenPrefersPoolUsesDefaultWhenPoolValueIsNull() { + poolAdapter = poolAdapter.preferPool(); + + when(mockValueProvider.getValue()).thenReturn(null); + + List defaultList = Collections.singletonList("default"); + + assertThat(poolAdapter.prefersPool(), is(true)); + assertThat((List) poolAdapter.defaultIfEmpty(defaultList, mockValueProvider), + is(equalTo(defaultList))); + + verify(mockValueProvider, times(1)).getValue(); + } + + @Test + @SuppressWarnings("unchecked") + public void defaultIfEmptyWhenPrefersPoolUsesDefaultWhenPoolValueIsEmpty() { + assertThat(poolAdapter.preferPool(), is(sameInstance(poolAdapter))); + + when(mockValueProvider.getValue()).thenReturn(Collections.emptyList()); + + List defaultList = Collections.singletonList("default"); + + assertThat(poolAdapter.prefersPool(), is(true)); + assertThat((List) poolAdapter.defaultIfEmpty(defaultList, mockValueProvider), + is(equalTo(defaultList))); + + verify(mockValueProvider, times(1)).getValue(); + } + + @Test + public void poolAdapterPreferringDefaultsUsesNonNullDefaults() { + assertThat(poolAdapter.preferDefault(), is(sameInstance(poolAdapter))); + + List defaultLocator = Collections.singletonList(newSocketAddress("boombox", 21668)); + List defaultServer = Collections.singletonList(newSocketAddress("skullbox", 42424)); + + assertThat(poolAdapter.getDelegate(), is(equalTo(mockPool))); + assertThat(poolAdapter.prefersDefault(), is(true)); + assertThat(poolAdapter.getFreeConnectionTimeout(10000), is(equalTo(10000))); + assertThat(poolAdapter.getIdleTimeout(300000l), is(equalTo(300000l))); + assertThat(poolAdapter.getLoadConditioningInterval(60000), is(equalTo(60000))); + assertThat(poolAdapter.getLocators(defaultLocator), is(equalTo(defaultLocator))); + assertThat(poolAdapter.getMaxConnections(100), is(equalTo(100))); + assertThat(poolAdapter.getMinConnections(10), is(equalTo(10))); + assertThat(poolAdapter.getMultiuserAuthentication(false), is(equalTo(false))); + assertThat(poolAdapter.getName(), is(equalTo("TestPool"))); + assertThat(poolAdapter.getPendingEventCount(), is(equalTo(2))); + assertThat(poolAdapter.getPingInterval(20000l), is(equalTo(20000l))); + assertThat(poolAdapter.getPRSingleHopEnabled(false), is(equalTo(false))); + assertThat(poolAdapter.getQueryService(mockQueryService), is(equalTo(mockQueryService))); + assertThat(poolAdapter.getReadTimeout(20000), is(equalTo(20000))); + assertThat(poolAdapter.getRetryAttempts(2), is(equalTo(2))); + assertThat(poolAdapter.getServerGroup("MockGroup"), is(equalTo("MockGroup"))); + assertThat(poolAdapter.getServers(defaultServer), is(equalTo(defaultServer))); + assertThat(poolAdapter.getSocketBufferSize(8192), is(equalTo(8192))); + assertThat(poolAdapter.getStatisticInterval(2000), is(equalTo(2000))); + assertThat(poolAdapter.getSubscriptionAckInterval(50), is(equalTo(50))); + assertThat(poolAdapter.getSubscriptionEnabled(false), is(equalTo(false))); + assertThat(poolAdapter.getSubscriptionMessageTrackingTimeout(15000), is(equalTo(15000))); + assertThat(poolAdapter.getSubscriptionRedundancy(1), is(equalTo(1))); + assertThat(poolAdapter.getThreadLocalConnections(true), is(equalTo(true))); + + verify(mockPool, times(1)).getName(); + verify(mockPool, times(1)).getPendingEventCount(); + verifyNoMoreInteractions(mockPool); + } + + @Test + public void poolAdapterPreferringDefaultsUsesPoolValuesWhenSomeDefaultValuesAreNull() { + assertThat(poolAdapter.preferDefault(), is(sameInstance(poolAdapter))); + + List defaultLocator = Collections.singletonList(newSocketAddress("boombox", 21668)); + List poolServer = Collections.singletonList(newSocketAddress("localhost", 40404)); + + assertThat(poolAdapter.getDelegate(), is(equalTo(mockPool))); + assertThat(poolAdapter.prefersDefault(), is(true)); + assertThat(poolAdapter.getFreeConnectionTimeout(null), is(equalTo(5000))); + assertThat(poolAdapter.getIdleTimeout(null), is(equalTo(120000l))); + assertThat(poolAdapter.getLoadConditioningInterval(60000), is(equalTo(60000))); + assertThat(poolAdapter.getLocators(defaultLocator), is(equalTo(defaultLocator))); + assertThat(poolAdapter.getMaxConnections(null), is(equalTo(500))); + assertThat(poolAdapter.getMinConnections(50), is(equalTo(50))); + assertThat(poolAdapter.getMultiuserAuthentication(null), is(equalTo(true))); + assertThat(poolAdapter.getName(), is(equalTo("TestPool"))); + assertThat(poolAdapter.getPendingEventCount(), is(equalTo(2))); + assertThat(poolAdapter.getPingInterval(null), is(equalTo(15000l))); + assertThat(poolAdapter.getPRSingleHopEnabled(true), is(equalTo(true))); + assertThat(poolAdapter.getQueryService(null), is(nullValue())); + assertThat(poolAdapter.getReadTimeout(20000), is(equalTo(20000))); + assertThat(poolAdapter.getRetryAttempts(null), is(equalTo(1))); + assertThat(poolAdapter.getServerGroup("MockGroup"), is(equalTo("MockGroup"))); + assertThat(poolAdapter.getServers(null), is(equalTo(poolServer))); + assertThat(poolAdapter.getSocketBufferSize(32768), is(equalTo(32768))); + assertThat(poolAdapter.getStatisticInterval(null), is(equalTo(1000))); + assertThat(poolAdapter.getSubscriptionAckInterval(50), is(equalTo(50))); + assertThat(poolAdapter.getSubscriptionEnabled(true), is(equalTo(true))); + assertThat(poolAdapter.getSubscriptionMessageTrackingTimeout(null), is(equalTo(20000))); + assertThat(poolAdapter.getSubscriptionRedundancy(null), is(equalTo(2))); + assertThat(poolAdapter.getThreadLocalConnections(null), is(equalTo(false))); + + verify(mockPool, times(1)).getFreeConnectionTimeout(); + verify(mockPool, times(1)).getIdleTimeout(); + verify(mockPool, times(1)).getMaxConnections(); + verify(mockPool, times(1)).getMultiuserAuthentication(); + verify(mockPool, times(1)).getName(); + verify(mockPool, times(1)).getPendingEventCount(); + verify(mockPool, times(1)).getPingInterval(); + verify(mockPool, times(1)).getQueryService(); + verify(mockPool, times(1)).getRetryAttempts(); + verify(mockPool, times(1)).getServers(); + verify(mockPool, times(1)).getStatisticInterval(); + verify(mockPool, times(1)).getSubscriptionMessageTrackingTimeout(); + verify(mockPool, times(1)).getSubscriptionRedundancy(); + verify(mockPool, times(1)).getThreadLocalConnections(); + verifyNoMoreInteractions(mockPool); + } + + @Test + public void poolAdapterPreferringDefaultsUsesPoolValuesExclusivelyWhenAllDefaultValuesAreNull() { + assertThat(poolAdapter.preferDefault(), is(sameInstance(poolAdapter))); + + List poolServer = Collections.singletonList(newSocketAddress("localhost", 40404)); + + assertThat(poolAdapter.getDelegate(), is(equalTo(mockPool))); + assertThat(poolAdapter.prefersDefault(), is(true)); + assertThat(poolAdapter.getFreeConnectionTimeout(null), is(equalTo(5000))); + assertThat(poolAdapter.getIdleTimeout(null), is(equalTo(120000l))); + assertThat(poolAdapter.getLoadConditioningInterval(null), is(equalTo(300000))); + assertThat(poolAdapter.getLocators(null), is(equalTo(Collections.emptyList()))); + assertThat(poolAdapter.getMaxConnections(null), is(equalTo(500))); + assertThat(poolAdapter.getMinConnections(null), is(equalTo(50))); + assertThat(poolAdapter.getMultiuserAuthentication(null), is(equalTo(true))); + assertThat(poolAdapter.getName(), is(equalTo("TestPool"))); + assertThat(poolAdapter.getPendingEventCount(), is(equalTo(2))); + assertThat(poolAdapter.getPingInterval(null), is(equalTo(15000l))); + assertThat(poolAdapter.getPRSingleHopEnabled(null), is(equalTo(true))); + assertThat(poolAdapter.getQueryService(null), is(nullValue())); + assertThat(poolAdapter.getReadTimeout(null), is(equalTo(30000))); + assertThat(poolAdapter.getRetryAttempts(null), is(equalTo(1))); + assertThat(poolAdapter.getServerGroup(null), is(equalTo("TestGroup"))); + assertThat(poolAdapter.getServers(null), is(equalTo(poolServer))); + assertThat(poolAdapter.getSocketBufferSize(null), is(equalTo(16384))); + assertThat(poolAdapter.getStatisticInterval(null), is(equalTo(1000))); + assertThat(poolAdapter.getSubscriptionAckInterval(null), is(equalTo(200))); + assertThat(poolAdapter.getSubscriptionEnabled(null), is(equalTo(true))); + assertThat(poolAdapter.getSubscriptionMessageTrackingTimeout(null), is(equalTo(20000))); + assertThat(poolAdapter.getSubscriptionRedundancy(null), is(equalTo(2))); + assertThat(poolAdapter.getThreadLocalConnections(null), is(equalTo(false))); + + verify(mockPool, times(1)).getFreeConnectionTimeout(); + verify(mockPool, times(1)).getIdleTimeout(); + verify(mockPool, times(1)).getLoadConditioningInterval(); + verify(mockPool, times(1)).getLocators(); + verify(mockPool, times(1)).getMaxConnections(); + verify(mockPool, times(1)).getMinConnections(); + verify(mockPool, times(1)).getMultiuserAuthentication(); + verify(mockPool, times(1)).getName(); + verify(mockPool, times(1)).getPendingEventCount(); + verify(mockPool, times(1)).getPingInterval(); + verify(mockPool, times(1)).getPRSingleHopEnabled(); + verify(mockPool, times(1)).getQueryService(); + verify(mockPool, times(1)).getReadTimeout(); + verify(mockPool, times(1)).getRetryAttempts(); + verify(mockPool, times(1)).getServerGroup(); + verify(mockPool, times(1)).getServers(); + verify(mockPool, times(1)).getSocketBufferSize(); + verify(mockPool, times(1)).getStatisticInterval(); + verify(mockPool, times(1)).getSubscriptionAckInterval(); + verify(mockPool, times(1)).getSubscriptionEnabled(); + verify(mockPool, times(1)).getSubscriptionMessageTrackingTimeout(); + verify(mockPool, times(1)).getSubscriptionRedundancy(); + verify(mockPool, times(1)).getThreadLocalConnections(); + verifyNoMoreInteractions(mockPool); + } + + @Test + public void poolAdapterPreferringPoolUsesUseNonNullPoolValues() { + assertThat(poolAdapter.preferPool(), is(sameInstance(poolAdapter))); + + List defaultServer = Collections.singletonList(newSocketAddress("jambox", 12480)); + List poolServer = Collections.singletonList(newSocketAddress("localhost", 40404)); + + assertThat(poolAdapter.getFreeConnectionTimeout(15000), is(equalTo(5000))); + assertThat(poolAdapter.getIdleTimeout(60000l), is(equalTo(120000l))); + assertThat(poolAdapter.getLoadConditioningInterval(180000), is(equalTo(300000))); + assertThat(poolAdapter.getLocators(Collections.emptyList()), + is(equalTo(Collections.emptyList()))); + assertThat(poolAdapter.getMaxConnections(999), is(equalTo(500))); + assertThat(poolAdapter.getMinConnections(99), is(equalTo(50))); + assertThat(poolAdapter.getMultiuserAuthentication(false), is(equalTo(true))); + assertThat(poolAdapter.getName(), is(equalTo("TestPool"))); + assertThat(poolAdapter.getPendingEventCount(), is(equalTo(2))); + assertThat(poolAdapter.getPingInterval(20000l), is(equalTo(15000l))); + assertThat(poolAdapter.getPRSingleHopEnabled(false), is(equalTo(true))); + assertThat(poolAdapter.getQueryService(null), is(nullValue())); + assertThat(poolAdapter.getReadTimeout(20000), is(equalTo(30000))); + assertThat(poolAdapter.getRetryAttempts(4), is(equalTo(1))); + assertThat(poolAdapter.getServerGroup("MockGroup"), is(equalTo("TestGroup"))); + assertThat(poolAdapter.getServers(defaultServer), is(equalTo(poolServer))); + assertThat(poolAdapter.getSocketBufferSize(8192), is(equalTo(16384))); + assertThat(poolAdapter.getStatisticInterval(2000), is(equalTo(1000))); + assertThat(poolAdapter.getSubscriptionAckInterval(50), is(equalTo(200))); + assertThat(poolAdapter.getSubscriptionEnabled(false), is(equalTo(true))); + assertThat(poolAdapter.getSubscriptionMessageTrackingTimeout(30000), is(equalTo(20000))); + assertThat(poolAdapter.getSubscriptionRedundancy(1), is(equalTo(2))); + assertThat(poolAdapter.getThreadLocalConnections(true), is(equalTo(false))); + + verify(mockPool, times(1)).getFreeConnectionTimeout(); + verify(mockPool, times(1)).getIdleTimeout(); + verify(mockPool, times(1)).getLoadConditioningInterval(); + verify(mockPool, times(1)).getLocators(); + verify(mockPool, times(1)).getMaxConnections(); + verify(mockPool, times(1)).getMinConnections(); + verify(mockPool, times(1)).getMultiuserAuthentication(); + verify(mockPool, times(1)).getName(); + verify(mockPool, times(1)).getPendingEventCount(); + verify(mockPool, times(1)).getPingInterval(); + verify(mockPool, times(1)).getPRSingleHopEnabled(); + verify(mockPool, times(1)).getQueryService(); + verify(mockPool, times(1)).getReadTimeout(); + verify(mockPool, times(1)).getRetryAttempts(); + verify(mockPool, times(1)).getServerGroup(); + verify(mockPool, times(1)).getServers(); + verify(mockPool, times(1)).getSocketBufferSize(); + verify(mockPool, times(1)).getStatisticInterval(); + verify(mockPool, times(1)).getSubscriptionAckInterval(); + verify(mockPool, times(1)).getSubscriptionEnabled(); + verify(mockPool, times(1)).getSubscriptionMessageTrackingTimeout(); + verify(mockPool, times(1)).getSubscriptionRedundancy(); + verify(mockPool, times(1)).getThreadLocalConnections(); + verifyNoMoreInteractions(mockPool); + } + + @Test + public void poolAdapterDestroyUsesPoolRegardlessOfPreference() { + assertThat(poolAdapter.preferDefault(), is(sameInstance(poolAdapter))); + assertThat(poolAdapter.getDelegate(), is(equalTo(mockPool))); + assertThat(poolAdapter.prefersDefault(), is(true)); + + poolAdapter.destroy(); + + assertThat(poolAdapter.preferPool(), is(sameInstance(poolAdapter))); + assertThat(poolAdapter.getDelegate(), is(equalTo(mockPool))); + assertThat(poolAdapter.prefersPool(), is(true)); + + poolAdapter.destroy(true); + + verify(mockPool, times(1)).destroy(); + verify(mockPool, times(1)).destroy(anyBoolean()); + } + + @Test + public void poolAdapterReleaseThreadLocalConnections() { + assertThat(poolAdapter.preferDefault(), is(sameInstance(poolAdapter))); + assertThat(poolAdapter.getDelegate(), is(equalTo(mockPool))); + assertThat(poolAdapter.prefersDefault(), is(true)); + + poolAdapter.releaseThreadLocalConnection(); + + assertThat(poolAdapter.preferPool(), is(sameInstance(poolAdapter))); + assertThat(poolAdapter.getDelegate(), is(equalTo(mockPool))); + assertThat(poolAdapter.prefersPool(), is(true)); + + poolAdapter.releaseThreadLocalConnection(); + + verify(mockPool, times(2)).releaseThreadLocalConnection(); + } + +} diff --git a/src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerTest.java b/src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerTest.java new file mode 100644 index 00000000..3fd026d7 --- /dev/null +++ b/src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerTest.java @@ -0,0 +1,284 @@ +/* + * Copyright 2012 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.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.mockito.Matchers.anyString; +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyZeroInteractions; +import static org.mockito.Mockito.when; + +import java.util.concurrent.Executor; + +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.NoSuchBeanDefinitionException; +import org.springframework.data.gemfire.GemfireUtils; +import org.springframework.data.gemfire.TestUtils; +import org.springframework.data.gemfire.config.GemfireConstants; + +import com.gemstone.gemfire.cache.client.Pool; +import com.gemstone.gemfire.cache.query.QueryService; +import com.gemstone.gemfire.internal.cache.PoolManagerImpl; + +/** + * The ContinuousQueryListenerContainerTest class is a test suite of test cases testing the contract and functionality + * of the {@link ContinuousQueryListenerContainer} class. + * + * @author John Blum + * @see org.junit.Rule + * @see org.junit.Test + * @see org.junit.rules.ExpectedException + * @see org.junit.runner.RunWith + * @see org.mockito.Mock + * @see org.mockito.Mockito + * @see org.mockito.runners.MockitoJUnitRunner + * @see org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer + * @since 1.8.0 + */ +public class ContinuousQueryListenerContainerTest { + + @Rule + public ExpectedException exception = ExpectedException.none(); + + private ContinuousQueryListenerContainer listenerContainer; + + @Before + public void setup() { + listenerContainer = new ContinuousQueryListenerContainer(); + } + + @Test + public void afterPropertiesSetAutoStarts() 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(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(); + } + finally { + assertThat(PoolManagerImpl.getPMI().unregister(mockPool), is(true)); + assertThat(listenerContainer.isAutoStartup(), is(true)); + assertThat((QueryService) TestUtils.readField("queryService", listenerContainer), is(equalTo(mockQueryService))); + assertThat(TestUtils.readField("taskExecutor", listenerContainer), is(instanceOf(Executor.class))); + assertThat(listenerContainer.isActive(), is(true)); + assertThat(listenerContainer.isRunning(), is(true)); + + 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)); + verify(mockBeanFactory, times(1)).getBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class)); + verify(mockPool, times(2)).getName(); + verify(mockPool, times(1)).getQueryService(); + verifyZeroInteractions(mockQueryService); + } + } + + @Test + public void afterPropertiesSetStartsManually() { + BeanFactory mockBeanFactory = mock(BeanFactory.class); + QueryService mockQueryService = mock(QueryService.class); + + PoolManagerImpl poolManagerSpy = spy(PoolManagerImpl.getPMI()); + + when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true); + + listenerContainer.setAutoStartup(false); + listenerContainer.setBeanFactory(mockBeanFactory); + listenerContainer.setPoolName("TestPool"); + listenerContainer.setQueryService(mockQueryService); + listenerContainer.afterPropertiesSet(); + + assertThat(listenerContainer.isActive(), is(true)); + assertThat(listenerContainer.isAutoStartup(), is(false)); + assertThat(listenerContainer.isRunning(), is(false)); + + 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()); + verifyZeroInteractions(mockQueryService); + } + + @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); + + 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"))); + } + finally { + assertThat(PoolManagerImpl.getPMI().unregister(mockPool), is(true)); + verify(mockPool, times(2)).getName(); + } + } + + @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(nullValue(Throwable.class)); + exception.expectMessage("No GemFire Pool with name [TestPool] was found"); + + listenerContainer.setBeanFactory(mockBeanFactory); + listenerContainer.eagerlyInitializePool("TestPool"); + } + finally { + verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class)); + verify(mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class)); + } + } + + @Test + public void initQueryServiceReturnsProvidedQueryService() { + QueryService mockQueryService = mock(QueryService.class); + + listenerContainer.setQueryService(mockQueryService); + + assertThat(listenerContainer.initQueryService(null), is(sameInstance(mockQueryService))); + + verifyZeroInteractions(mockQueryService); + } + + @Test + public void initializesQueryServiceFromPool() { + Pool mockPool = mock(Pool.class); + QueryService mockQueryService = mock(QueryService.class); + + when(mockPool.getName()).thenReturn("TestPool"); + when(mockPool.getQueryService()).thenReturn(mockQueryService); + + try { + PoolManagerImpl.getPMI().register(mockPool); + assertThat(listenerContainer.initQueryService("TestPool"), is(equalTo(mockQueryService))); + } + finally { + assertThat(PoolManagerImpl.getPMI().unregister(mockPool), is(true)); + verify(mockPool, times(2)).getName(); + verify(mockPool, times(1)).getQueryService(); + verifyZeroInteractions(mockQueryService); + } + } + + @Test + public void initExecutorReturnsProvidedExecutor() { + Executor mockExecutor = mock(Executor.class); + + listenerContainer.setTaskExecutor(mockExecutor); + + assertThat(listenerContainer.initExecutor(), is(sameInstance(mockExecutor))); + + verifyZeroInteractions(mockExecutor); + } + + @Test + public void initializesDefaultTaskExecutor() { + assertThat(listenerContainer.initExecutor(), is(instanceOf(Executor.class))); + } + +} diff --git a/src/test/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapterTest.java b/src/test/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapterTest.java index 99adff99..4d81e899 100644 --- a/src/test/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapterTest.java +++ b/src/test/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapterTest.java @@ -85,8 +85,7 @@ public class QueryListenerAdapterTest { }; } - @SuppressWarnings("unused") - public static interface Delegate { + interface Delegate { void handleEvent(CqEvent event); @@ -116,8 +115,7 @@ public class QueryListenerAdapterTest { @Test public void testThatTheDefaultHandlingMethodNameIsTheConstantDefault() throws Exception { - assertEquals(ContinuousQueryListenerAdapter.ORIGINAL_DEFAULT_LISTENER_METHOD, - adapter.getDefaultListenerMethod()); + assertEquals(ContinuousQueryListenerAdapter.DEFAULT_LISTENER_METHOD_NAME, adapter.getDefaultListenerMethod()); } @Test @@ -235,8 +233,8 @@ public class QueryListenerAdapterTest { SampleListener listener = new SampleListener(); ContinuousQueryListener listenerAdapter = new ContinuousQueryListenerAdapter(listener) { - protected void handleListenerException(Throwable ex) { - throw new RuntimeException(ex); + protected void handleListenerException(Throwable cause) { + throw new RuntimeException(cause); } };