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

Fixes JIRA issue SGF-244 involving the use of a ref attribute on the GatewaySender element's nested inner EventFilter element.

Browse Source
This commit is contained in:
John Blum
2013-12-19 16:04:18 -08:00
parent a861346b95
commit c278676423
3 changed files with 156 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.4.xsd
View File

@@ -72,7 +72,6 @@ use inner bean declarations.
</xsd:attribute> </xsd:attribute>
</xsd:complexType> </xsd:complexType>
</xsd:element> </xsd:element>
<xsd:element name="dynamic-region-factory" minOccurs="0" <xsd:element name="dynamic-region-factory" minOccurs="0"
maxOccurs="1"> maxOccurs="1">
<xsd:annotation> <xsd:annotation>
@@ -2335,19 +2334,25 @@ A gateway event filter for this gateway sender
</xsd:annotation> </xsd:annotation>
<xsd:complexType> <xsd:complexType>
<xsd:sequence> <xsd:sequence>
<xsd:any namespace="##other" processContents="skip" <xsd:any namespace="##other" processContents="skip" minOccurs="0" maxOccurs="unbounded">
minOccurs="0" maxOccurs="unbounded">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Inner bean definition of the event filter Inner bean definition for the event filter.
]]></xsd:documentation> ]]></xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:any> </xsd:any>
</xsd:sequence> </xsd:sequence>
<xsd:attribute name="ref" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The name of the event filter bean referred to by this declaration. Used for convenience. If no reference exists,
use inner bean declarations.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType> </xsd:complexType>
</xsd:element> </xsd:element>
<xsd:element name="transport-filter" type="gatewayTransportFilterType" <xsd:element name="transport-filter" type="gatewayTransportFilterType" minOccurs="0" maxOccurs="1" />
minOccurs="0" maxOccurs="1" />
</xsd:sequence> </xsd:sequence>
<xsd:attributeGroup ref="commonWANQueueAttributes" /> <xsd:attributeGroup ref="commonWANQueueAttributes" />
<xsd:attribute name="name" type="xsd:string" use="optional"> <xsd:attribute name="name" type="xsd:string" use="optional">
@@ -2772,8 +2777,7 @@ A transport filter for this gateway component
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
<xsd:sequence> <xsd:sequence>
<xsd:any namespace="##other" processContents="skip" <xsd:any namespace="##other" processContents="skip" minOccurs="0" maxOccurs="unbounded">
minOccurs="0" maxOccurs="unbounded">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Inner bean definition of the transport filter. Inner bean definition of the transport filter.
@@ -2784,7 +2788,7 @@ Inner bean definition of the transport filter.
<xsd:attribute name="ref" type="xsd:string" use="optional"> <xsd:attribute name="ref" type="xsd:string" use="optional">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
The name of the transport filter bean referred by this declaration. Used as a convenience method. If no reference exists, The name of the transport filter bean referred to by this declaration. Used for convenience. If no reference exists,
use inner bean declarations. use inner bean declarations.
]]></xsd:documentation> ]]></xsd:documentation>
</xsd:annotation> </xsd:annotation>

141
src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java
View File

@@ -18,6 +18,7 @@ package org.springframework.data.gemfire.config;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue; import static org.junit.Assert.assertTrue;
import java.io.File; import java.io.File;
@@ -50,11 +51,11 @@ import com.gemstone.gemfire.cache.wan.GatewayTransportFilter;
/** /**
* This test is only valid for GF 7.0 and above * This test is only valid for GF 7.0 and above
* * <p/>
* @author David Turanski * @author David Turanski
* * @author John Blum
*/ */
@SuppressWarnings("unused")
public class GemfireV7GatewayNamespaceTest extends RecreatingContextTest { public class GemfireV7GatewayNamespaceTest extends RecreatingContextTest {
/* (non-Javadoc) /* (non-Javadoc)
@@ -107,73 +108,113 @@ public class GemfireV7GatewayNamespaceTest extends RecreatingContextTest {
@Test @Test
public void testGatewaySender() throws Exception { public void testGatewaySender() throws Exception {
GatewaySenderFactoryBean gwsfb = ctx.getBean("&gateway-sender", GatewaySenderFactoryBean.class); GatewaySenderFactoryBean gatewaySenderFactoryBean = ctx.getBean("&gateway-sender", GatewaySenderFactoryBean.class);
Cache cache = TestUtils.readField("cache", gwsfb);
assertNotNull(cache); assertNotNull(gatewaySenderFactoryBean);
List<GatewayEventFilter> eventFilters = TestUtils.readField("eventFilters", gwsfb); assertNotNull(TestUtils.readField("cache", gatewaySenderFactoryBean));
assertEquals(2, TestUtils.readField("remoteDistributedSystemId", gatewaySenderFactoryBean));
assertEquals(10, TestUtils.readField("alertThreshold", gatewaySenderFactoryBean));
assertEquals(11, TestUtils.readField("batchSize", gatewaySenderFactoryBean));
assertEquals(12, TestUtils.readField("dispatcherThreads", gatewaySenderFactoryBean));
assertEquals(true, TestUtils.readField("manualStart", gatewaySenderFactoryBean));
List<GatewayEventFilter> eventFilters = TestUtils.readField("eventFilters", gatewaySenderFactoryBean);
assertNotNull(eventFilters); assertNotNull(eventFilters);
assertEquals(2, eventFilters.size()); assertEquals(2, eventFilters.size());
assertTrue(eventFilters.get(0) instanceof TestEventFilter); assertTrue(eventFilters.get(0) instanceof TestEventFilter);
List<GatewayTransportFilter> transportFilters = TestUtils.readField("transportFilters", gwsfb); List<GatewayTransportFilter> transportFilters = TestUtils.readField("transportFilters", gatewaySenderFactoryBean);
assertNotNull(transportFilters); assertNotNull(transportFilters);
assertEquals(2, transportFilters.size()); assertEquals(2, transportFilters.size());
assertTrue(transportFilters.get(0) instanceof TestTransportFilter); assertTrue(transportFilters.get(0) instanceof TestTransportFilter);
assertEquals(2, TestUtils.readField("remoteDistributedSystemId", gwsfb));
assertEquals(10, TestUtils.readField("alertThreshold", gwsfb));
assertEquals(11, TestUtils.readField("batchSize", gwsfb));
assertEquals(12, TestUtils.readField("dispatcherThreads", gwsfb));
assertEquals(true, TestUtils.readField("manualStart", gwsfb));
} }
@SuppressWarnings("rawtypes") @SuppressWarnings("rawtypes")
@Test @Test
public void testInnerGatewaySender() throws Exception { public void testInnerGatewaySender() throws Exception {
Region<?, ?> region = ctx.getBean("region-inner-gateway-sender", Region.class); Region<?, ?> region = ctx.getBean("region-inner-gateway-sender", Region.class);
GatewaySender gws = ctx.getBean("gateway-sender", GatewaySender.class);
assertNotNull(region.getAttributes().getGatewaySenderIds()); assertNotNull(region.getAttributes().getGatewaySenderIds());
assertEquals(2, region.getAttributes().getGatewaySenderIds().size()); assertEquals(2, region.getAttributes().getGatewaySenderIds().size());
RegionFactoryBean rfb = ctx.getBean("&region-inner-gateway-sender", RegionFactoryBean.class); RegionFactoryBean regionFactoryBean = ctx.getBean("&region-inner-gateway-sender", RegionFactoryBean.class);
Object[] gwsenders = TestUtils.readField("gatewaySenders", rfb);
gws = (GatewaySender) gwsenders[0]; Object[] gatewaySenders = TestUtils.readField("gatewaySenders", regionFactoryBean);
List<GatewayEventFilter> eventFilters = gws.getGatewayEventFilters();
assertNotNull(gatewaySenders);
assertEquals(2, gatewaySenders.length);
GatewaySender gatewaySender = (GatewaySender) gatewaySenders[0];
assertNotNull(gatewaySender);
assertEquals(1, gatewaySender.getRemoteDSId());
assertEquals(false, gatewaySender.isManualStart());
assertEquals(true,gatewaySender.isRunning());
assertEquals(10, gatewaySender.getAlertThreshold());
assertEquals(11, gatewaySender.getBatchSize());
assertEquals(3000, gatewaySender.getBatchTimeInterval());
assertEquals(2, gatewaySender.getDispatcherThreads());
assertEquals("diskstore", gatewaySender.getDiskStoreName());
assertTrue(gatewaySender.isBatchConflationEnabled());
assertEquals(50, gatewaySender.getMaximumQueueMemory());
assertEquals(OrderPolicy.THREAD, gatewaySender.getOrderPolicy());
assertTrue(gatewaySender.isPersistenceEnabled());
assertFalse(gatewaySender.isParallel());
assertEquals(16536, gatewaySender.getSocketBufferSize());
assertEquals(3000, gatewaySender.getSocketReadTimeout());
List<GatewayEventFilter> eventFilters = gatewaySender.getGatewayEventFilters();
assertNotNull(eventFilters); assertNotNull(eventFilters);
assertEquals(1, eventFilters.size()); assertEquals(1, eventFilters.size());
assertTrue(eventFilters.get(0) instanceof TestEventFilter); assertTrue(eventFilters.get(0) instanceof TestEventFilter);
List<GatewayTransportFilter> transportFilters = gws.getGatewayTransportFilters(); List<GatewayTransportFilter> transportFilters = gatewaySender.getGatewayTransportFilters();
assertNotNull(transportFilters); assertNotNull(transportFilters);
assertEquals(1, transportFilters.size()); assertEquals(1, transportFilters.size());
assertTrue(transportFilters.get(0) instanceof TestTransportFilter); assertTrue(transportFilters.get(0) instanceof TestTransportFilter);
assertEquals(1, gws.getRemoteDSId());
assertEquals(false, gws.isManualStart());
assertEquals(true,gws.isRunning());
assertEquals(10, gws.getAlertThreshold());
assertEquals(11, gws.getBatchSize());
assertEquals(3000, gws.getBatchTimeInterval());
assertEquals(2, gws.getDispatcherThreads());
assertEquals("diskstore", gws.getDiskStoreName());
assertTrue(gws.isBatchConflationEnabled());
assertEquals(50, gws.getMaximumQueueMemory());
assertEquals(OrderPolicy.THREAD, gws.getOrderPolicy());
assertTrue(gws.isPersistenceEnabled());
assertFalse(gws.isParallel());
assertEquals(16536, gws.getSocketBufferSize());
assertEquals(3000, gws.getSocketReadTimeout());
} }
@Test @Test
public void testInnerGatewayReceiver() { public void testGatewaySenderWithEventTransportFilterRefs() throws Exception {
GatewayReceiver gwr = ctx.getBean("gateway-receiver", GatewayReceiver.class); GatewaySenderFactoryBean gatewaySenderFactoryBean = ctx.getBean("&gateway-sender-with-event-transport-filter-refs",
assertEquals(12345, gwr.getStartPort()); GatewaySenderFactoryBean.class);
assertEquals(23456, gwr.getEndPort());
assertEquals("192.168.0.1", gwr.getBindAddress()); assertNotNull(gatewaySenderFactoryBean);
assertEquals(3000, gwr.getMaximumTimeBetweenPings()); assertNotNull(TestUtils.readField("cache", gatewaySenderFactoryBean));
assertEquals(16536, gwr.getSocketBufferSize()); assertEquals(3, TestUtils.readField("remoteDistributedSystemId", gatewaySenderFactoryBean));
assertEquals(50, TestUtils.readField("batchSize", gatewaySenderFactoryBean));
assertEquals(10, TestUtils.readField("dispatcherThreads", gatewaySenderFactoryBean));
assertEquals(true, TestUtils.readField("manualStart", gatewaySenderFactoryBean));
List<GatewayEventFilter> eventFilters = TestUtils.readField("eventFilters", gatewaySenderFactoryBean);
assertNotNull(eventFilters);
assertEquals(1, eventFilters.size());
assertTrue(eventFilters.get(0) instanceof TestEventFilter);
assertSame(ctx.getBean("event-filter"), eventFilters.get(0));
List<GatewayTransportFilter> transportFilters = TestUtils.readField("transportFilters", gatewaySenderFactoryBean);
assertNotNull(transportFilters);
assertEquals(1, transportFilters.size());
assertTrue(transportFilters.get(0) instanceof TestTransportFilter);
assertSame(ctx.getBean("transport-filter"), transportFilters.get(0));
}
@Test
public void testGatewayReceiver() {
GatewayReceiver gatewayReceiver = ctx.getBean("gateway-receiver", GatewayReceiver.class);
assertNotNull(gatewayReceiver);
assertEquals("192.168.0.1", gatewayReceiver.getBindAddress());
assertEquals(12345, gatewayReceiver.getStartPort());
assertEquals(23456, gatewayReceiver.getEndPort());
assertEquals(3000, gatewayReceiver.getMaximumTimeBetweenPings());
assertEquals(16536, gatewayReceiver.getSocketBufferSize());
} }
@SuppressWarnings("rawtypes") @SuppressWarnings("rawtypes")
@@ -181,49 +222,38 @@ public class GemfireV7GatewayNamespaceTest extends RecreatingContextTest {
@Override @Override
public void close() { public void close() {
// TODO Auto-generated method stub
} }
@Override @Override
public void afterAcknowledgement(GatewayQueueEvent arg0) { public void afterAcknowledgement(GatewayQueueEvent arg0) {
// TODO Auto-generated method stub
} }
@Override @Override
public boolean beforeEnqueue(GatewayQueueEvent arg0) { public boolean beforeEnqueue(GatewayQueueEvent arg0) {
// TODO Auto-generated method stub
return false; return false;
} }
@Override @Override
public boolean beforeTransmit(GatewayQueueEvent arg0) { public boolean beforeTransmit(GatewayQueueEvent arg0) {
// TODO Auto-generated method stub
return false; return false;
} }
} }
public static class TestTransportFilter implements GatewayTransportFilter { public static class TestTransportFilter implements GatewayTransportFilter {
@Override @Override
public void close() { public void close() {
// TODO Auto-generated method stub
} }
@Override @Override
public InputStream getInputStream(InputStream arg0) { public InputStream getInputStream(InputStream arg0) {
// TODO Auto-generated method stub
return null; return null;
} }
@Override @Override
public OutputStream getOutputStream(OutputStream arg0) { public OutputStream getOutputStream(OutputStream arg0) {
// TODO Auto-generated method stub
return null; return null;
} }
} }
@SuppressWarnings("rawtypes") @SuppressWarnings("rawtypes")
@@ -231,15 +261,12 @@ public class GemfireV7GatewayNamespaceTest extends RecreatingContextTest {
@Override @Override
public void close() { public void close() {
// TODO Auto-generated method stub
} }
@Override @Override
public boolean processEvents(List<AsyncEvent> arg0) { public boolean processEvents(List<AsyncEvent> arg0) {
// TODO Auto-generated method stub
return false; return false;
} }
} }
} }

116
src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml
View File

@@ -1,83 +1,83 @@
<?xml version="1.0" encoding="UTF-8"?> <?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" <beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:gfe="http://www.springframework.org/schema/gemfire"
xmlns:gfe="http://www.springframework.org/schema/gemfire" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p" xsi:schemaLocation="
xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd" default-lazy-init="true"> http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd
" default-lazy-init="true">
<gfe:cache/> <gfe:cache/>
<gfe:partitioned-region id="region-inner-gateway-sender" > <gfe:partitioned-region id="region-inner-gateway-sender" >
<gfe:gateway-sender <gfe:gateway-sender remote-distributed-system-id="1"
manual-start="false" manual-start="false"
remote-distributed-system-id="1" alert-threshold="10"
alert-threshold="10" batch-size="11"
batch-size="11" batch-time-interval="3000"
batch-time-interval="3000" dispatcher-threads="2"
dispatcher-threads="2" disk-store-ref="diskstore"
disk-store-ref="diskstore" enable-batch-conflation="true"
enable-batch-conflation="true" maximum-queue-memory="50"
maximum-queue-memory="50" order-policy="THREAD"
order-policy="THREAD" persistent="true"
persistent="true" parallel="false"
parallel="false" socket-buffer-size="16536"
socket-buffer-size="16536" socket-read-timeout="3000">
socket-read-timeout="3000"> <gfe:event-filter>
<gfe:event-filter> <bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestEventFilter"/>
<bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestEventFilter"/> </gfe:event-filter>
</gfe:event-filter> <gfe:transport-filter>
<gfe:transport-filter> <bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestTransportFilter"/>
<bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestTransportFilter"/> </gfe:transport-filter>
</gfe:transport-filter> </gfe:gateway-sender>
</gfe:gateway-sender> <gfe:gateway-sender-ref bean="gateway-sender"/>
<gfe:gateway-sender-ref bean="gateway-sender"/>
</gfe:partitioned-region> </gfe:partitioned-region>
<gfe:async-event-queue id="async-event-queue" <gfe:async-event-queue id="async-event-queue"
batch-size="10" batch-conflation-enabled="true"
persistent="true" batch-size="10"
disk-store-ref="diskstore" batch-time-interval="3"
maximum-queue-memory="50" dispatcher-threads="4"
parallel="false" disk-store-ref="diskstore"
batch-conflation-enabled="true" disk-synchronous="true"
batch-time-interval="3" maximum-queue-memory="50"
dispatcher-threads="4" parallel="false"
disk-synchronous="true" persistent="true"
order-policy="KEY"> order-policy="KEY">
<gfe:async-event-listener> <gfe:async-event-listener>
<bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestAsyncEventListener"/> <bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestAsyncEventListener"/>
</gfe:async-event-listener> </gfe:async-event-listener>
</gfe:async-event-queue> </gfe:async-event-queue>
<gfe:disk-store id="diskstore"/> <gfe:disk-store id="diskstore"/>
<gfe:gateway-receiver id="gateway-receiver" <gfe:gateway-receiver id="gateway-receiver" bind-address="192.168.0.1" start-port="12345" end-port="23456"
start-port="12345" end-port="23456" bind-address="192.168.0.1" maximum-time-between-pings="3000" socket-buffer-size="16536"> maximum-time-between-pings="3000" socket-buffer-size="16536">
<gfe:transport-filter> <gfe:transport-filter>
<bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestTransportFilter"/> <bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestTransportFilter"/>
</gfe:transport-filter> </gfe:transport-filter>
</gfe:gateway-receiver> </gfe:gateway-receiver>
<!-- need manual-start=true for the unit test because GF will throw an exception if no locators are configured --> <!-- need manual-start=true for the unit test because GF will throw an exception if no locators are configured -->
<gfe:gateway-sender id="gateway-sender" <gfe:gateway-sender id="gateway-sender" remote-distributed-system-id="2" alert-threshold="10" batch-size="11"
remote-distributed-system-id="2" dispatcher-threads="12" manual-start="true">
alert-threshold="10" <gfe:event-filter>
batch-size="11" <ref bean="event-filter"/>
dispatcher-threads="12" <bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestEventFilter"/>
manual-start="true"> </gfe:event-filter>
<gfe:event-filter> <gfe:transport-filter>
<ref bean="event-filter"/> <ref bean="transport-filter"/>
<bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestEventFilter"/> <bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestTransportFilter"/>
</gfe:event-filter> </gfe:transport-filter>
<gfe:transport-filter>
<ref bean="transport-filter"/>
<bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestTransportFilter"/>
</gfe:transport-filter>
</gfe:gateway-sender> </gfe:gateway-sender>
<gfe:gateway-sender id="gateway-sender-with-event-transport-filter-refs" remote-distributed-system-id="3"
batch-size="50" dispatcher-threads="10" manual-start="true">
<gfe:event-filter ref="event-filter"/>
<gfe:transport-filter ref="transport-filter"/>
</gfe:gateway-sender>
<bean id="event-filter" class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestEventFilter"/> <bean id="event-filter" class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestEventFilter"/>
<bean id="transport-filter" class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestTransportFilter"/> <bean id="transport-filter" class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestTransportFilter"/>