diff --git a/src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java b/src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java
index 661bc270..292396d5 100644
--- a/src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java
+++ b/src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java
@@ -26,7 +26,7 @@ import org.w3c.dom.Element;
/**
* @author David Turanski
- *
+ * @author John Blum
*/
public class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
@@ -38,28 +38,29 @@ public class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
builder.setLazyInit(false);
- Element asyncEventListenerElement = DomUtils.getChildElementByTagName(element, "async-event-listener");
- Object asyncEventListener = ParsingUtils.parseRefOrSingleNestedBeanDeclaration(parserContext,
- asyncEventListenerElement, builder);
- String cacheName = !StringUtils.hasText(element.getAttribute("cache-ref")) ? GemfireConstants.DEFAULT_GEMFIRE_CACHE_NAME
- : element.getAttribute("cache-ref");
+ Element asyncEventListenerElement = DomUtils.getChildElementByTagName(element, "async-event-listener");
+
+ Object asyncEventListener = ParsingUtils.parseRefOrSingleNestedBeanDeclaration(parserContext,
+ asyncEventListenerElement, builder);
+
+ builder.addPropertyValue("asyncEventListener", asyncEventListener);
+
+ String cacheRefAttribute = element.getAttribute("cache-ref");
+
+ String cacheName = (StringUtils.hasText(cacheRefAttribute) ? cacheRefAttribute
+ : GemfireConstants.DEFAULT_GEMFIRE_CACHE_NAME);
builder.addConstructorArgReference(cacheName);
- builder.addConstructorArgValue(asyncEventListener);
+
ParsingUtils.setPropertyValue(element, builder, "batch-size");
ParsingUtils.setPropertyValue(element, builder, "maximum-queue-memory");
- ParsingUtils.setPropertyValue(element, builder, "disk-store-ref");
-
- String diskStoreRef = element.getAttribute("disk-store-ref");
- if (StringUtils.hasText(diskStoreRef)) {
- builder.addDependsOn(diskStoreRef);
- }
-
ParsingUtils.setPropertyValue(element, builder, "persistent");
ParsingUtils.setPropertyValue(element, builder, "parallel");
-
- if (GemfireUtils.GEMFIRE_VERSION.compareTo("7.0.1") >= 0 ) {
+
+ parseDiskStore(element, builder);
+
+ if (GemfireUtils.GEMFIRE_VERSION.compareTo("7.0.1") >= 0) {
ParsingUtils.setPropertyValue(element, builder, "batch-conflation-enabled");
ParsingUtils.setPropertyValue(element, builder, "disk-synchronous");
ParsingUtils.setPropertyValue(element, builder, "batch-time-interval");
@@ -87,4 +88,15 @@ public class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
}
}
}
+
+ private void parseDiskStore(final Element element, final BeanDefinitionBuilder builder) {
+ ParsingUtils.setPropertyValue(element, builder, "disk-store-ref");
+
+ String diskStoreRef = element.getAttribute("disk-store-ref");
+
+ if (StringUtils.hasText(diskStoreRef)) {
+ builder.addDependsOn(diskStoreRef);
+ }
+ }
+
}
diff --git a/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java b/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java
index 92c2b5e0..6a5503fb 100644
--- a/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java
+++ b/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java
@@ -20,57 +20,59 @@ import java.util.List;
import org.springframework.util.Assert;
-import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueue;
-import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueueFactory;
import com.gemstone.gemfire.cache.Cache;
import com.gemstone.gemfire.cache.CacheClosedException;
import com.gemstone.gemfire.cache.asyncqueue.AsyncEventListener;
+import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueue;
+import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueueFactory;
import com.gemstone.gemfire.cache.util.Gateway;
/**
* FactoryBean for creating GemFire {@link AsyncEventQueue}s.
+ *
* @author David Turanski
- *
+ * @author John Blum
*/
public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean {
+
private static List validOrderPolicyValues = Arrays.asList("KEY", "PARTITION", "THREAD");
- public void setDiskStoreRef(String diskStoreRef) {
- this.diskStoreRef = diskStoreRef;
- }
-
- private final AsyncEventListener asyncEventListener;
+ private AsyncEventListener asyncEventListener;
private AsyncEventQueue asyncEventQueue;
- private Integer batchSize;
-
- private Integer dispatcherThreads;
-
- private Integer batchTimeInterval;
-
- private Integer maximumQueueMemory;
-
+ private Boolean batchConflationEnabled;
+ private Boolean diskSynchronous;
+ private Boolean parallel;
private Boolean persistent;
+ private Integer batchSize;
+ private Integer batchTimeInterval;
+ private Integer dispatcherThreads;
+ private Integer maximumQueueMemory;
+
private String diskStoreRef;
-
- private Boolean parallel;
-
- private Boolean diskSynchronous;
-
private String orderPolicy;
- private Boolean batchConflationEnabled;
+ /**
+ * Constructs an instance of the AsyncEventQueueFactoryBean for creating an GemFire AsyncEventQueue.
+ *
+ * @param cache the GemFire Cache reference.
+ * @see #AsyncEventQueueFactoryBean(com.gemstone.gemfire.cache.Cache, com.gemstone.gemfire.cache.asyncqueue.AsyncEventListener)
+ */
+ public AsyncEventQueueFactoryBean(Cache cache) {
+ this(cache, null);
+ }
/**
- *
- * @param cache the gemfire cache
+ * Constructs an instance of the AsyncEventQueueFactoryBean for creating an GemFire AsyncEventQueue.
+ *
+ * @param cache the GemFire Cache reference.
* @param asyncEventListener required {@link AsyncEventListener}
*/
public AsyncEventQueueFactoryBean(Cache cache, AsyncEventListener asyncEventListener) {
super(cache);
- this.asyncEventListener = asyncEventListener;
+ setAsyncEventListener(asyncEventListener);
}
@Override
@@ -85,25 +87,23 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
@Override
protected void doInit() {
- Assert.notNull(asyncEventListener, "AsyncEventListener cannot be null");
+ Assert.notNull(this.asyncEventListener, "The AsyncEventListener cannot be null.");
- AsyncEventQueueFactory asyncEventQueueFactory = null;
- if (this.factory == null) {
- asyncEventQueueFactory = cache.createAsyncEventQueueFactory();
- } else {
- asyncEventQueueFactory = (AsyncEventQueueFactory) factory;
- }
+ AsyncEventQueueFactory asyncEventQueueFactory = (this.factory != null ? (AsyncEventQueueFactory) factory
+ : cache.createAsyncEventQueueFactory());
if (diskStoreRef != null) {
persistent = (persistent == null) ? Boolean.TRUE : persistent;
- Assert.isTrue(persistent, "specifying a disk store requires persistent property to be true");
+ Assert.isTrue(persistent, "Specifying a 'disk store' requires the persistent property to be true.");
asyncEventQueueFactory.setDiskStoreName(diskStoreRef);
}
+
if (diskSynchronous != null) {
persistent = (persistent == null) ? Boolean.TRUE : persistent;
- Assert.isTrue(persistent, "specifying a disk synchronous requires persistent property to be true");
+ Assert.isTrue(persistent, "Specifying 'disk synchronous' requires the persistent property to be true.");
asyncEventQueueFactory.setDiskSynchronous(diskSynchronous);
}
+
if (persistent != null) {
asyncEventQueueFactory.setPersistent(persistent);
}
@@ -135,25 +135,36 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
if (orderPolicy != null) {
Assert.isTrue(parallel, "specifying an order policy requires the parallel property to be true");
- Assert.isTrue(validOrderPolicyValues.contains(orderPolicy.toUpperCase()), "The value of order policy:'"
- + orderPolicy + "' is invalid");
+ Assert.isTrue(validOrderPolicyValues.contains(orderPolicy.toUpperCase()), String.format(
+ "The value of order policy:'$1%s'' is invalid.", orderPolicy));
+
asyncEventQueueFactory.setOrderPolicy(Gateway.OrderPolicy.valueOf(orderPolicy.toUpperCase()));
}
- asyncEventQueue = asyncEventQueueFactory.create(getName(), asyncEventListener);
+ asyncEventQueue = asyncEventQueueFactory.create(getName(), this.asyncEventListener);
}
@Override
public void destroy() throws Exception {
if (!cache.isClosed()) {
try {
- asyncEventListener.close();
- } catch (CacheClosedException cce) {
- // nothing to see folks, move on.
+ this.asyncEventListener.close();
+ }
+ catch (CacheClosedException ignore) {
}
}
}
+ public final void setAsyncEventListener(AsyncEventListener listener) {
+ Assert.state(this.asyncEventQueue == null,
+ "Setting an AsyncEventListener is not allowed once the AsyncEventQueue has been created.");
+ this.asyncEventListener = listener;
+ }
+
+ public void setDiskStoreRef(String diskStoreRef) {
+ this.diskStoreRef = diskStoreRef;
+ }
+
public void setBatchSize(Integer batchSize) {
this.batchSize = batchSize;
}
@@ -219,4 +230,5 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
public void setOrderPolicy(String orderPolicy) {
this.orderPolicy = orderPolicy;
}
+
}
diff --git a/src/test/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBeanTest.java b/src/test/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBeanTest.java
new file mode 100644
index 00000000..0959cfba
--- /dev/null
+++ b/src/test/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBeanTest.java
@@ -0,0 +1,129 @@
+/*
+ * Copyright 2010-2013 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.wan;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.notNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.junit.Test;
+import org.springframework.data.gemfire.TestUtils;
+
+import com.gemstone.gemfire.cache.Cache;
+import com.gemstone.gemfire.cache.asyncqueue.AsyncEventListener;
+import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueue;
+import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueueFactory;
+
+/**
+ * The AsyncEventQueueFactoryBeanTest class is a test suite of test cases testing the contract and functionality
+ * of the AsyncEventQueueFactoryBean class.
+ *
+ * @author John Blum
+ * @see org.junit.Test
+ * @see org.mockito.Mockito
+ * @see org.springframework.data.gemfire.wan.AsyncEventQueueFactoryBean
+ * @since 1.3.3
+ */
+public class AsyncEventQueueFactoryBeanTest {
+
+ protected Cache createMockCacheWithAsyncInfrastructure(final String asyncEventQueueId) {
+ Cache mockCache = mock(Cache.class);
+ AsyncEventQueueFactory mockAsynEventQueueFactory = mock(AsyncEventQueueFactory.class);
+ AsyncEventQueue mockAsyncEventQueue = mock(AsyncEventQueue.class);
+
+ when((mockCache.createAsyncEventQueueFactory())).thenReturn(mockAsynEventQueueFactory);
+ when(mockAsynEventQueueFactory.create(eq(asyncEventQueueId), notNull(AsyncEventListener.class)))
+ .thenReturn(mockAsyncEventQueue);
+ when(mockAsyncEventQueue.getId()).thenReturn(asyncEventQueueId);
+
+ return mockCache;
+ }
+
+ protected AsyncEventListener createMockAsyncEventListener() {
+ return mock(AsyncEventListener.class);
+ }
+
+ @Test
+ public void testSetAsyncEventListener() throws Exception {
+ AsyncEventQueueFactoryBean factoryBean = new AsyncEventQueueFactoryBean(
+ createMockCacheWithAsyncInfrastructure("testEventQueue"));
+
+ AsyncEventListener listenerOne = createMockAsyncEventListener();
+
+ factoryBean.setAsyncEventListener(listenerOne);
+
+ assertSame(listenerOne, TestUtils.readField("asyncEventListener", factoryBean));
+
+ AsyncEventListener listenerTwo = createMockAsyncEventListener();
+
+ factoryBean.setAsyncEventListener(listenerTwo);
+
+ assertSame(listenerTwo, TestUtils.readField("asyncEventListener", factoryBean));
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testSetAsyncEventListenerAfterAyncEventQueueCreation() throws Exception {
+ String asyncEventQueueId = "testEventQueue";
+
+ AsyncEventQueueFactoryBean factoryBean = new AsyncEventQueueFactoryBean(
+ createMockCacheWithAsyncInfrastructure(asyncEventQueueId));
+
+ factoryBean.setName(asyncEventQueueId);
+
+ AsyncEventListener listenerOne = createMockAsyncEventListener();
+
+ factoryBean.setAsyncEventListener(listenerOne);
+
+ assertSame(listenerOne, TestUtils.readField("asyncEventListener", factoryBean));
+
+ factoryBean.doInit();
+
+ assertNotNull(TestUtils.readField("asyncEventQueue", factoryBean));
+
+ try {
+ factoryBean.setAsyncEventListener(createMockAsyncEventListener());
+ }
+ catch (IllegalStateException expected) {
+ assertEquals("Setting an AsyncEventListener is not allowed once the AsyncEventQueue has been created.",
+ expected.getMessage());
+ assertSame(listenerOne, TestUtils.readField("asyncEventListener", factoryBean));
+ throw expected;
+ }
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testDoInitWhenAsyncEventListenerIsNull() throws Exception {
+ try {
+ AsyncEventQueueFactoryBean factoryBean = new AsyncEventQueueFactoryBean(
+ createMockCacheWithAsyncInfrastructure("testEventQueue"));
+
+ assertNull(TestUtils.readField("asyncEventListener", factoryBean));
+
+ factoryBean.doInit();
+ }
+ catch (Exception e) {
+ assertEquals("The AsyncEventListener cannot be null.", e.getMessage());
+ throw e;
+ }
+ }
+
+}