diff --git a/spring-batch-core/pom.xml b/spring-batch-core/pom.xml index 674cb81c4..df693a546 100644 --- a/spring-batch-core/pom.xml +++ b/spring-batch-core/pom.xml @@ -69,6 +69,14 @@ junit junit + + aspectj + aspectjrt + + + aspectj + aspectjweaver + cglib cglib-nodep diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepContext.java b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepContext.java new file mode 100644 index 000000000..c5cbd2111 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepContext.java @@ -0,0 +1,169 @@ +/* + * 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.scope; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.UnexpectedJobExecutionException; +import org.springframework.batch.repeat.context.SynchronizedAttributeAccessor; +import org.springframework.util.Assert; + +/** + * Simple implementation of {@link StepContext}. + * + * @author Dave Syer + * + */ +public class StepContext extends SynchronizedAttributeAccessor { + + private StepExecution stepExecution; + + private Map> callbacks = new HashMap>(); + + /** + * Create a new instance of {@link StepContext} for this + * {@link StepExecution}. + * + * @param stepExecution a step execution + */ + public StepContext(StepExecution stepExecution) { + super(); + Assert.notNull(stepExecution, "A StepContext must have a non-null StepExecution"); + this.stepExecution = stepExecution; + } + + /** + * Allow clients to register callbacks for clean up on close. + * + * @param name the callback id (unique attribute key in this context) + * @param callback a callback to execute on close + */ + public void registerDestructionCallback(String name, Runnable callback) { + synchronized (callbacks) { + Set set = callbacks.get(name); + if (set == null) { + set = new HashSet(); + callbacks.put(name, set); + } + set.add(callback); + } + } + + private void unregisterDestructionCallbacks(String name) { + synchronized (callbacks) { + callbacks.remove(name); + } + } + + /** + * Override base class behaviour to ensure destruction callbacks are + * unregistered as well as the default behaviour. + * + * @see SynchronizedAttributeAccessor#removeAttribute(String) + */ + @Override + public Object removeAttribute(String name) { + unregisterDestructionCallbacks(name); + return super.removeAttribute(name); + } + + /** + * Clean up the context at the end of a step execution. Must be called once + * at the end of a step execution to honour the destruction callback + * contract from the {@link StepScope}. + */ + public void close() { + + List errors = new ArrayList(); + + Map> copy = Collections.unmodifiableMap(callbacks); + + for (String key : copy.keySet()) { + Set set = copy.get(key); + for (Runnable callback : set) { + if (callback != null) { + /* + * The documentation of the interface says that these + * callbacks must not throw exceptions, but we don't trust + * them necessarily... + */ + try { + callback.run(); + } + catch (RuntimeException t) { + errors.add(t); + } + } + } + } + + if (errors.isEmpty()) { + return; + } + + Exception error = (Exception) errors.get(0); + if (error instanceof RuntimeException) { + throw (RuntimeException) error; + } + else { + throw new UnexpectedJobExecutionException("Could not close step context, rethrowing first of " + + errors.size() + " execptions.", error); + } + } + + public StepExecution getStepExecution() { + return stepExecution; + } + + /** + * Extend the base class method to include the step execution itself as a + * key (i.e. two contexts are only equal if their step executions are the + * same). + * + * @see SynchronizedAttributeAccessor#equals(Object) + */ + @Override + public boolean equals(Object other) { + if (!(other instanceof StepContext)) + return false; + if (other == this) + return true; + StepContext context = (StepContext) other; + if (context.stepExecution == stepExecution) { + return true; + } + return stepExecution.equals(context.stepExecution); + } + + /** + * Overrides the default behaviour to provide a hash code based only on the + * step execution. + * + * @see SynchronizedAttributeAccessor#hashCode() + */ + @Override + public int hashCode() { + return stepExecution.hashCode(); + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepContextRepeatCallback.java b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepContextRepeatCallback.java new file mode 100644 index 000000000..a4314cc92 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepContextRepeatCallback.java @@ -0,0 +1,66 @@ +/* + * 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.scope; + +import org.springframework.batch.core.Step; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.batch.repeat.RepeatCallback; +import org.springframework.batch.repeat.RepeatContext; + +/** + * Convenient base class for clients who need to do something in a repeat + * callback inside a {@link Step}. + * + * @author Dave Syer + * + */ +public abstract class StepContextRepeatCallback implements RepeatCallback { + + private final StepContext stepContext; + + /** + * @param stepContext + */ + public StepContextRepeatCallback(StepContext stepContext) { + this.stepContext = stepContext; + } + + /** + * Manage the {@link StepContext} lifecycle to ensure that the current + * thread has a reference to the context, even if the callback is executed + * in a pooled thread. + * + * @see RepeatCallback#doInIteration(RepeatContext) + */ + public ExitStatus doInIteration(RepeatContext context) throws Exception { + StepSynchronizationManager.register(stepContext); + try { + return doInStepContext(context, stepContext); + } + finally { + StepSynchronizationManager.close(); + } + } + + /** + * @param context + * @param stepContext + * @return the exit status from the execution + * @throws Exception + */ + public abstract ExitStatus doInStepContext(RepeatContext context, StepContext stepContext) throws Exception; + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepScope.java b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepScope.java new file mode 100644 index 000000000..c8380c039 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepScope.java @@ -0,0 +1,224 @@ +/* + * 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.scope; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.aop.scope.ScopedProxyUtils; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.ObjectFactory; +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.BeanDefinitionHolder; +import org.springframework.beans.factory.config.BeanDefinitionVisitor; +import org.springframework.beans.factory.config.BeanFactoryPostProcessor; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.config.Scope; +import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.core.Ordered; +import org.springframework.util.Assert; +import org.springframework.util.StringValueResolver; + +/** + * Scope for step context. Objects in this scope with <aop:scoped-proxy/> + * use the Spring container as an object factory, so there is only one instance + * of such a bean per executing step. + * + * @author Dave Syer + * + */ +public class StepScope implements Scope, BeanFactoryPostProcessor, Ordered { + + private Log logger = LogFactory.getLog(getClass()); + + private int order = Ordered.LOWEST_PRECEDENCE; + + private Object mutex = new Object(); + + /** + * @param order the order value to set priority of callback execution for + * the {@link BeanFactoryPostProcessor} part of this scope bean. + */ + public void setOrder(int order) { + this.order = order; + } + + public int getOrder() { + return order; + } + + /** + * Context key for clients to use for conversation identifier. + */ + public static final String ID_KEY = "STEP_IDENTIFIER"; + + private String name = "step"; + + private boolean proxyTargetClass; + + /** + * @see Scope#get(String, ObjectFactory) + */ + public Object get(String name, ObjectFactory objectFactory) { + + StepContext context = getContext(); + Object scopedObject = context.getAttribute(name); + + if (scopedObject == null) { + + synchronized (mutex) { + scopedObject = context.getAttribute(name); + if (scopedObject == null) { + + logger.debug(String.format("Creating object in scope=%s, name=%s", this.name, name)); + + /** + * Here is where we need to inject some context (a root + * object for expressions). The ObjectFactory could take a + * parameter? + */ + scopedObject = objectFactory.getObject(); + context.setAttribute(name, scopedObject); + + } + + } + + } + return scopedObject; + } + + /** + * @see Scope#getConversationId() + */ + public String getConversationId() { + StepContext context = getContext(); + Object id = context.getAttribute(ID_KEY); + return "" + id; + } + + /** + * @see Scope#registerDestructionCallback(String, Runnable) + */ + public void registerDestructionCallback(String name, Runnable callback) { + StepContext context = getContext(); + logger.debug(String.format("Registered destruction callback in scope=%s, name=%s", this.name, name)); + context.registerDestructionCallback(name, callback); + } + + /** + * @see Scope#remove(String) + */ + public Object remove(String name) { + StepContext context = getContext(); + logger.debug(String.format("Removing from scope=%s, name=%s", this.name, name)); + return context.removeAttribute(name); + } + + /** + * Get an attribute accessor in the form of a {@link StepContext} that can + * be used to store scoped bean instances. + * + * @return the current step context which we can use as a scope storage + * medium + */ + private StepContext getContext() { + StepContext context = StepSynchronizationManager.getContext(); + if (context == null) { + throw new IllegalStateException("No context holder available for step scope"); + } + return context; + } + + /** + * Register this scope with the enclosing BeanFactory. + * + * @see BeanFactoryPostProcessor#postProcessBeanFactory(ConfigurableListableBeanFactory) + * + * @param beanFactory the BeanFactory to register with + * @throws BeansException if there is a problem. + */ + public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException { + beanFactory.registerScope(name, this); + Assert.state(beanFactory instanceof BeanDefinitionRegistry, + "BeanFactory was not a BeanDefinitionRegistry, so StepScope cannot be used."); + Scopifier scopifier = new Scopifier((BeanDefinitionRegistry) beanFactory, name, proxyTargetClass); + for (String beanName : beanFactory.getBeanDefinitionNames()) { + BeanDefinition definition = beanFactory.getBeanDefinition(beanName); + // Replace this or any of its inner beans with scoped proxy if it + // has this scope + scopifier.visitBeanDefinition(definition); + } + } + + /** + * Public setter for the name property. This can then be used as a bean + * definition attribute, e.g. scope="step". Defaults to "step". + * + * @param name the name to set for this scope. + */ + public void setName(String name) { + this.name = name; + } + + /** + * Helper class to scan a bean definition hierarchy looking for scoped + * objects and modifying their properties. In particular it forces the use + * of auto-proxy for step scoped beans. + * + * @author Dave Syer + * + */ + private static class Scopifier extends BeanDefinitionVisitor { + + private final boolean proxyTargetClass; + + private final BeanDefinitionRegistry registry; + + private final String scope; + + public Scopifier(BeanDefinitionRegistry registry, String scope, boolean proxyTargetClass) { + super(new StringValueResolver() { + public String resolveStringValue(String value) { + return value; + } + }); + this.registry = registry; + this.proxyTargetClass = proxyTargetClass; + this.scope = scope; + } + + @Override + protected Object resolveValue(Object value) { + if (value instanceof BeanDefinition) { + BeanDefinition definition = (BeanDefinition) value; + if (scope.equals(definition.getScope())) { + String beanName = BeanDefinitionReaderUtils.generateBeanName(definition, registry); + return ScopedProxyUtils.createScopedProxy(new BeanDefinitionHolder(definition, beanName), registry, + proxyTargetClass); + } + } + else if (value instanceof BeanDefinitionHolder) { + BeanDefinitionHolder definition = (BeanDefinitionHolder) value; + if (scope.equals(definition.getBeanDefinition().getScope())) { + return ScopedProxyUtils.createScopedProxy(definition, registry, proxyTargetClass); + } + } + return value; + } + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepScopeManager.java b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepScopeManager.java new file mode 100644 index 000000000..f33c45037 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepScopeManager.java @@ -0,0 +1,32 @@ +package org.springframework.batch.core.scope; + +import org.aspectj.lang.annotation.Around; +import org.aspectj.lang.annotation.Aspect; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; + +/** + * Convenient aspect to wrap a single threaded step execution, where the + * implementation of the {@link Step} is not step scope aware (i.e. not the ones + * provided by the framework). + * + * @author Dave Syer + * + */ +// TODO: bin this? +@Aspect +public class StepScopeManager { + + @Around("execution(void org.springframework.batch.core.Step+.execute(*)) && target(step) && args(stepExecution)") + public void execute(Step step, StepExecution stepExecution) throws JobInterruptedException { + StepSynchronizationManager.register(new StepContext(stepExecution)); + try { + step.execute(stepExecution); + } + finally { + StepSynchronizationManager.release(); + } + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepSynchronizationManager.java b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepSynchronizationManager.java new file mode 100644 index 000000000..c0da1059e --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepSynchronizationManager.java @@ -0,0 +1,108 @@ +/* + * 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.scope; + +import java.util.Stack; + +import org.springframework.batch.core.Step; + +/** + * Central convenience class for framework use in managing the step scope + * context. Generally only to be used by implementations of {@link Step}. N.B. + * it is the responsibility of every {@link Step} implementation to ensure that + * a {@link StepContext} is available on every thread that might be involved in + * a step execution, including worker threads from a pool. + * + * @author Dave Syer + * + */ +public class StepSynchronizationManager { + + /** + * Don't use InheritableThreadLocal because there are side effects if a step + * is trying to run multiple child steps (e.g. with partitioning). + */ + private static final ThreadLocal> contextHolder = new ThreadLocal>(); + + /** + * Getter for the current context if there is one, otherwise returns null. + * + * @return the current {@link StepContext} or null if there is none (if one + * has not been registered for this thread). + */ + public static StepContext getContext() { + Stack current = getCurrent(); + if (current.isEmpty()) { + return null; + } + return current.peek(); + } + + /** + * Method for registering a context with the current thread - always put a + * matching {@link #close()} call in a finally block to ensure that the + * correct context is available in the enclosing block. + * + * @param context the step context to register + */ + public static void register(StepContext context) { + if (context == null) { + return; + } + getCurrent().push(context); + } + + /** + * Method for de-registering the current context - should always and only be + * used by in conjunction with a matching {@link #register(StepContext)} to + * ensure that {@link #getContext()} always returns the correct value. Does + * not call {@link StepContext#close()} - that is left up to the caller + * because he has a reference to the context (having registered it) and only + * he has knowledge of when the step actually ended. + */ + public static void close() { + StepContext oldSession = getContext(); + if (oldSession == null) { + return; + } + getCurrent().pop(); + } + + private static Stack getCurrent() { + if (contextHolder.get() == null) { + contextHolder.set(new Stack()); + } + return contextHolder.get(); + } + + /** + * A "deep" close operation. Call this instead of {@link #close()} if the + * step execution for the current context is ending. Delegates to + * {@link StepContext#close()} and then ensures that {@link #close()} is + * also called in a finally block. + */ + public static void release() { + StepContext context = getContext(); + try { + if (context!=null) { + context.close(); + } + } finally { + close(); + } + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java index 7307797e6..601e04c16 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java @@ -31,6 +31,8 @@ import org.springframework.batch.core.launch.NoSuchJobException; import org.springframework.batch.core.launch.support.ExitCodeMapper; import org.springframework.batch.core.listener.CompositeStepExecutionListener; import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.scope.StepContext; +import org.springframework.batch.core.scope.StepSynchronizationManager; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.repeat.ExitStatus; import org.springframework.beans.factory.BeanNameAware; @@ -143,12 +145,12 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw /** * Extension point for subclasses to execute business logic. * - * @param stepExecution + * @param stepContext the current step context * @return {@link ExitStatus} to show whether the step is finished * processing. * @throws Exception */ - protected abstract ExitStatus doExecute(StepExecution stepExecution) throws Exception; + protected abstract ExitStatus doExecute(StepContext stepContext) throws Exception; /** * Extension point for subclasses to provide callbacks to their @@ -175,7 +177,7 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw /** * Template method for step execution logic - calls abstract methods for * resource initialization ({@link #open(ExecutionContext)}), execution - * logic ({@link #doExecute(StepExecution)}) and resource closing ({@link #close(ExecutionContext)}). + * logic ({@link #doExecute(StepContext)}) and resource closing ({@link #close(ExecutionContext)}). */ public final void execute(StepExecution stepExecution) throws JobInterruptedException, UnexpectedJobExecutionException { @@ -186,11 +188,14 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw ExitStatus exitStatus = ExitStatus.FAILED; Exception commitException = null; + StepContext stepContext = new StepContext(stepExecution); + StepSynchronizationManager.register(stepContext); + try { getCompositeListener().beforeStep(stepExecution); open(stepExecution.getExecutionContext()); - exitStatus = doExecute(stepExecution); + exitStatus = doExecute(stepContext); // Check if someone is trying to stop us if (stepExecution.isTerminateOnly()) { @@ -256,6 +261,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw logger.error("Exception while closing step execution resources", e); stepExecution.addFailureException(e); } + + StepSynchronizationManager.release(); if (commitException != null) { stepExecution.setStatus(BatchStatus.UNKNOWN); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java index 09a505a49..d2b175cc8 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java @@ -26,6 +26,8 @@ import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.StepExecutionListener; import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.scope.StepContext; +import org.springframework.batch.core.scope.StepContextRepeatCallback; import org.springframework.batch.core.step.AbstractStep; import org.springframework.batch.core.step.StepExecutionSynchronizer; import org.springframework.batch.core.step.StepExecutionSynchronizerFactory; @@ -37,7 +39,6 @@ import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.support.CompositeItemStream; import org.springframework.batch.repeat.ExitStatus; -import org.springframework.batch.repeat.RepeatCallback; import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.RepeatOperations; import org.springframework.batch.repeat.support.RepeatTemplate; @@ -225,16 +226,19 @@ public class TaskletStep extends AbstractStep { * execution * */ - protected ExitStatus doExecute(final StepExecution stepExecution) throws Exception { + @Override + protected ExitStatus doExecute(StepContext stepContext) throws Exception { + StepExecution stepExecution = stepContext.getStepExecution(); stream.update(stepExecution.getExecutionContext()); getJobRepository().updateExecutionContext(stepExecution); - return stepOperations.iterate(new RepeatCallback() { + return stepOperations.iterate(new StepContextRepeatCallback(stepContext ) { final Queue attributeQueue = new LinkedBlockingQueue(); - public ExitStatus doInIteration(RepeatContext context) throws Exception { + public ExitStatus doInStepContext(RepeatContext context, StepContext stepContext) throws Exception { + StepExecution stepExecution = stepContext.getStepExecution(); ExceptionHolder fatalException = new ExceptionHolder(); StepContribution contribution = stepExecution.createStepContribution(); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/Collaborator.java b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/Collaborator.java new file mode 100644 index 000000000..80ae0f0a9 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/Collaborator.java @@ -0,0 +1,7 @@ +package org.springframework.batch.core.scope; + +public interface Collaborator { + + public abstract String getName(); + +} \ No newline at end of file diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepContextRepeatCallbackTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepContextRepeatCallbackTests.java new file mode 100644 index 000000000..f06b64816 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepContextRepeatCallbackTests.java @@ -0,0 +1,45 @@ +/* + * 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.scope; + +import static org.junit.Assert.assertEquals; + +import org.junit.Test; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.batch.repeat.RepeatContext; + +/** + * @author Dave Syer + * + */ +public class StepContextRepeatCallbackTests { + + @Test + public void testDoInIteration() throws Exception { + StepContext stepContext = new StepContext(new StepExecution("foo", new JobExecution(0L), 123L)); + StepContextRepeatCallback callback = new StepContextRepeatCallback(stepContext ) { + @Override + public ExitStatus doInStepContext(RepeatContext context, StepContext stepContext) throws Exception { + assertEquals(Long.valueOf(123), stepContext.getStepExecution().getId()); + return ExitStatus.NOOP; + } + }; + assertEquals(ExitStatus.NOOP, callback.doInIteration(null)); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepContextTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepContextTests.java new file mode 100644 index 000000000..3977e844a --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepContextTests.java @@ -0,0 +1,131 @@ +/* + * 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.scope; + +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 static org.junit.Assert.fail; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.Test; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.StepExecution; + +/** + * @author Dave Syer + * + */ +public class StepContextTests { + + private List list = new ArrayList(); + + private StepExecution stepExecution = new StepExecution("step", new JobExecution(0L)); + + private StepContext context = new StepContext(stepExecution); + + @Test + public void testGetStepExecution() { + context = new StepContext(stepExecution); + assertNotNull(context.getStepExecution()); + } + + @Test + public void testNullStepExecution() { + try { + context = new StepContext(null); + fail("Expected IllegalArgumentException"); + } + catch (IllegalArgumentException e) { + // expected + } + } + + @Test + public void testEqualsSelf() { + assertEquals(context, context); + } + + @Test + public void testNotEqualsNull() { + assertFalse(context.equals(null)); + } + + @Test + public void testEqualsContextWithSameStepExecution() { + assertEquals(new StepContext(stepExecution), context); + } + + @Test + public void testDestructionCallbackSunnyDay() throws Exception { + context.setAttribute("foo", "FOO"); + context.registerDestructionCallback("foo", new Runnable() { + public void run() { + list.add("bar"); + } + }); + context.close(); + assertEquals(1, list.size()); + assertEquals("bar", list.get(0)); + } + + @Test + public void testDestructionCallbackMissingAttribute() throws Exception { + context.registerDestructionCallback("foo", new Runnable() { + public void run() { + list.add("bar"); + } + }); + context.close(); + // Yes the callback should be called even if the attribute is missing - + // for inner beans + assertEquals(1, list.size()); + } + + @Test + public void testDestructionCallbackWithException() throws Exception { + context.setAttribute("foo", "FOO"); + context.setAttribute("bar", "BAR"); + context.registerDestructionCallback("bar", new Runnable() { + public void run() { + list.add("spam"); + throw new RuntimeException("fail!"); + } + }); + context.registerDestructionCallback("foo", new Runnable() { + public void run() { + list.add("bar"); + throw new RuntimeException("fail!"); + } + }); + try { + context.close(); + fail("Expected RuntimeException"); + } + catch (RuntimeException e) { + // We don't care which one was thrown... + assertEquals("fail!", e.getMessage()); + } + // ...but we do care that both were executed: + assertEquals(2, list.size()); + assertTrue(list.contains("bar")); + assertTrue(list.contains("spam")); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepScopeIntegrationTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepScopeIntegrationTests.java new file mode 100644 index 000000000..c7fd5c518 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepScopeIntegrationTests.java @@ -0,0 +1,73 @@ +package org.springframework.batch.core.scope; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + + +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class StepScopeIntegrationTests { + + @Autowired + @Qualifier("vanilla") + private Step vanilla; + + @Autowired + @Qualifier("proxied") + private Step proxied; + + @Autowired + @Qualifier("enhanced") + private Step enhanced; + + @Before + @After + public void start() { + TestStep.reset(); + } + + @Test + public void testScopeCreation() throws Exception { + vanilla.execute(new StepExecution("foo",new JobExecution(11L))); + assertNotNull(TestStep.getContext()); + assertNull(StepSynchronizationManager.getContext()); + } + + @Test + public void testScopedProxy() throws Exception { + proxied.execute(new StepExecution("foo",new JobExecution(11L))); + assertTrue(TestStep.getContext().attributeNames().length>0); + String collaborator = (String) TestStep.getContext().getAttribute("collaborator"); + assertNotNull(collaborator); + assertEquals("bar", collaborator); + } + + @Test + public void testExecutionContext() throws Exception { + StepExecution stepExecution = new StepExecution("foo",new JobExecution(11L)); + ExecutionContext executionContext = new ExecutionContext(); + executionContext.put("name", "spam"); + stepExecution.setExecutionContext(executionContext); + enhanced.execute(stepExecution); + assertTrue(TestStep.getContext().attributeNames().length>0); + String collaborator = (String) TestStep.getContext().getAttribute("collaborator"); + assertNotNull(collaborator); + assertEquals("bar", collaborator); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepScopeTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepScopeTests.java new file mode 100644 index 000000000..4022d9b34 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepScopeTests.java @@ -0,0 +1,188 @@ +/* + * 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.scope; + +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 static org.junit.Assert.fail; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.StepExecution; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.ObjectFactory; +import org.springframework.context.support.StaticApplicationContext; + +/** + * @author Dave Syer + * + */ +public class StepScopeTests { + + private StepScope scope = new StepScope(); + + private StepContext context = new StepContext(new StepExecution("foo", new JobExecution(0L))); + + @Before + public void setUp() throws Exception { + StepSynchronizationManager.register(context); + } + + @After + public void tearDown() throws Exception { + StepSynchronizationManager.close(); + } + + @Test + public void testGetWithNoContext() throws Exception { + final String foo = "bar"; + StepSynchronizationManager.close(); + try { + scope.get("foo", new ObjectFactory() { + public Object getObject() throws BeansException { + return foo; + } + }); + fail("Expected IllegalStateException"); + } + catch (IllegalStateException e) { + // expected + } + + } + + @Test + public void testGetWithNothingAlreadyThere() { + final String foo = "bar"; + Object value = scope.get("foo", new ObjectFactory() { + public Object getObject() throws BeansException { + return foo; + } + }); + assertEquals(foo, value); + assertTrue(context.hasAttribute("foo")); + } + + @Test + public void testGetWithSomethingAlreadyThere() { + context.setAttribute("foo", "bar"); + Object value = scope.get("foo", new ObjectFactory() { + public Object getObject() throws BeansException { + return null; + } + }); + assertEquals("bar", value); + assertTrue(context.hasAttribute("foo")); + } + + @Test + public void testGetWithSomethingAlreadyInParentContext() { + context.setAttribute("foo", "bar"); + StepContext context = new StepContext(new StepExecution("bar", new JobExecution(0L))); + StepSynchronizationManager.register(context); + Object value = scope.get("foo", new ObjectFactory() { + public Object getObject() throws BeansException { + return "spam"; + } + }); + assertEquals("spam", value); + assertTrue(context.hasAttribute("foo")); + StepSynchronizationManager.close(); + assertEquals("bar", scope.get("foo", null)); + } + + @Test + public void testGetConversationId() { + String id = scope.getConversationId(); + assertNotNull(id); + } + + @Test + public void testGetConversationIdFromAttribute() { + context.setAttribute(StepScope.ID_KEY, "foo"); + String id = scope.getConversationId(); + assertEquals("foo", id); + } + + @Test + public void testRegisterDestructionCallback() { + final List list = new ArrayList(); + context.setAttribute("foo", "bar"); + scope.registerDestructionCallback("foo", new Runnable() { + public void run() { + list.add("foo"); + } + }); + assertEquals(0, list.size()); + // When the context is closed, provided the attribute exists the + // callback is called... + context.close(); + assertEquals(1, list.size()); + } + + @Test + public void testRegisterAnotherDestructionCallback() { + final List list = new ArrayList(); + context.setAttribute("foo", "bar"); + scope.registerDestructionCallback("foo", new Runnable() { + public void run() { + list.add("foo"); + } + }); + scope.registerDestructionCallback("foo", new Runnable() { + public void run() { + list.add("bar"); + } + }); + assertEquals(0, list.size()); + // When the context is closed, provided the attribute exists the + // callback is called... + context.close(); + assertEquals(2, list.size()); + } + + @Test + public void testRemove() { + context.setAttribute("foo", "bar"); + scope.remove("foo"); + assertFalse(context.hasAttribute("foo")); + } + + @Test + public void testOrder() throws Exception { + assertEquals(Integer.MAX_VALUE, scope.getOrder()); + scope.setOrder(11); + assertEquals(11, scope.getOrder()); + } + + @Test + public void testName() throws Exception { + scope.setName("foo"); + StaticApplicationContext beanFactory = new StaticApplicationContext(); + scope.postProcessBeanFactory(beanFactory.getDefaultListableBeanFactory()); + String[] scopes = beanFactory.getDefaultListableBeanFactory().getRegisteredScopeNames(); + assertEquals(1, scopes.length); + assertEquals("foo", scopes[0]); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepSynchronizationManagerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepSynchronizationManagerTests.java new file mode 100644 index 000000000..36a34b101 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepSynchronizationManagerTests.java @@ -0,0 +1,87 @@ +package org.springframework.batch.core.scope; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.StepExecution; + +public class StepSynchronizationManagerTests { + + private StepExecution stepExecution = new StepExecution("step", new JobExecution(0L)); + + @Before + @After + public void start() { + while (StepSynchronizationManager.getContext() != null) { + StepSynchronizationManager.close(); + } + } + + @Test + public void testGetContext() { + assertNull(StepSynchronizationManager.getContext()); + StepSynchronizationManager.register(new StepContext(stepExecution)); + assertNotNull(StepSynchronizationManager.getContext()); + } + + @Test + public void testClose() { + StepContext context = new StepContext(stepExecution); + final List list = new ArrayList(); + context.registerDestructionCallback("foo", new Runnable() { + public void run() { + list.add("foo"); + } + }); + StepSynchronizationManager.register(context); + StepSynchronizationManager.close(); + assertNull(StepSynchronizationManager.getContext()); + assertEquals(0, list.size()); + } + + @Test + public void testRelease() { + StepContext context = new StepContext(stepExecution); + final List list = new ArrayList(); + context.registerDestructionCallback("foo", new Runnable() { + public void run() { + list.add("foo"); + } + }); + StepSynchronizationManager.register(context); + // On release we expect the destruction callbacks to be called + StepSynchronizationManager.release(); + assertNull(StepSynchronizationManager.getContext()); + assertEquals(1, list.size()); + } + + @Test + public void testRegisterNull() { + assertNull(StepSynchronizationManager.getContext()); + StepSynchronizationManager.register(null); + assertNull(StepSynchronizationManager.getContext()); + } + + @Test + public void testRegisterTwice() { + StepContext context = new StepContext(stepExecution); + StepSynchronizationManager.register(context); + StepSynchronizationManager.register(context); + StepSynchronizationManager.close(); + // if someone registers you have to assume they are going to close, so + // the last thing you want is for the close to remove another context + // that someone else has registered + assertNotNull(StepSynchronizationManager.getContext()); + StepSynchronizationManager.close(); + assertNull(StepSynchronizationManager.getContext()); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/TestCollaborator.java b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/TestCollaborator.java new file mode 100644 index 000000000..fa4e97b37 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/TestCollaborator.java @@ -0,0 +1,19 @@ +package org.springframework.batch.core.scope; + + +public class TestCollaborator implements Collaborator { + + private String name; + + /* (non-Javadoc) + * @see org.springframework.batch.core.scope.Collaborator#getName() + */ + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/TestStep.java b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/TestStep.java new file mode 100644 index 000000000..efe6daf30 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/TestStep.java @@ -0,0 +1,42 @@ +package org.springframework.batch.core.scope; + +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; + +public class TestStep implements Step { + + private static StepContext context; + + private Collaborator collaborator; + + public void setCollaborator(Collaborator collaborator) { + this.collaborator = collaborator; + } + + public static StepContext getContext() { + return context; + } + + public static void reset() { + context = null; + } + + public void execute(StepExecution stepExecution) throws JobInterruptedException { + context = StepSynchronizationManager.getContext(); + context.setAttribute("collaborator", collaborator.getName()); + } + + public String getName() { + return "foo"; + } + + public int getStartLimit() { + return Integer.MAX_VALUE; + } + + public boolean isAllowStartIfComplete() { + return false; + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/AbstractStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/AbstractStepTests.java index 9e63e8420..5d45acc29 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/AbstractStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/AbstractStepTests.java @@ -16,6 +16,7 @@ import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.StepExecutionListener; +import org.springframework.batch.core.scope.StepContext; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.repeat.ExitStatus; import org.springframework.util.Assert; @@ -54,8 +55,8 @@ public class AbstractStepTests { events.add("open"); } - protected ExitStatus doExecute(StepExecution stepExecution) throws Exception { - assertSame(execution, stepExecution); + protected ExitStatus doExecute(StepContext context) throws Exception { + assertSame(execution, context.getStepExecution()); events.add("doExecute"); return ExitStatus.FINISHED; } @@ -165,8 +166,9 @@ public class AbstractStepTests { @Test public void testFailure() throws Exception { tested = new EventTrackingStep() { - protected ExitStatus doExecute(StepExecution stepExecution) throws Exception { - super.doExecute(stepExecution); + @Override + protected ExitStatus doExecute(StepContext context) throws Exception { + super.doExecute(context); throw new RuntimeException("crash!"); } }; @@ -202,9 +204,10 @@ public class AbstractStepTests { @Test public void testStoppedStep() throws Exception { tested = new EventTrackingStep() { - protected ExitStatus doExecute(StepExecution stepExecution) throws Exception { - stepExecution.setTerminateOnly(); - return super.doExecute(stepExecution); + @Override + protected ExitStatus doExecute(StepContext context) throws Exception { + context.getStepExecution().setTerminateOnly(); + return super.doExecute(context); } }; tested.setJobRepository(repository); @@ -237,8 +240,9 @@ public class AbstractStepTests { @Test public void testFailureInSavingExecutionContext() throws Exception { tested = new EventTrackingStep() { - protected ExitStatus doExecute(StepExecution stepExecution) throws Exception { - super.doExecute(stepExecution); + @Override + protected ExitStatus doExecute(StepContext context) throws Exception { + super.doExecute(context); return ExitStatus.FINISHED; } }; diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/scope/StepScopeIntegrationTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/scope/StepScopeIntegrationTests-context.xml new file mode 100644 index 000000000..802f16568 --- /dev/null +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/scope/StepScopeIntegrationTests-context.xml @@ -0,0 +1,35 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java index 4a3d096f9..94c983c85 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java @@ -19,6 +19,7 @@ import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.UnexpectedJobExecutionException; +import org.springframework.batch.core.scope.StepContext; import org.springframework.batch.core.step.AbstractStep; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.repeat.ExitStatus; @@ -93,15 +94,15 @@ public class MessageOrientedStep extends AbstractStep { this.source = source; } - /* - * (non-Javadoc) - * @see org.springframework.batch.core.step.AbstractStep#execute(org.springframework.batch.core.StepExecution) + /** + * @see AbstractStep#execute(StepExecution) */ @Override - public ExitStatus doExecute(StepExecution stepExecution) throws JobInterruptedException, + public ExitStatus doExecute(StepContext stepContext) throws JobInterruptedException, UnexpectedJobExecutionException { - JobExecutionRequest request = new JobExecutionRequest(stepExecution.getJobExecution()); + StepExecution stepExecution = stepContext.getStepExecution(); + JobExecutionRequest request = new JobExecutionRequest(stepExecution .getJobExecution()); ExecutionContext executionContext = stepExecution.getExecutionContext(); diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/StepExecutionMessageHandler.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/StepExecutionMessageHandler.java index 85eb99c45..09795d624 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/StepExecutionMessageHandler.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/StepExecutionMessageHandler.java @@ -25,12 +25,14 @@ import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.item.ExecutionContext; import org.springframework.beans.factory.annotation.Required; +import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.ServiceActivator; /** * @author Dave Syer * */ +@MessageEndpoint public class StepExecutionMessageHandler { private Step step;