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

Added support for gateway conflict resolver and function service

Browse Source
This commit is contained in:
David Turanski
2012-07-23 17:30:13 -04:00
parent 94468bec9a
commit 57b33c007a
11 changed files with 390 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/main/java/org/springframework/data/gemfire/CacheFactoryBean.java
View File

@@ -47,6 +47,9 @@ import com.gemstone.gemfire.cache.DynamicRegionFactory;
import com.gemstone.gemfire.cache.GemFireCache;
import com.gemstone.gemfire.cache.TransactionListener;
import com.gemstone.gemfire.cache.TransactionWriter;
import com.gemstone.gemfire.cache.execute.Function;
import com.gemstone.gemfire.cache.execute.FunctionService;
import com.gemstone.gemfire.cache.util.GatewayConflictResolver;
import com.gemstone.gemfire.distributed.DistributedMember;
import com.gemstone.gemfire.distributed.DistributedSystem;
import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
@@ -129,6 +132,10 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl
if (messageSyncInterval != null) {
cacheImpl.setMessageSyncInterval(messageSyncInterval);
}
if (gatewayConflictResolver != null) {
cacheImpl.setGatewayConflictResolver(gatewayConflictResolver);
}
}
}
@@ -258,6 +265,10 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl
protected List<JndiDataSource> jndiDataSources;
protected List<Function> functions;
protected GatewayConflictResolver gatewayConflictResolver;
@Override
public void afterPropertiesSet() throws Exception {
// initialize locator
@@ -324,12 +335,24 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl
registerTransactionListeners();
registerTransactionWriter();
registerJndiDataSources();
registerFunctions();
}
finally {
th.setContextClassLoader(oldTCCL);
}
}
/**
*
*/
private void registerFunctions() {
if (!CollectionUtils.isEmpty(functions)) {
for (Function function : functions) {
FunctionService.registerFunction(function);
}
}
}
private void registerJndiDataSources() {
if (jndiDataSources != null) {
for (JndiDataSource jndiDataSource : jndiDataSources) {
@@ -605,6 +628,14 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl
this.transactionWriter = transactionWriter;
}
public void setFunctions(List<Function> functions) {
this.functions = functions;
}
public void setGatewayConflictResolver(GatewayConflictResolver gatewayConflictResolver) {
this.gatewayConflictResolver = gatewayConflictResolver;
}
public void setDynamicRegionSupport(DynamicRegionSupport dynamicRegionSupport) {
this.dynamicRegionSupport = dynamicRegionSupport;
}

63
src/main/java/org/springframework/data/gemfire/FunctionServiceFactoryBean.java Normal file
View File

@@ -0,0 +1,63 @@
/*
* Copyright 2010-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;
import java.util.List;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.CollectionUtils;
import com.gemstone.gemfire.cache.execute.Function;
import com.gemstone.gemfire.cache.execute.FunctionService;
/**
* @author David Turanski
*
*/
public class FunctionServiceFactoryBean implements FactoryBean<FunctionService>, InitializingBean {
static FunctionService functionService;
private List<Function> functions;
@Override
public void afterPropertiesSet() throws Exception {
if (!CollectionUtils.isEmpty(functions)) {
for (Function function : functions) {
FunctionService.registerFunction(function);
}
}
}
public void setFunctions(List<Function> functions) {
this.functions = functions;
}
@Override
public FunctionService getObject() throws Exception {
return functionService;
}
@Override
public Class<?> getObjectType() {
return FunctionService.class;
}
@Override
public boolean isSingleton() {
return true;
}
}

14
src/main/java/org/springframework/data/gemfire/config/CacheParser.java
View File

@@ -84,6 +84,20 @@ class CacheParser extends AbstractSimpleBeanDefinitionParser {
ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, txWriter, builder));
}
Element gatewayConflictResolver = DomUtils.getChildElementByTagName(element, "gateway-conflict-resolver");
if (gatewayConflictResolver != null) {
ParsingUtils.throwExceptionIfNotGemfireV7(element.getLocalName(), "gateway-conflict-resolver",
parserContext);
builder.addPropertyValue("gatewayConflictResolver",
ParsingUtils.parseRefOrSingleNestedBeanDeclaration(parserContext, gatewayConflictResolver, builder));
}
Element function = DomUtils.getChildElementByTagName(element, "function");
if (function != null) {
builder.addPropertyValue("functions",
ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, function, builder));
}
parseDynamicRegionFactory(element, builder);
parseJndiBindings(element, builder);
}

61
src/main/java/org/springframework/data/gemfire/config/FunctionServiceParser.java Normal file
View File

@@ -0,0 +1,61 @@
/*
* Copyright 2010-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.config;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.data.gemfire.FunctionServiceFactoryBean;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
/**
* Parser for &lt;function-service;gt; definitions.
* @author David Turanski
*/
class FunctionServiceParser extends AbstractSimpleBeanDefinitionParser {
@Override
protected Class<?> getBeanClass(Element element) {
return FunctionServiceFactoryBean.class;
}
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
builder.setLazyInit(false);
super.doParse(element, builder);
Element function = DomUtils.getChildElementByTagName(element, "function");
if (function != null) {
builder.addPropertyValue("functions",
ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, function, builder));
}
}
@Override
protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
throws BeanDefinitionStoreException {
String name = super.resolveId(element, definition, parserContext);
if (!StringUtils.hasText(name)) {
name = "gemfire-function-service";
}
return name;
}
}

1
src/main/java/org/springframework/data/gemfire/config/GemfireNamespaceHandler.java
View File

@@ -64,6 +64,7 @@ class GemfireNamespaceHandler extends NamespaceHandlerSupport {
registerBeanDefinitionParser("async-event-queue", new AsyncEventQueueParser());
registerBeanDefinitionParser("gateway-sender", new GatewaySenderParser());
registerBeanDefinitionParser("gateway-receiver", new GatewayReceiverParser());
registerBeanDefinitionParser("function-service", new FunctionServiceParser());
// V6 WAN parsers
registerBeanDefinitionParser("gateway-hub", new GatewayHubParser());
}

91
src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.2.xsd
View File

@@ -35,6 +35,45 @@ and may be nested or referenced.
]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="gateway-conflict-resolver"
minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation
source="com.gemstone.gemfire.cache.util.GatewayConflictResolver"><![CDATA[
A gateway conflict resolver for this cache. A gateway conflict resolver handles conflicts in the case of concurrent updates using a WAN gateway. The bean
must implement com.gemstone.gemfire.cache.util.GatewayConflictResolver. Requires Gemfire version 7.0 or higher.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:exports
type="com.gemstone.gemfire.cache.util.GatewayConflictResolver" />
</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 of the gateway conflict resolver.
]]></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 gateway conflict resolver bean referred by this declaration. Used as a convenience method. If no reference exists,
use inner bean declarations.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="dynamic-region-factory"
minOccurs="0" maxOccurs="1">
<xsd:annotation>
@@ -2438,6 +2477,58 @@ The id of the cache - default is gemfire-cache
<xsd:enumeration value="PRELOADED" />
</xsd:restriction>
</xsd:simpleType>
<xsd:element name="function-service">
<xsd:complexType>
<xsd:sequence>
<xsd:element name="function" minOccurs="0"
maxOccurs="1">
<xsd:annotation>
<xsd:documentation
source="com.gemstone.gemfire.cache.execute.Function"><![CDATA[
Declares one or more remote functions for this cache and register's with them the FunctionService. each bean
must implement com.gemstone.gemfire.cache.execute.Function
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:exports
type="com.gemstone.gemfire.cache.execute.Function" />
</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 of the remote function.
]]></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 remote function bean referred by this declaration. Used as a convenience method. If no reference exists,
use inner bean declarations.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The id of the function service (optional)
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<!-- -->
<xsd:simpleType name="scopeType">
<xsd:annotation>

22
src/test/java/org/springframework/data/gemfire/config/CacheNamespaceTest.java
View File

@@ -33,6 +33,11 @@ import org.springframework.data.gemfire.TestUtils;
import org.springframework.data.gemfire.client.ClientCacheFactoryBean;
import org.springframework.test.util.ReflectionTestUtils;
import com.gemstone.gemfire.cache.Cache;
import com.gemstone.gemfire.cache.util.GatewayConflictHelper;
import com.gemstone.gemfire.cache.util.GatewayConflictResolver;
import com.gemstone.gemfire.cache.util.TimestampedEntryEvent;
/**
* @author Costin Leau
*/
@@ -48,9 +53,8 @@ public class CacheNamespaceTest extends RecreatingContextTest {
testBasicCache();
testNamedCache();
testCacheWithXml();
// testBasicClientCache();
// testBasicClientCacheWithXml();
testHeapTunedCache();
testCacheWithGatewayConflictResolver();
}
private void testBasicCache() throws Exception {
@@ -79,6 +83,12 @@ public class CacheNamespaceTest extends RecreatingContextTest {
}
private void testCacheWithGatewayConflictResolver() {
Cache cache = ctx.getBean("cache-with-conflict-resolver", Cache.class);
assertNotNull(cache.getGatewayConflictResolver());
assertTrue(cache.getGatewayConflictResolver() instanceof TestConflictResolver);
}
@Test(expected = IllegalArgumentException.class)
public void testNoBeanFactory() throws Exception {
assertTrue(ctx.containsBean("no-bl"));
@@ -121,4 +131,12 @@ public class CacheNamespaceTest extends RecreatingContextTest {
assertEquals(70, chp, 0.0001);
assertEquals(60, ehp, 0.0001);
}
public static class TestConflictResolver implements GatewayConflictResolver {
@Override
public void onEvent(TimestampedEntryEvent arg0, GatewayConflictHelper arg1) {
// TODO Auto-generated method stub
}
}
}

77
src/test/java/org/springframework/data/gemfire/config/FunctionServiceNamespaceTest.java Normal file
View File

@@ -0,0 +1,77 @@
/*
* Copyright 2010-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.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import org.junit.Test;
import org.springframework.data.gemfire.RecreatingContextTest;
import com.gemstone.gemfire.cache.execute.FunctionAdapter;
import com.gemstone.gemfire.cache.execute.FunctionContext;
import com.gemstone.gemfire.cache.execute.FunctionService;
/**
* @author Costin Leau
*/
public class FunctionServiceNamespaceTest extends RecreatingContextTest {
@Override
protected String location() {
return "org/springframework/data/gemfire/config/function-service-ns.xml";
}
@Test
public void testFunctionsRegistered() throws Exception {
assertEquals(2, FunctionService.getRegisteredFunctions().size());
assertNotNull(FunctionService.getFunction("function1"));
assertNotNull(FunctionService.getFunction("function2"));
}
@SuppressWarnings("serial")
public static class Function1 extends FunctionAdapter {
@Override
public void execute(FunctionContext arg0) {
// TODO Auto-generated method stub
}
@Override
public String getId() {
return "function1";
}
}
@SuppressWarnings("serial")
public static class Function2 extends FunctionAdapter {
@Override
public void execute(FunctionContext arg0) {
// TODO Auto-generated method stub
}
@Override
public String getId() {
return "function2";
}
}
}

11
src/test/resources/org/springframework/data/gemfire/config/cache-ns.xml
View File

@@ -12,7 +12,7 @@
<!-- all beans are lazy to allow the same config to be used between multiple tests -->
<!-- as there can be only one cache per VM -->
<gfe:cache />
<gfe:cache/>
<gfe:cache id="cache-with-name"/>
@@ -29,4 +29,11 @@
<gfe:client-cache id="client-cache-with-xml" cache-xml-location="classpath:gemfire-client-cache.xml"/>
<gfe:cache id="heap-tuned-cache" critical-heap-percentage="70.0" eviction-heap-percentage="60.0"/>
</beans>
<gfe:cache id="cache-with-conflict-resolver">
<gfe:gateway-conflict-resolver>
<bean class="org.springframework.data.gemfire.config.CacheNamespaceTest.TestConflictResolver"/>
</gfe:gateway-conflict-resolver>
</gfe:cache>
</beans>

21
src/test/resources/org/springframework/data/gemfire/config/function-service-ns.xml Normal file
View File

@@ -0,0 +1,21 @@
<?xml version="1.0" encoding="UTF-8"?>
<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:p="http://www.springframework.org/schema/p"
default-lazy-init="true"
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/util http://www.springframework.org/schema/util/spring-util.xsd">
<gfe:function-service>
<gfe:function>
<bean class="org.springframework.data.gemfire.config.FunctionServiceNamespaceTest.Function1"/>
<ref bean="function2"/>
</gfe:function>
</gfe:function-service>
<bean id="function2" class="org.springframework.data.gemfire.config.FunctionServiceNamespaceTest.Function2"/>
</beans>

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

@@ -11,7 +11,7 @@
<gfe:cache />
<!-- need manual-start=true for the unit test because GF will throw an exception if no locators are configured -->
<gfe:local-region id="region-inner-gateway-sender" >
<gfe:partitioned-region id="region-inner-gateway-sender" >
<gfe:gateway-sender
remote-distributed-system-id="1"
manual-start="true"
@@ -35,7 +35,7 @@
</gfe:transport-filter>
</gfe:gateway-sender>
<gfe:gateway-sender-ref bean="gateway-sender"/>
</gfe:local-region>
</gfe:partitioned-region>
<gfe:async-event-queue id="async-event-queue" batch-size="10" persistent="true" disk-store-ref="diskstore"
maximum-queue-memory="50">