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

BATCH-1120: Added chunk-completion-policy to <tasklet/> element.

Browse Source
This commit is contained in:
dhgarrette
2009-03-09 21:25:22 +00:00
parent c2a1e5c384
commit b7bc7695e0
12 changed files with 258 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/TaskletElementParser.java
View File

@@ -18,6 +18,7 @@ package org.springframework.batch.core.configuration.xml;
import java.util.ArrayList;
import java.util.List;
import org.springframework.beans.MutablePropertyValues;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.BeanReference;
import org.springframework.beans.factory.config.RuntimeBeanReference;
@@ -77,56 +78,70 @@ public class TaskletElementParser {
else {
bd = new RootBeanDefinition("org.springframework.batch.core.step.item.SimpleStepFactoryBean", null, null);
}
MutablePropertyValues propertyValues = bd.getPropertyValues();
String readerBeanId = element.getAttribute("reader");
if (StringUtils.hasText(readerBeanId)) {
RuntimeBeanReference readerRef = new RuntimeBeanReference(readerBeanId);
bd.getPropertyValues().addPropertyValue("itemReader", readerRef);
propertyValues.addPropertyValue("itemReader", readerRef);
}
String processorBeanId = element.getAttribute("processor");
if (StringUtils.hasText(processorBeanId)) {
RuntimeBeanReference processorRef = new RuntimeBeanReference(processorBeanId);
bd.getPropertyValues().addPropertyValue("itemProcessor", processorRef);
propertyValues.addPropertyValue("itemProcessor", processorRef);
}
String writerBeanId = element.getAttribute("writer");
if (StringUtils.hasText(writerBeanId)) {
RuntimeBeanReference writerRef = new RuntimeBeanReference(writerBeanId);
bd.getPropertyValues().addPropertyValue("itemWriter", writerRef);
propertyValues.addPropertyValue("itemWriter", writerRef);
}
String taskExecutorBeanId = element.getAttribute("task-executor");
if (StringUtils.hasText(taskExecutorBeanId)) {
RuntimeBeanReference taskExecutorRef = new RuntimeBeanReference(taskExecutorBeanId);
bd.getPropertyValues().addPropertyValue("taskExecutor", taskExecutorRef);
propertyValues.addPropertyValue("taskExecutor", taskExecutorRef);
}
String commitInterval = element.getAttribute("commit-interval");
if (StringUtils.hasText(commitInterval)) {
bd.getPropertyValues().addPropertyValue("commitInterval", commitInterval);
propertyValues.addPropertyValue("commitInterval", commitInterval);
}
if (StringUtils.hasText(skipLimit)) {
bd.getPropertyValues().addPropertyValue("skipLimit", skipLimit);
String completionPolicyRef = element.getAttribute("chunk-completion-policy");
if (StringUtils.hasText(completionPolicyRef)) {
RuntimeBeanReference completionPolicy = new RuntimeBeanReference(completionPolicyRef);
propertyValues.addPropertyValue("chunkCompletionPolicy", completionPolicy);
}
if (propertyValues.contains("commitInterval") == propertyValues.contains("chunkCompletionPolicy")) {
parserContext.getReaderContext().error(
"The 'tasklet' element must contain either 'commit-interval' "
+ "or 'chunk-completion-policy', but not both.", element);
}
if (StringUtils.hasText(skipLimit)) {
propertyValues.addPropertyValue("skipLimit", skipLimit);
}
if (StringUtils.hasText(retryLimit)) {
bd.getPropertyValues().addPropertyValue("retryLimit", retryLimit);
propertyValues.addPropertyValue("retryLimit", retryLimit);
}
if (StringUtils.hasText(cacheCapacity)) {
bd.getPropertyValues().addPropertyValue("cacheCapacity", cacheCapacity);
propertyValues.addPropertyValue("cacheCapacity", cacheCapacity);
}
String transactionAttribute = element.getAttribute("transaction-attribute");
if (StringUtils.hasText(transactionAttribute)) {
bd.getPropertyValues().addPropertyValue("transactionAttribute", transactionAttribute);
propertyValues.addPropertyValue("transactionAttribute", transactionAttribute);
}
if (StringUtils.hasText(isReaderTransactionalQueue)) {
if (isFaultTolerant) {
bd.getPropertyValues().addPropertyValue("isReaderTransactionalQueue", isReaderTransactionalQueue);
propertyValues.addPropertyValue("isReaderTransactionalQueue", isReaderTransactionalQueue);
}
}

23
spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.0.xsd
View File

@@ -356,10 +356,11 @@
</xsd:simpleType>
</xsd:element>
</xsd:all>
<xsd:attribute name="commit-interval" type="xsd:string" use="required">
<xsd:attribute name="commit-interval" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The number of items that will be processed before commit is called for the transaction.
The number of items that will be processed before commit is called for the transaction.
Either set this or the chunk-completion-policy but not both.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -437,7 +438,7 @@
<xsd:annotation>
<xsd:documentation
source="java:org.springframework.core.task.TaskExecutor"><![CDATA[
The task executor responsible for executing the task..
The task executor responsible for executing the task.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
@@ -447,6 +448,22 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="chunk-completion-policy" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation
source="java:org.springframework.batch.repeat.CompletionPolicy"><![CDATA[
A transaction will be committed when this policy decides to
complete. Defaults to a SimpleCompletionPolicy with chunk size
equal to the commit-interval attribute.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="java:org.springframework.batch.repeat.CompletionPolicy" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:complexType name="nextType">

33
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/DummyCompletionPolicy.java Normal file
View File

@@ -0,0 +1,33 @@
package org.springframework.batch.core.configuration.xml;
import org.springframework.batch.repeat.CompletionPolicy;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.RepeatStatus;
/**
* @author Dan Garrette
* @since 2.0
*/
public class DummyCompletionPolicy implements CompletionPolicy {
public boolean isComplete(RepeatContext context, RepeatStatus result) {
// TODO Auto-generated method stub
return false;
}
public boolean isComplete(RepeatContext context) {
// TODO Auto-generated method stub
return false;
}
public RepeatContext start(RepeatContext parent) {
// TODO Auto-generated method stub
return null;
}
public void update(RepeatContext context) {
// TODO Auto-generated method stub
}
}

17
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/DummyItemReader.java Normal file
View File

@@ -0,0 +1,17 @@
package org.springframework.batch.core.configuration.xml;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
/**
* @author Dan Garrette
* @since 2.0
*/
public class DummyItemReader implements ItemReader<Object> {
public Object read() throws Exception, UnexpectedInputException, ParseException {
return null;
}
}

16
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/DummyItemWriter.java Normal file
View File

@@ -0,0 +1,16 @@
package org.springframework.batch.core.configuration.xml;
import java.util.List;
import org.springframework.batch.item.ItemWriter;
/**
* @author Dan Garrette
* @since 2.0
*/
public class DummyItemWriter implements ItemWriter<Object> {
public void write(List<? extends Object> items) throws Exception {
}
}

86
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserTests.java
View File

@@ -19,45 +19,94 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.lang.reflect.Field;
import java.util.Map;
import org.junit.Test;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.step.item.ChunkOrientedTasklet;
import org.springframework.batch.core.step.item.ChunkProvider;
import org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean;
import org.springframework.batch.core.step.item.SimpleChunkProvider;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.repeat.CompletionPolicy;
import org.springframework.batch.repeat.RepeatOperations;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
* @author Thomas Risberg
*/
public class StepParserTests {
@SuppressWarnings("unchecked")
@Test
public void testTaskletStepAttributes() throws Exception {
ConfigurableApplicationContext ctx =
new ClassPathXmlApplicationContext("org/springframework/batch/core/configuration/xml/StepParserTaskletAttributesTests-context.xml");
ConfigurableApplicationContext ctx = new ClassPathXmlApplicationContext(
"org/springframework/batch/core/configuration/xml/StepParserTaskletAttributesTests-context.xml");
Map<String, Object> beans = ctx.getBeansOfType(FaultTolerantStepFactoryBean.class);
String factoryName = (String) beans.keySet().toArray()[0];
FaultTolerantStepFactoryBean<Object, Object> factory = (FaultTolerantStepFactoryBean<Object, Object>) beans.get(factoryName);
FaultTolerantStepFactoryBean<Object, Object> factory = (FaultTolerantStepFactoryBean<Object, Object>) beans
.get(factoryName);
TaskletStep bean = (TaskletStep) factory.getObject();
assertEquals("wrong start-limit:", 25, bean.getStartLimit());
}
@SuppressWarnings("unchecked")
@Test
public void testStepParserBeanName() throws Exception {
ConfigurableApplicationContext ctx =
new ClassPathXmlApplicationContext("org/springframework/batch/core/configuration/xml/StepParserBeanNameTests-context.xml");
ConfigurableApplicationContext ctx = new ClassPathXmlApplicationContext(
"org/springframework/batch/core/configuration/xml/StepParserBeanNameTests-context.xml");
Map<String, Object> beans = ctx.getBeansOfType(Step.class);
assertTrue("'s1' bean not found", beans.containsKey("s1"));
Step s1 = (Step)ctx.getBean("s1");
Step s1 = (Step) ctx.getBean("s1");
assertEquals("wrong name", "s1", s1.getName());
}
@Test(expected = BeanDefinitionParsingException.class)
public void testStepParserCommitIntervalCompletionPolicy() throws Exception {
new ClassPathXmlApplicationContext(
"org/springframework/batch/core/configuration/xml/StepParserCommitIntervalCompletionPolicyTests-context.xml");
}
@SuppressWarnings("unchecked")
@Test
public void testStepParserCommitInterval() throws Exception {
ConfigurableApplicationContext ctx = new ClassPathXmlApplicationContext(
"org/springframework/batch/core/configuration/xml/StepParserCommitIntervalTests-context.xml");
Map<String, Object> beans = ctx.getBeansOfType(Step.class);
assertTrue("'s1' bean not found", beans.containsKey("s1"));
Step s1 = (Step) ctx.getBean("s1");
CompletionPolicy completionPolicy = getCompletionPolicy(s1);
assertTrue(completionPolicy instanceof SimpleCompletionPolicy);
Field chunkSizeField = SimpleCompletionPolicy.class.getDeclaredField("chunkSize");
chunkSizeField.setAccessible(true);
assertEquals(25, chunkSizeField.get(completionPolicy));
}
@SuppressWarnings("unchecked")
@Test
public void testStepParserCompletionPolicy() throws Exception {
ConfigurableApplicationContext ctx = new ClassPathXmlApplicationContext(
"org/springframework/batch/core/configuration/xml/StepParserCompletionPolicyTests-context.xml");
Map<String, Object> beans = ctx.getBeansOfType(Step.class);
assertTrue("'s1' bean not found", beans.containsKey("s1"));
Step s1 = (Step) ctx.getBean("s1");
CompletionPolicy completionPolicy = getCompletionPolicy(s1);
System.err.println(completionPolicy);
assertTrue(completionPolicy instanceof DummyCompletionPolicy);
}
@Test(expected = BeanDefinitionParsingException.class)
public void testStepParserNoCommitIntervalOrCompletionPolicy() throws Exception {
new ClassPathXmlApplicationContext(
"org/springframework/batch/core/configuration/xml/StepParserNoCommitIntervalOrCompletionPolicyTests-context.xml");
}
@Test
public void testTaskletStepWithBadStepListener() throws Exception {
loadContextWithBadListener("org/springframework/batch/core/configuration/xml/StepParserBadStepListenerTests-context.xml");
@@ -72,10 +121,25 @@ public class StepParserTests {
try {
new ClassPathXmlApplicationContext(contextLocation);
fail("Context should not load!");
}
catch (BeanDefinitionParsingException e) {
} catch (BeanDefinitionParsingException e) {
assertTrue(e.getMessage().contains("'ref' and 'class'"));
}
}
@SuppressWarnings("unchecked")
private CompletionPolicy getCompletionPolicy(Step s1) throws NoSuchFieldException, IllegalAccessException {
Field taskletField = TaskletStep.class.getDeclaredField("tasklet");
taskletField.setAccessible(true);
Tasklet tasklet = (Tasklet) taskletField.get(s1);
Field chunkProviderField = ChunkOrientedTasklet.class.getDeclaredField("chunkProvider");
chunkProviderField.setAccessible(true);
ChunkProvider chunkProvider = (ChunkProvider) chunkProviderField.get(tasklet);
Field repeatOperationsField = SimpleChunkProvider.class.getDeclaredField("repeatOperations");
repeatOperationsField.setAccessible(true);
RepeatOperations repeatOperations = (RepeatOperations) repeatOperationsField.get(chunkProvider);
Field completionPolicyField = RepeatTemplate.class.getDeclaredField("completionPolicy");
completionPolicyField.setAccessible(true);
return (CompletionPolicy) completionPolicyField.get(repeatOperations);
}
}

18
spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserCommitIntervalCompletionPolicyTests-context.xml Normal file
View File

@@ -0,0 +1,18 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.0.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
<beans:import resource="common-context.xml" />
<job id="job">
<step id="s1">
<tasklet reader="reader" writer="writer"
commit-interval="25" chunk-completion-policy="completionPolicy"/>
</step>
</job>
<beans:bean id="completionPolicy" class="org.springframework.batch.core.configuration.xml.DummyCompletionPolicy"/>
</beans:beans>

16
spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserCommitIntervalTests-context.xml Normal file
View File

@@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.0.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
<beans:import resource="common-context.xml" />
<job id="job">
<step id="s1">
<tasklet reader="reader" writer="writer"
commit-interval="25"/>
</step>
</job>
</beans:beans>

18
spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserCompletionPolicyTests-context.xml Normal file
View File

@@ -0,0 +1,18 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.0.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
<beans:import resource="common-context.xml" />
<job id="job">
<step id="s1">
<tasklet reader="reader" writer="writer"
chunk-completion-policy="completionPolicy"/>
</step>
</job>
<beans:bean id="completionPolicy" class="org.springframework.batch.core.configuration.xml.DummyCompletionPolicy"/>
</beans:beans>

15
spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserNoCommitIntervalOrCompletionPolicyTests-context.xml Normal file
View File

@@ -0,0 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.0.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
<beans:import resource="common-context.xml" />
<job id="job">
<step id="s1">
<tasklet reader="reader" writer="writer"/>
</step>
</job>
</beans:beans>

3
spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/common-context.xml
View File

@@ -34,4 +34,7 @@
<bean id="stepNamesList" class="java.util.ArrayList"/>
<bean id="reader" class="org.springframework.batch.core.configuration.xml.DummyItemReader"/>
<bean id="writer" class="org.springframework.batch.core.configuration.xml.DummyItemWriter"/>
</beans>

2
spring-batch-samples/src/main/resources/jobs/multilineOrderJob.xml
View File

@@ -14,7 +14,7 @@
<job id="multilineOrderJob">
<step id="step1">
<tasklet reader="reader" processor="processor" writer="fileItemWriter" commit-interval="">
<tasklet reader="reader" processor="processor" writer="fileItemWriter" commit-interval="5">
<streams>
<stream ref="fileItemWriter"/>
<stream ref="fileItemReader"/>