diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/DecisionParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/DecisionParser.java index c893780a3..fb97b1ea2 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/DecisionParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/DecisionParser.java @@ -17,10 +17,10 @@ package org.springframework.batch.core.configuration.xml; import java.util.Collection; +import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; - import org.w3c.dom.Element; /** @@ -46,7 +46,7 @@ public class DecisionParser { * {@link org.springframework.batch.core.job.flow.support.StateTransition} * instances objects */ - public Collection parse(Element element, ParserContext parserContext) { + public Collection parse(Element element, ParserContext parserContext) { String refAttribute = element.getAttribute("decider"); String idAttribute = element.getAttribute("id"); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/FlowParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/FlowParser.java index 0021a1c3e..9edd9b72b 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/FlowParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/FlowParser.java @@ -23,10 +23,7 @@ import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.job.flow.support.SimpleFlow; import org.springframework.beans.factory.config.BeanDefinition; -import org.springframework.beans.factory.config.RuntimeBeanReference; -import org.springframework.beans.factory.parsing.BeanComponentDefinition; import org.springframework.beans.factory.parsing.CompositeComponentDefinition; -import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.ManagedList; import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; @@ -78,7 +75,7 @@ public class FlowParser extends AbstractSingleBeanDefinitionParser { */ @Override protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { - List stateTransitions = new ArrayList(); + List stateTransitions = new ArrayList(); InlineStepParser stepParser = new InlineStepParser(); DecisionParser decisionParser = new DecisionParser(); @@ -125,10 +122,10 @@ public class FlowParser extends AbstractSingleBeanDefinitionParser { * {@link org.springframework.batch.core.job.flow.support.StateTransition} * references */ - protected static Collection getNextElements(ParserContext parserContext, BeanDefinition stateDef, + protected static Collection getNextElements(ParserContext parserContext, BeanDefinition stateDef, Element element) { - Collection list = new ArrayList(); + Collection list = new ArrayList(); String shortNextAttribute = element.getAttribute(NEXT); boolean hasNextAttribute = StringUtils.hasText(shortNextAttribute); @@ -188,7 +185,7 @@ public class FlowParser extends AbstractSingleBeanDefinitionParser { * {@link org.springframework.batch.core.job.flow.support.StateTransition} * references */ - private static Collection parseTransitionElement(Element transitionElement, + private static Collection parseTransitionElement(Element transitionElement, BeanDefinition stateDef, ParserContext parserContext) { BatchStatus batchStatus = getBatchStatusFromEndTransitionName(transitionElement.getNodeName()); @@ -214,10 +211,10 @@ public class FlowParser extends AbstractSingleBeanDefinitionParser { * {@link org.springframework.batch.core.job.flow.support.StateTransition} * references */ - private static Collection createTransition(BatchStatus batchStatus, String on, String next, + private static Collection createTransition(BatchStatus batchStatus, String on, String next, String exitCode, BeanDefinition stateDef, ParserContext parserContext) { - RuntimeBeanReference endState = null; + BeanDefinition endState = null; if (batchStatus == BatchStatus.STOPPED || batchStatus == BatchStatus.COMPLETED || batchStatus == BatchStatus.FAILED) { @@ -238,7 +235,7 @@ public class FlowParser extends AbstractSingleBeanDefinitionParser { } - Collection list = new ArrayList(); + Collection list = new ArrayList(); list.add(getStateTransitionReference(parserContext, stateDef, on, next)); if (endState != null) { // @@ -276,7 +273,7 @@ public class FlowParser extends AbstractSingleBeanDefinitionParser { * @param next the next step id * @return a bean definition for a {@link org.springframework.batch.core.job.flow.support.StateTransition} */ - public static RuntimeBeanReference getStateTransitionReference(ParserContext parserContext, + public static BeanDefinition getStateTransitionReference(ParserContext parserContext, BeanDefinition stateDefinition, String on, String next) { BeanDefinitionBuilder nextBuilder = @@ -295,13 +292,7 @@ public class FlowParser extends AbstractSingleBeanDefinitionParser { nextBuilder.setFactoryMethod("createEndStateTransition"); } - // TODO: do we need to use RuntimeBeanReference? - AbstractBeanDefinition nextDef = nextBuilder.getBeanDefinition(); - String nextDefName = parserContext.getReaderContext().generateBeanName(nextDef); - BeanComponentDefinition nextDefComponent = new BeanComponentDefinition(nextDef, nextDefName); - parserContext.registerBeanComponent(nextDefComponent); - - return new RuntimeBeanReference(nextDefName); + return nextBuilder.getBeanDefinition(); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/InlineStepParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/InlineStepParser.java index bdbb5b6c7..785e4827f 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/InlineStepParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/InlineStepParser.java @@ -18,7 +18,7 @@ package org.springframework.batch.core.configuration.xml; import java.util.Collection; import java.util.List; -import org.springframework.beans.factory.config.RuntimeBeanReference; +import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.parsing.BeanComponentDefinition; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; @@ -49,7 +49,7 @@ public class InlineStepParser extends AbstractStepParser { * @return a collection of bean definitions for {@link org.springframework.batch.core.job.flow.support.StateTransition} * instances objects */ - public Collection parse(Element element, ParserContext parserContext) { + public Collection parse(Element element, ParserContext parserContext) { BeanDefinitionBuilder stateBuilder = BeanDefinitionBuilder.genericBeanDefinition("org.springframework.batch.core.job.flow.support.state.StepState"); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SplitParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SplitParser.java index 7ebfcf8a5..46ef874d7 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SplitParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SplitParser.java @@ -20,12 +20,10 @@ import java.util.Collection; import java.util.List; import org.springframework.beans.factory.config.BeanDefinition; -import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.ManagedList; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.util.xml.DomUtils; - import org.w3c.dom.Element; /** @@ -51,7 +49,7 @@ public class SplitParser { * {@link org.springframework.batch.core.job.flow.support.StateTransition} * instances objects */ - public Collection parse(Element element, ParserContext parserContext) { + public Collection parse(Element element, ParserContext parserContext) { String idAttribute = element.getAttribute("id"); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParser.java new file mode 100644 index 000000000..bd8e411aa --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParser.java @@ -0,0 +1,572 @@ +/* + * Copyright 2006-2009 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.batch.core.configuration.xml; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.BeanReference; +import org.springframework.beans.factory.config.RuntimeBeanReference; +import org.springframework.beans.factory.parsing.BeanComponentDefinition; +import org.springframework.beans.factory.parsing.CompositeComponentDefinition; +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.ManagedList; +import org.springframework.beans.factory.support.ManagedMap; +import org.springframework.beans.factory.support.RootBeanDefinition; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.util.StringUtils; +import org.springframework.util.xml.DomUtils; +import org.w3c.dom.Element; +import org.w3c.dom.NamedNodeMap; + +/** + * Internal parser for the <step/> elements inside a job. A step element + * references a bean definition for a {@link org.springframework.batch.core.Step} and goes on to (optionally) + * list a set of transitions from that step to others with <next on="pattern" + * to="stepName"/>. Used by the {@link JobParser}. + * + * @see JobParser + * + * @author Dave Syer + * @author Thomas Risberg + * @since 2.0 + */ +public class StepParser { + + // For generating unique state names for end transitions + private static int endCounter = 0; + + /** + * Parse the step and turn it into a list of transitions. + * + * @param element the <step/gt; element to parse + * @param parserContext the parser context for the bean factory + * @return a collection of bean definitions for {@link org.springframework.batch.core.job.flow.support.StateTransition} + * instances objects + */ + public Collection parse(Element element, ParserContext parserContext) { + + BeanDefinitionBuilder stateBuilder = + BeanDefinitionBuilder.genericBeanDefinition("org.springframework.batch.core.job.flow.support.state.StepState"); + String stepRef = element.getAttribute("name"); + String taskletRef = element.getAttribute("tasklet"); + + if (!StringUtils.hasText(stepRef)) { + parserContext.getReaderContext().error("The name attribute can't be empty for <" + element.getNodeName() + ">", element); + } + + @SuppressWarnings("unchecked") + List processTaskElements = (List) DomUtils.getChildElementsByTagName(element, "tasklet"); + if (StringUtils.hasText(taskletRef)) { + handleTaskletRef(element, taskletRef, parserContext); + stateBuilder.addConstructorArgReference(stepRef); + } + else if (processTaskElements.size() > 0) { + Element taskElement = processTaskElements.get(0); + handleTaskletElement(element, taskElement, parserContext); + stateBuilder.addConstructorArgReference(stepRef); + } + else if (StringUtils.hasText(stepRef)) { + stateBuilder.addConstructorArgReference(stepRef); + } + else { + parserContext.getReaderContext().error("Incomplete configuration detected while creating step with name " + stepRef, element); + } + return getNextElements(parserContext, stateBuilder.getBeanDefinition(), element); + + } + + /** + * @param parserContext + * @param stateDef + * @param element + * @return a collection of {@link org.springframework.batch.core.job.flow.support.StateTransition} references + */ + public static Collection getNextElements(ParserContext parserContext, + BeanDefinition stateDef, Element element) { + + Collection list = new ArrayList(); + + String shortNextAttribute = element.getAttribute("next"); + boolean hasNextAttribute = StringUtils.hasText(shortNextAttribute); + if (hasNextAttribute) { + list.add(getStateTransitionReference(parserContext, stateDef, null, shortNextAttribute)); + } + + @SuppressWarnings("unchecked") + List nextElements = (List) DomUtils.getChildElementsByTagName(element, "next"); + @SuppressWarnings("unchecked") + List stopElements = (List) DomUtils.getChildElementsByTagName(element, "stop"); + nextElements.addAll(stopElements); + @SuppressWarnings("unchecked") + List endElements = (List) DomUtils.getChildElementsByTagName(element, "end"); + nextElements.addAll(endElements); + + for (Element nextElement : nextElements) { + String onAttribute = nextElement.getAttribute("on"); + String nextAttribute = nextElement.getAttribute("to"); + if (hasNextAttribute && onAttribute.equals("*")) { + parserContext.getReaderContext().error("Duplicate transition pattern found for '*' " + + "(only specify one of next= attribute at step level and next element with on='*')", + element); + } + + BeanDefinition additionalState = null; + + String name = nextElement.getNodeName(); + if ("stop".equals(name) || "end".equals(name)) { + + String statusName = nextElement.getAttribute("status"); + String status = StringUtils.hasText(statusName) ? statusName : "STOPPED"; + String nextOnEnd = StringUtils.hasText(statusName) ? null : nextAttribute; + + BeanDefinitionBuilder endBuilder = + BeanDefinitionBuilder.genericBeanDefinition("org.springframework.batch.core.job.flow.support.state.EndState"); + endBuilder.addConstructorArgValue(status); + String endName = "stop".equals(name) ? "end" + (endCounter++) : null; + + endBuilder.addConstructorArgValue(endName); + additionalState = getStateTransitionReference(parserContext, endBuilder.getBeanDefinition(), onAttribute, nextOnEnd); + nextAttribute = endName; + + } + list.add(getStateTransitionReference(parserContext, stateDef, onAttribute, nextAttribute)); + if(additionalState != null) + { + // + // Must be added after the state to ensure that the state is the first in the list + // + list.add(additionalState); + } + } + + if(hasNextAttribute && nextElements.isEmpty()) + { + list.add(getStateTransitionReference(parserContext, stateDef, "FAILED", null)); + } + + if (list.isEmpty() && !hasNextAttribute) { + list.add(getStateTransitionReference(parserContext, stateDef, null, null)); + } + + return list; + } + + /** + * @param parserContext the parser context + * @param stateDefinition a reference to the state implementation + * @param on the pattern value + * @param next the next step id + * @return a bean definition for a {@link org.springframework.batch.core.job.flow.support.StateTransition} + */ + public static BeanDefinition getStateTransitionReference(ParserContext parserContext, + BeanDefinition stateDefinition, String on, String next) { + + BeanDefinitionBuilder nextBuilder = + BeanDefinitionBuilder.genericBeanDefinition("org.springframework.batch.core.job.flow.support.StateTransition"); + nextBuilder.addConstructorArgValue(stateDefinition); + + if (StringUtils.hasText(on)) { + nextBuilder.addConstructorArgValue(on); + } + + if (StringUtils.hasText(next)) { + nextBuilder.setFactoryMethod("createStateTransition"); + nextBuilder.addConstructorArgValue(next); + } + else { + nextBuilder.setFactoryMethod("createEndStateTransition"); + } + + nextBuilder.setRole(BeanDefinition.ROLE_INFRASTRUCTURE); + return nextBuilder.getBeanDefinition(); + + } + + /** + * @param stepElement + * @param taskletRef + * @param parserContext + */ + private void handleTaskletRef(Element stepElement, String taskletRef, ParserContext parserContext) { + + RootBeanDefinition bd = new RootBeanDefinition("org.springframework.batch.core.step.tasklet.TaskletStep", null, null); + + if (StringUtils.hasText(taskletRef)) { + RuntimeBeanReference taskletBeanRef = new RuntimeBeanReference(taskletRef); + bd.getPropertyValues().addPropertyValue("tasklet", taskletBeanRef); + } + + String jobRepositoryRef = stepElement.getAttribute("job-repository"); + RuntimeBeanReference jobRepositoryBeanRef = new RuntimeBeanReference(jobRepositoryRef); + bd.getPropertyValues().addPropertyValue("jobRepository", jobRepositoryBeanRef); + + String transactionManagerRef = stepElement.getAttribute("transaction-manager"); + RuntimeBeanReference transactionManagerBeanRef = new RuntimeBeanReference(transactionManagerRef); + bd.getPropertyValues().addPropertyValue("transactionManager", transactionManagerBeanRef); + + handleListenersElement(stepElement, bd, parserContext, "stepExecutionListeners"); + + bd.setRole(BeanDefinition.ROLE_SUPPORT); + + bd.setSource(parserContext.extractSource(stepElement)); + parserContext.registerBeanComponent(new BeanComponentDefinition(bd, stepElement.getAttribute("name"))); + + } + + /** + * @param element + * @param parserContext + */ + private void handleTaskletElement(Element stepElement, Element element, ParserContext parserContext) { + + RootBeanDefinition bd; + + boolean isFaultTolerant = false; + + String skipLimit = element.getAttribute("skip-limit"); + if (!isFaultTolerant) { + isFaultTolerant = checkIntValueForFaultToleranceNeeded(skipLimit); + } + String retryLimit = element.getAttribute("retry-limit"); + if (!isFaultTolerant) { + isFaultTolerant = checkIntValueForFaultToleranceNeeded(retryLimit); + } + String cacheCapacity = element.getAttribute("cache-capacity"); + if (!isFaultTolerant) { + isFaultTolerant = checkIntValueForFaultToleranceNeeded(cacheCapacity); + } + String isReaderTransactionalQueue = element.getAttribute("is-reader-transactional-queue"); + if (!isFaultTolerant && StringUtils.hasText(isReaderTransactionalQueue)) { + if ("true".equals(isReaderTransactionalQueue)) { + isFaultTolerant = true; + } + } + checkExceptionElementForFaultToleranceNeeded(element, "skippable-exception-classes"); + checkExceptionElementForFaultToleranceNeeded(element, "retryable-exception-classes"); + checkExceptionElementForFaultToleranceNeeded(element, "fatal-exception-classes"); + + if (isFaultTolerant) { + bd = new RootBeanDefinition("org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean", null, null); + } + else { + bd = new RootBeanDefinition("org.springframework.batch.core.step.item.SimpleStepFactoryBean", null, null); + } + + // now, set the properties on the new bean + String startLimit = element.getAttribute("start-limit"); + if (StringUtils.hasText(startLimit)) { + bd.getPropertyValues().addPropertyValue("startLimit", startLimit); + } + String allowStartIfComplete = element.getAttribute("allow-start-if-complete"); + if (StringUtils.hasText(allowStartIfComplete)) { + bd.getPropertyValues().addPropertyValue("allowStartIfComplete", allowStartIfComplete); + } + + String readerBeanId = element.getAttribute("reader"); + if (StringUtils.hasText(readerBeanId)) { + RuntimeBeanReference readerRef = new RuntimeBeanReference(readerBeanId); + bd.getPropertyValues().addPropertyValue("itemReader", readerRef); + } + + String processorBeanId = element.getAttribute("processor"); + if (StringUtils.hasText(processorBeanId)) { + RuntimeBeanReference processorRef = new RuntimeBeanReference(processorBeanId); + bd.getPropertyValues().addPropertyValue("itemProcessor", processorRef); + } + + String writerBeanId = element.getAttribute("writer"); + if (StringUtils.hasText(writerBeanId)) { + RuntimeBeanReference writerRef = new RuntimeBeanReference(writerBeanId); + bd.getPropertyValues().addPropertyValue("itemWriter", writerRef); + } + + String taskExecutorBeanId = element.getAttribute("task-executor"); + if (StringUtils.hasText(taskExecutorBeanId)) { + RuntimeBeanReference taskExecutorRef = new RuntimeBeanReference(taskExecutorBeanId); + bd.getPropertyValues().addPropertyValue("taskExecutor", taskExecutorRef); + } + + String jobRepository = stepElement.getAttribute("job-repository"); + RuntimeBeanReference jobRepositoryRef = new RuntimeBeanReference(jobRepository); + bd.getPropertyValues().addPropertyValue("jobRepository", jobRepositoryRef); + + String transactionManager = stepElement.getAttribute("transaction-manager"); + RuntimeBeanReference tx = new RuntimeBeanReference(transactionManager); + bd.getPropertyValues().addPropertyValue("transactionManager", tx); + + String commitInterval = element.getAttribute("commit-interval"); + if (StringUtils.hasText(commitInterval)) { + bd.getPropertyValues().addPropertyValue("commitInterval", commitInterval); + } + + if (StringUtils.hasText(skipLimit)) { + bd.getPropertyValues().addPropertyValue("skipLimit", skipLimit); + } + + if (StringUtils.hasText(retryLimit)) { + bd.getPropertyValues().addPropertyValue("retryLimit", retryLimit); + } + + if (StringUtils.hasText(cacheCapacity)) { + bd.getPropertyValues().addPropertyValue("cacheCapacity", cacheCapacity); + } + + String transactionAttribute = element.getAttribute("transaction-attribute"); + if (StringUtils.hasText(transactionAttribute)) { + bd.getPropertyValues().addPropertyValue("transactionAttribute", transactionAttribute); + } + + if (StringUtils.hasText(isReaderTransactionalQueue)) { + if (isFaultTolerant) { + bd.getPropertyValues().addPropertyValue("isReaderTransactionalQueue", isReaderTransactionalQueue); + } + } + + handleExceptionElement(element, parserContext, bd, "skippable-exception-classes", "skippableExceptionClasses", isFaultTolerant); + + handleExceptionElement(element, parserContext, bd, "retryable-exception-classes", "retryableExceptionClasses", isFaultTolerant); + + handleExceptionElement(element, parserContext, bd, "fatal-exception-classes", "fatalExceptionClasses", isFaultTolerant); + + handleListenersElement(stepElement, bd, parserContext, "listeners"); + + handleRetryListenersElement(element, bd, parserContext); + + handleStreamsElement(element, bd, parserContext); + + bd.setRole(BeanDefinition.ROLE_SUPPORT); + + bd.setSource(parserContext.extractSource(stepElement)); + parserContext.registerBeanComponent(new BeanComponentDefinition(bd, stepElement.getAttribute("name"))); + + } + + private boolean checkIntValueForFaultToleranceNeeded(String stringValue) { + if (StringUtils.hasText(stringValue)) { + int value = Integer.valueOf(stringValue); + if (value > 0) { + return true; + } + } + return false; + } + + private boolean checkExceptionElementForFaultToleranceNeeded(Element element, String subElementName) { + String exceptions = + DomUtils.getChildElementValueByTagName(element, subElementName); + if (StringUtils.hasLength(exceptions)) { + return true; + } + return false; + } + + private void handleExceptionElement(Element element, ParserContext parserContext, BeanDefinition bd, + String subElementName, String propertyName, boolean isFaultTolerant) { + String exceptions = + DomUtils.getChildElementValueByTagName(element, subElementName); + if (StringUtils.hasLength(exceptions)) { + if (isFaultTolerant) { + String[] exceptionArray = StringUtils.tokenizeToStringArray( + StringUtils.delete(exceptions, ","), "\n"); + if (exceptionArray.length > 0) { + bd.getPropertyValues().addPropertyValue(propertyName, exceptionArray); + } + } + else { + parserContext.getReaderContext().error(subElementName + " can only be specified for fault-tolerant " + + "configurations providing skip-limit, retry-limit or cache-capacity", element); + } + } + } + + @SuppressWarnings("unchecked") + private void handleListenersElement(Element element, BeanDefinition bd, ParserContext parserContext, String property) { + Element listenersElement = + DomUtils.getChildElementByTagName(element, "listeners"); + if (listenersElement != null) { + CompositeComponentDefinition compositeDef = + new CompositeComponentDefinition(listenersElement.getTagName(), parserContext.extractSource(element)); + parserContext.pushContainingComponent(compositeDef); + List listenerBeans = new ArrayList(); + handleStepListenerElements(parserContext, listenersElement, + listenerBeans); + ManagedList arguments = new ManagedList(); + arguments.addAll(listenerBeans); + bd.getPropertyValues().addPropertyValue(property, arguments); + parserContext.popAndRegisterContainingComponent(); + } + } + + @SuppressWarnings("unchecked") + private void handleRetryListenersElement(Element element, BeanDefinition bd, ParserContext parserContext) { + Element listenersElement = + DomUtils.getChildElementByTagName(element, "retry-listeners"); + if (listenersElement != null) { + CompositeComponentDefinition compositeDef = + new CompositeComponentDefinition(listenersElement.getTagName(), parserContext.extractSource(element)); + parserContext.pushContainingComponent(compositeDef); + List retryListenerBeans = new ArrayList(); + handleRetryListenerElements(parserContext, listenersElement, + retryListenerBeans); + ManagedList arguments = new ManagedList(); + arguments.addAll(retryListenerBeans); + bd.getPropertyValues().addPropertyValue("retryListeners", arguments); + parserContext.popAndRegisterContainingComponent(); + } + } + + @SuppressWarnings("unchecked") + private void handleRetryListenerElements(ParserContext parserContext, + Element element, List beans) { + List listenerElements = + DomUtils.getChildElementsByTagName(element, "listener"); + if (listenerElements != null) { + for (Element listenerElement : listenerElements) { + String id = listenerElement.getAttribute("id"); + String listenerRef = listenerElement.getAttribute("ref"); + String className = listenerElement.getAttribute("class"); + checkListenerElementAttributes(parserContext, element, + listenerElement, id, listenerRef, className); + if (StringUtils.hasText(listenerRef)) { + BeanReference bean = new RuntimeBeanReference(listenerRef); + beans.add(bean); + } + else if (StringUtils.hasText(className)) { + RootBeanDefinition beanDef = new RootBeanDefinition(className, null, null); + if (!StringUtils.hasText(id)) { + id = parserContext.getReaderContext().generateBeanName(beanDef); + } + beans.add(beanDef); + } + else { + parserContext.getReaderContext().error("Neither 'ref' or 'class' specified for <" + listenerElement.getTagName() + "> element", element); + } + } + } + } + + @SuppressWarnings("unchecked") + private void handleStepListenerElements(ParserContext parserContext, + Element element, List beans) { + List listenerElements = + DomUtils.getChildElementsByTagName(element, "listener"); + if (listenerElements != null) { + for (Element listenerElement : listenerElements) { + BeanDefinitionBuilder listenerBuilder = + BeanDefinitionBuilder.genericBeanDefinition("org.springframework.batch.core.listener.StepListenerFactoryBean"); + String id = listenerElement.getAttribute("id"); + String listenerRef = listenerElement.getAttribute("ref"); + String className = listenerElement.getAttribute("class"); + checkListenerElementAttributes(parserContext, element, + listenerElement, id, listenerRef, className); + if (StringUtils.hasText(listenerRef)) { + listenerBuilder.addPropertyReference("delegate", listenerRef); + } + else if (StringUtils.hasText(className)) { + RootBeanDefinition beanDef = new RootBeanDefinition(className, null, null); + listenerBuilder.addPropertyValue("delegate", beanDef); + } + else { + parserContext.getReaderContext().error("Neither 'ref' or 'class' specified for <" + listenerElement.getTagName() + "> element", element); + } + + ManagedMap metaDataMap = new ManagedMap(); + String[] methodNameAttributes = new String[] { + "before-step-method", + "after-step-method", + "before-chunk-method", + "after-chunk-method", + "before-read-method", + "after-read-method", + "on-read-error-method", + "before-process-method", + "after-process-method", + "on-process-error-method", + "before-write-method", + "after-write-method", + "on-write-error-method", + "on-skip-in-read-method", + "on-skip-in-process-method", + "on-skip-in-write-method" + }; + for (String metaDataPropertyName : methodNameAttributes) { + String listenerMethod = listenerElement.getAttribute(metaDataPropertyName); + if(StringUtils.hasText(listenerMethod)){ + metaDataMap.put(metaDataPropertyName, listenerMethod); + } + } + listenerBuilder.addPropertyValue("metaDataMap", metaDataMap); + + AbstractBeanDefinition beanDef = listenerBuilder.getBeanDefinition(); + if (!StringUtils.hasText(id)) { + id = parserContext.getReaderContext().generateBeanName(beanDef); + } + beans.add(beanDef); + } + } + } + + private void checkListenerElementAttributes(ParserContext parserContext, + Element element, Element listenerElement, String id, + String listenerRef, String className) { + if ((StringUtils.hasText(id) || StringUtils.hasText(className)) + && StringUtils.hasText(listenerRef)) { + NamedNodeMap attributeNodes = listenerElement.getAttributes(); + StringBuilder attributes = new StringBuilder(); + for (int i = 0; i < attributeNodes.getLength(); i++) { + if (i > 0) { + attributes.append(" "); + } + attributes.append(attributeNodes.item(i)); + } + parserContext.getReaderContext().error("Both 'ref' and " + + (StringUtils.hasText(id) ? "'id'" : "'class'") + + " specified; use 'class' with an optional 'id' or just 'ref' for <" + + listenerElement.getTagName() + "> element specified with attributes: " + attributes, element); + } + } + + @SuppressWarnings("unchecked") + private void handleStreamsElement(Element element, BeanDefinition bd, ParserContext parserContext) { + Element streamsElement = + DomUtils.getChildElementByTagName(element, "streams"); + if (streamsElement != null) { + List streamBeans = new ArrayList(); + List streamElements = + DomUtils.getChildElementsByTagName(streamsElement, "stream"); + if (streamElements != null) { + for (Element streamElement : streamElements) { + String streamRef = streamElement.getAttribute("ref"); + if (StringUtils.hasText(streamRef)) { + BeanReference bean = new RuntimeBeanReference(streamRef); + streamBeans.add(bean); + } + else { + parserContext.getReaderContext().error("ref not specified for <" + streamElement.getTagName() + "> element", element); + } + } + } + ManagedList arguments = new ManagedList(); + arguments.addAll(streamBeans); + bd.getPropertyValues().addPropertyValue("streams", arguments); + } + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java index 79f2f8c53..5d2e77632 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java @@ -35,6 +35,7 @@ import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.listener.CompositeExecutionJobListener; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.repository.JobRestartException; +import org.springframework.batch.core.step.StepLocator; import org.springframework.batch.item.ExecutionContext; import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.InitializingBean; @@ -50,7 +51,7 @@ import org.springframework.util.ClassUtils; * @author Lucas Ward * @author Dave Syer */ -public abstract class AbstractJob implements Job, BeanNameAware, InitializingBean { +public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, InitializingBean { protected static final Log logger = LogFactory.getLog(AbstractJob.class); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/Flow.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/Flow.java index 9a8fa564b..8221cc917 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/Flow.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/Flow.java @@ -15,7 +15,6 @@ */ package org.springframework.batch.core.job.flow; -import org.springframework.batch.core.job.flow.support.State; /** * @author Dave Syer diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java index f261fab2c..4e0a1c783 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java @@ -24,10 +24,8 @@ import org.springframework.batch.core.StartLimitExceededException; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.job.AbstractJob; -import org.springframework.batch.core.job.flow.support.State; -import org.springframework.batch.core.job.flow.support.state.StepState; import org.springframework.batch.core.repository.JobRestartException; - +import org.springframework.batch.core.step.StepHolder; /** * @author Dave Syer @@ -36,7 +34,7 @@ import org.springframework.batch.core.repository.JobRestartException; public class FlowJob extends AbstractJob { private Flow flow; - + /** * Create a {@link FlowJob} with null name and no flow (invalid state). */ @@ -59,18 +57,20 @@ public class FlowJob extends AbstractJob { this.flow = flow; } - /* + /* * (non-Javadoc) - * @see org.springframework.batch.core.job.AbstractJob#getStep(java.lang.String) + * + * @see + * org.springframework.batch.core.job.AbstractJob#getStep(java.lang.String) */ - public Step getStep(String stepName){ + public Step getStep(String stepName) { State state = this.flow.getState(stepName); - if(state instanceof StepState){ - return ((StepState) state).getStep(); + if (state instanceof StepHolder) { + return ((StepHolder) state).getStep(); } return null; } - + /** * @see AbstractJob#doExecute(JobExecution) */ @@ -78,7 +78,7 @@ public class FlowJob extends AbstractJob { protected void doExecute(final JobExecution execution) throws JobExecutionException { try { FlowExecution flowExecution = flow.start(new JobFlowExecutor(execution)); - + synchronized (execution) { FlowExecutionStatus status = flowExecution.getStatus(); execution.upgradeStatus(status.getBatchStatus()); @@ -95,11 +95,12 @@ public class FlowJob extends AbstractJob { /** * @author Dave Syer - * + * */ private class JobFlowExecutor implements FlowExecutor { private final ThreadLocal stepExecutionHolder = new ThreadLocal(); + private final JobExecution execution; /** @@ -110,25 +111,27 @@ public class FlowJob extends AbstractJob { stepExecutionHolder.set(null); } - public String executeStep(Step step) throws JobInterruptedException, JobRestartException, StartLimitExceededException { + public String executeStep(Step step) throws JobInterruptedException, JobRestartException, + StartLimitExceededException { StepExecution lastStepExecution = stepExecutionHolder.get(); - if (lastStepExecution!=null && lastStepExecution.getStatus()==BatchStatus.FAILED) { + if (lastStepExecution != null && lastStepExecution.getStatus() == BatchStatus.FAILED) { lastStepExecution.setStatus(BatchStatus.INCOMPLETE); updateStepExecution(lastStepExecution); } StepExecution stepExecution = handleStep(step, execution); stepExecutionHolder.set(stepExecution); - return stepExecution==null ? ExitStatus.COMPLETED.getExitCode() : stepExecution.getExitStatus().getExitCode(); + return stepExecution == null ? ExitStatus.COMPLETED.getExitCode() : stepExecution.getExitStatus() + .getExitCode(); } public JobExecution getJobExecution() { return execution; } - + public StepExecution getStepExecution() { return stepExecutionHolder.get(); } - + public void close(FlowExecution result) { stepExecutionHolder.set(null); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/State.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/State.java new file mode 100644 index 000000000..fa99a50f7 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/State.java @@ -0,0 +1,48 @@ +/* + * Copyright 2006-2007 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.batch.core.job.flow; + + + +/** + * @author Dave Syer + * + */ +public interface State { + + /** + * The name of the state. Should be unique within a flow. + * + * @return the name of this state + */ + String getName(); + + /** + * Handle some business or processing logic and return a status that can be + * used to drive a flow to the next {@link State}. The status can be any + * string, but special meaning is assigned to the static constants in + * {@link FlowExecution}. The context can be used by implementations to do + * whatever they need to do. The same context will be passed to all + * {@link State} instances, so implementations should be careful that the + * context is thread safe, or used in a thread safe manner. + * + * @param executor the context passed in by the caller + * @return a status for the execution + * @throws Exception if anything goes wrong + */ + FlowExecutionStatus handle(FlowExecutor executor) throws Exception; + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/SimpleFlow.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/SimpleFlow.java index 2f78cb394..486990b2f 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/SimpleFlow.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/SimpleFlow.java @@ -30,6 +30,7 @@ import org.springframework.batch.core.job.flow.FlowExecution; import org.springframework.batch.core.job.flow.FlowExecutionException; import org.springframework.batch.core.job.flow.FlowExecutionStatus; import org.springframework.batch.core.job.flow.FlowExecutor; +import org.springframework.batch.core.job.flow.State; import org.springframework.beans.factory.InitializingBean; /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/StateTransition.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/StateTransition.java index 74c26eef5..d12b269f8 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/StateTransition.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/StateTransition.java @@ -16,6 +16,7 @@ package org.springframework.batch.core.job.flow.support; import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.job.flow.State; import org.springframework.batch.support.PatternMatcher; import org.springframework.util.StringUtils; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/AbstractState.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/AbstractState.java index 518066be5..1f6b3d30c 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/AbstractState.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/AbstractState.java @@ -17,7 +17,7 @@ package org.springframework.batch.core.job.flow.support.state; import org.springframework.batch.core.job.flow.FlowExecutionStatus; import org.springframework.batch.core.job.flow.FlowExecutor; -import org.springframework.batch.core.job.flow.support.State; +import org.springframework.batch.core.job.flow.State; /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/EndState.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/EndState.java index ae8d5e262..2c250092a 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/EndState.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/EndState.java @@ -21,7 +21,7 @@ import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.job.flow.FlowExecutionStatus; import org.springframework.batch.core.job.flow.FlowExecutor; -import org.springframework.batch.core.job.flow.support.State; +import org.springframework.batch.core.job.flow.State; /** * {@link State} implementation for ending a job if it is in progress and diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/SplitState.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/SplitState.java index 1d07216d9..01e24a5fb 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/SplitState.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/SplitState.java @@ -26,7 +26,7 @@ import org.springframework.batch.core.job.flow.FlowExecution; import org.springframework.batch.core.job.flow.FlowExecutionException; import org.springframework.batch.core.job.flow.FlowExecutionStatus; import org.springframework.batch.core.job.flow.FlowExecutor; -import org.springframework.batch.core.job.flow.support.State; +import org.springframework.batch.core.job.flow.State; import org.springframework.core.task.SyncTaskExecutor; import org.springframework.core.task.TaskExecutor; import org.springframework.core.task.TaskRejectedException; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/StepState.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/StepState.java index 288267605..eed66d639 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/StepState.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/StepState.java @@ -19,7 +19,8 @@ package org.springframework.batch.core.job.flow.support.state; import org.springframework.batch.core.Step; import org.springframework.batch.core.job.flow.FlowExecutionStatus; import org.springframework.batch.core.job.flow.FlowExecutor; -import org.springframework.batch.core.job.flow.support.State; +import org.springframework.batch.core.job.flow.State; +import org.springframework.batch.core.step.StepHolder; /** * {@link State} implementation that delegates to a {@link FlowExecutor} to @@ -28,7 +29,7 @@ import org.springframework.batch.core.job.flow.support.State; * @author Dave Syer * */ -public class StepState extends AbstractState { +public class StepState extends AbstractState implements StepHolder { private final Step step; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/StepHolder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/StepHolder.java new file mode 100644 index 000000000..b07d9fd44 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/StepHolder.java @@ -0,0 +1,16 @@ +package org.springframework.batch.core.step; + +import org.springframework.batch.core.Step; + +/** + * Interface for holders of a {@link Step} as a convenience for callers who need + * access to the underlying instance. + * + * @author Dave Syer + * + */ +public interface StepHolder { + + Step getStep(); + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/StepLocator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/StepLocator.java new file mode 100644 index 000000000..6f78f2678 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/StepLocator.java @@ -0,0 +1,15 @@ +package org.springframework.batch.core.step; + +import org.springframework.batch.core.Step; + +/** + * Interface for locating a {@link Step} instance by name. + * + * @author Dave Syer + * + */ +public interface StepLocator { + + Step getStep(String stepName); + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/support/SimpleFlowTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/support/SimpleFlowTests.java index 266fb3c05..834ae069d 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/support/SimpleFlowTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/support/SimpleFlowTests.java @@ -31,6 +31,7 @@ import org.springframework.batch.core.job.flow.FlowExecution; import org.springframework.batch.core.job.flow.FlowExecutionException; import org.springframework.batch.core.job.flow.FlowExecutionStatus; import org.springframework.batch.core.job.flow.FlowExecutor; +import org.springframework.batch.core.job.flow.State; /** * @author Dave Syer diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/support/StateSupport.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/support/StateSupport.java index fd1e1fd57..e7c5b45b6 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/support/StateSupport.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/support/StateSupport.java @@ -17,6 +17,7 @@ package org.springframework.batch.core.job.flow.support; import org.springframework.batch.core.job.flow.FlowExecutionStatus; import org.springframework.batch.core.job.flow.FlowExecutor; +import org.springframework.batch.core.job.flow.State; import org.springframework.batch.core.job.flow.support.state.AbstractState; /**