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

SGF-332 - Add support for the GatewaySender 'eventSubstitutionFilter' property in the SDG XSD and GatewaySenderFactoryBean API.

Browse Source
This commit is contained in:
John Blum
2014-09-25 21:39:23 -07:00
parent d8b4b97834
commit fccb737849
7 changed files with 252 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/main/java/org/springframework/data/gemfire/config/GatewaySenderParser.java
View File

@@ -66,6 +66,13 @@ class GatewaySenderParser extends AbstractSimpleBeanDefinitionParser {
parserContext, eventFilterElement, builder));
}
Element eventSubstitutionFilterElement = DomUtils.getChildElementByTagName(element, "event-substitution-filter");
if (eventSubstitutionFilterElement != null) {
builder.addPropertyValue("eventSubstitutionFilter", ParsingUtils.parseRefOrSingleNestedBeanDeclaration(
parserContext, eventSubstitutionFilterElement, builder));
}
ParsingUtils.parseTransportFilters(element, parserContext, builder);
// set the name for the GatewaySender as an inner bean

5
src/main/java/org/springframework/data/gemfire/config/ParsingUtils.java
View File

@@ -251,13 +251,12 @@ abstract class ParsingUtils {
return false;
}
static void parseTransportFilters(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
Element transportFilterElement = DomUtils.getChildElementByTagName(element, "transport-filter");
if (transportFilterElement != null) {
builder.addPropertyValue("transportFilters",
parseRefOrNestedBeanDeclaration(parserContext, transportFilterElement, builder));
builder.addPropertyValue("transportFilters", parseRefOrNestedBeanDeclaration(parserContext,
transportFilterElement, builder));
}
}

11
src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java
View File

@@ -24,6 +24,7 @@ import org.springframework.util.CollectionUtils;
import com.gemstone.gemfire.cache.Cache;
import com.gemstone.gemfire.cache.util.Gateway;
import com.gemstone.gemfire.cache.wan.GatewayEventFilter;
import com.gemstone.gemfire.cache.wan.GatewayEventSubstitutionFilter;
import com.gemstone.gemfire.cache.wan.GatewaySender;
import com.gemstone.gemfire.cache.wan.GatewaySenderFactory;
import com.gemstone.gemfire.cache.wan.GatewayTransportFilter;
@@ -60,6 +61,8 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
private Boolean parallel;
private Boolean persistent;
private GatewayEventSubstitutionFilter eventSubstitutionFilter;
private Integer alertThreshold;
private Integer batchSize;
private Integer batchTimeInterval;
@@ -130,6 +133,10 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
}
}
if (eventSubstitutionFilter != null) {
gatewaySenderFactory.setGatewayEventSubstitutionFilter(eventSubstitutionFilter);
}
gatewaySenderFactory.setManualStart(true);
if (maximumQueueMemory != null) {
@@ -209,6 +216,10 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
this.dispatcherThreads = dispatcherThreads;
}
public void setEventSubstitutionFilter(final GatewayEventSubstitutionFilter eventSubstitutionFilter) {
this.eventSubstitutionFilter = eventSubstitutionFilter;
}
public void setManualStart(Boolean manualStart) {
this.manualStart = Boolean.TRUE.equals(manualStart);
}

129
src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-2.0.xsd
View File

@@ -2572,39 +2572,9 @@ A gateway sender gateway definition (requires Gemfire 7.0 or later)
</xsd:appinfo>
</xsd:annotation>
<xsd:sequence>
<xsd:element name="event-filter" minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation
source="com.gemstone.gemfire.cache.wan.GatewayEventFilter"><![CDATA[
A gateway event filter for this gateway sender
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:exports type="com.gemstone.gemfire.cache.wan.GatewayEventFilter" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:any namespace="##other" processContents="skip" minOccurs="0" maxOccurs="unbounded">
<xsd:annotation>
<xsd:documentation><![CDATA[
Inner bean definition for the event filter.
]]></xsd:documentation>
</xsd:annotation>
</xsd:any>
</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:element>
<xsd:element name="transport-filter" type="gatewayTransportFilterType" minOccurs="0" maxOccurs="1" />
<xsd:element name="event-filter" type="gatewayEventFilterType" minOccurs="0" maxOccurs="1"/>
<xsd:element name="event-substitution-filter" type="gatewayEventSubstitutionFilterType" minOccurs="0" maxOccurs="1"/>
<xsd:element name="transport-filter" type="gatewayTransportFilterType" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attributeGroup ref="commonWANQueueAttributes" />
<xsd:attribute name="name" type="xsd:string" use="optional">
@@ -2629,8 +2599,7 @@ to when the acknowldgement was received from the gateway receiver.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="batch-time-interval" type="xsd:string"
use="optional">
<xsd:attribute name="batch-time-interval" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The maximum time interval that can elapse before a partial batch is sent from a GatewaySender to its corresponding GatewayReceiver.
@@ -3017,9 +2986,10 @@ The id of the function service (optional)
<xsd:simpleType name="scopeType">
<xsd:annotation>
<xsd:documentation><![CDATA[
Determines how updates to region entries are distributed to the other caches in the distributed system where the region and entry are defined.
Scope also determines whether to allow remote invocation of some of the region’s event handlers
]]></xsd:documentation>
Determines how updates to Region Entries are distributed to the other Caches in the Distributed System where
the Region and Entry are defined. Scope also determines whether to allow remote invocation of some of
the Region’s event handlers
]]></xsd:documentation>
</xsd:annotation>
<xsd:restriction base="xsd:string">
<xsd:enumeration value="distributed-ack" />
@@ -3027,17 +2997,15 @@ Scope also determines whether to allow remote invocation of some of the region
<xsd:enumeration value="global" />
</xsd:restriction>
</xsd:simpleType>
<xsd:complexType name="gatewayTransportFilterType">
<!-- -->
<xsd:complexType name="gatewayEventFilterType">
<xsd:annotation>
<xsd:documentation
source="com.gemstone.gemfire.cache.wan.GatewayTransportFilter"><![CDATA[
A transport filter for this gateway component
]]></xsd:documentation>
<xsd:documentation source="com.gemstone.gemfire.cache.wan.GatewayEventFilter"><![CDATA[
A Gateway Event Filter for this GatewaySender.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:exports
type="com.gemstone.gemfire.cache.wan.GatewayTransportFilter" />
<tool:exports type="com.gemstone.gemfire.cache.wan.GatewayEventFilter"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
@@ -3045,16 +3013,77 @@ A transport filter for this gateway component
<xsd:any namespace="##other" processContents="skip" minOccurs="0" maxOccurs="unbounded">
<xsd:annotation>
<xsd:documentation><![CDATA[
Inner bean definition of the transport filter.
]]></xsd:documentation>
Inner bean definition(s) to declare and add GatewayEventFilter(s) to the GatewaySender.
]]></xsd:documentation>
</xsd:annotation>
</xsd:any>
</xsd:sequence>
<xsd:attribute name="ref" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The name of the transport filter bean referred to by this declaration. Used for convenience. If no reference exists,
use inner bean declarations.
The name of the GatewaySender 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 name="gatewayEventSubstitutionFilterType">
<xsd:annotation>
<xsd:documentation source="com.gemstone.gemfire.cache.wan.GatewayEventSubstitutionFilter"><![CDATA[
A Gateway Event Substitution Filter for this GatewaySender.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:exports type="com.gemstone.gemfire.cache.wan.GatewayEventSubstitutionFilter"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
<xsd:sequence>
<xsd:any namespace="##other" processContents="skip" minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation><![CDATA[
Inner bean definition to declare and add a single GatewayEventSubstitutionFilter to the GatewaySender.
]]></xsd:documentation>
</xsd:annotation>
</xsd:any>
</xsd:sequence>
<xsd:attribute name="ref" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The name of the GatewaySender Event Sustitution 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 name="gatewayTransportFilterType">
<xsd:annotation>
<xsd:documentation source="com.gemstone.gemfire.cache.wan.GatewayTransportFilter"><![CDATA[
A Gateway Transport Filter for this GatewaySender.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:exports type="com.gemstone.gemfire.cache.wan.GatewayTransportFilter"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
<xsd:sequence>
<xsd:any namespace="##other" processContents="skip" minOccurs="0" maxOccurs="unbounded">
<xsd:annotation>
<xsd:documentation><![CDATA[
Inner bean definition(s) for the Gateway Transport Filter(s) to add to this GatewaySender.
]]></xsd:documentation>
</xsd:annotation>
</xsd:any>
</xsd:sequence>
<xsd:attribute name="ref" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The name of the GatewaySender Transport Filter bean referred to by this declaration.
Used for convenience. If no reference exists, use inner bean declarations.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>

117
src/test/java/org/springframework/data/gemfire/config/GemfireV8GatewayNamespaceTest.java Normal file
View File

@@ -0,0 +1,117 @@
/*
* 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.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.gemfire.test.GemfireTestApplicationContextInitializer;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.gemstone.gemfire.cache.EntryEvent;
import com.gemstone.gemfire.cache.wan.GatewayEventSubstitutionFilter;
import com.gemstone.gemfire.cache.wan.GatewaySender;
/**
* The GemfireV8GatewayNamespaceTest class is a test suite of test cases testing the contract and functionality of
* GemFire 8 Gateway Sender/Receiver support.
*
* @author John Blum
* @see org.junit.Test
* @see org.junit.runner.RunWith
* @see org.springframework.data.gemfire.test.GemfireTestApplicationContextInitializer
* @see org.springframework.data.gemfire.wan.GatewaySenderFactoryBean
* @see org.springframework.test.context.ContextConfiguration
* @see org.springframework.test.context.junit4.SpringJUnit4ClassRunner
* @see com.gemstone.gemfire.cache.wan.GatewaySender
* @see com.gemstone.gemfire.cache.wan.GatewayEventSubstitutionFilter
* @since 2.0.0
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(value = "gateway-v8-ns.xml", initializers = GemfireTestApplicationContextInitializer.class)
@SuppressWarnings("unused")
public class GemfireV8GatewayNamespaceTest {
@Autowired
@Qualifier("gateway-sender-with-event-substitution-filter")
private GatewaySender gatewaySenderWithEventSubstitutionFilter;
@Autowired
@Qualifier("gateway-sender-with-event-substitution-filter-ref")
private GatewaySender gatewaySenderWithEventSubstitutionFilterRef;
@Test
public void testGatewaySenderEventSubstitutionFilter() {
assertNotNull("The 'gatewaySenderEventSubtitutionFilter' bean was not properly configured and initialized!",
gatewaySenderWithEventSubstitutionFilter);
assertEquals("gateway-sender-with-event-substitution-filter", gatewaySenderWithEventSubstitutionFilter.getId());
assertEquals(3, gatewaySenderWithEventSubstitutionFilter.getRemoteDSId());
assertEquals(10, gatewaySenderWithEventSubstitutionFilter.getDispatcherThreads());
assertTrue(gatewaySenderWithEventSubstitutionFilter.isParallel());
assertFalse(gatewaySenderWithEventSubstitutionFilter.isRunning());
assertNotNull(gatewaySenderWithEventSubstitutionFilter.getGatewayEventSubstitutionFilter());
assertEquals("inner", gatewaySenderWithEventSubstitutionFilter.getGatewayEventSubstitutionFilter().toString());
}
@Test
public void testGatewaySenderEventSubstitutionFilterRef() {
assertNotNull("The 'gatewaySenderEventSubtitutionFilter' bean was not properly configured and initialized!",
gatewaySenderWithEventSubstitutionFilterRef);
assertEquals("gateway-sender-with-event-substitution-filter-ref", gatewaySenderWithEventSubstitutionFilterRef.getId());
assertEquals(33, gatewaySenderWithEventSubstitutionFilterRef.getRemoteDSId());
assertEquals(1, gatewaySenderWithEventSubstitutionFilterRef.getDispatcherThreads());
assertFalse(gatewaySenderWithEventSubstitutionFilterRef.isParallel());
assertFalse(gatewaySenderWithEventSubstitutionFilterRef.isRunning());
assertNotNull(gatewaySenderWithEventSubstitutionFilterRef.getGatewayEventSubstitutionFilter());
assertEquals("ref", gatewaySenderWithEventSubstitutionFilterRef.getGatewayEventSubstitutionFilter().toString());
}
public static class TestGatewayEventSubstitutionFilter implements GatewayEventSubstitutionFilter<Object, Object> {
private String name;
public final void setName(final String name) {
this.name = name;
}
protected String getName() {
return name;
}
@Override
public Object getSubstituteValue(final EntryEvent<Object, Object> objectObjectEntryEvent) {
throw new UnsupportedOperationException("Not Implemented!");
}
@Override
public void close() {
}
@Override
public String toString() {
return getName();
}
}
}

2
src/test/java/org/springframework/data/gemfire/test/StubGatewaySenderFactory.java
View File

@@ -44,7 +44,7 @@ public class StubGatewaySenderFactory implements GatewaySenderFactory {
private boolean manualStart;
private boolean parallel;
private boolean persistenceEnabled;
private boolean running = true;
private boolean running = false;
private int alertThreshold;
private int batchSize;

35
src/test/resources/org/springframework/data/gemfire/config/gateway-v8-ns.xml Normal file
View File

@@ -0,0 +1,35 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:gfe="http://www.springframework.org/schema/gemfire"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
">
<util:properties id="gemfireProperties">
<prop key="name">GemfireV8GatewayNamespaceTest</prop>
<prop key="mcast-port">0</prop>
<prop key="log-level">config</prop>
</util:properties>
<gfe:cache properties-ref="gemfireProperties"/>
<gfe:gateway-sender id="gateway-sender-with-event-substitution-filter" remote-distributed-system-id="3"
dispatcher-threads="10" parallel="true" manual-start="true">
<gfe:event-substitution-filter>
<bean class="org.springframework.data.gemfire.config.GemfireV8GatewayNamespaceTest$TestGatewayEventSubstitutionFilter" p:name="inner"/>
</gfe:event-substitution-filter>
</gfe:gateway-sender>
<gfe:gateway-sender id="gateway-sender-with-event-substitution-filter-ref" remote-distributed-system-id="33"
dispatcher-threads="1" parallel="false" manual-start="true">
<gfe:event-substitution-filter ref="testGatewayEventSubstitutionFilter"/>
</gfe:gateway-sender>
<bean id="testGatewayEventSubstitutionFilter" class="org.springframework.data.gemfire.config.GemfireV8GatewayNamespaceTest$TestGatewayEventSubstitutionFilter" p:name="ref"/>
</beans>