diff --git a/spring-batch-integration/.classpath b/spring-batch-integration/.classpath new file mode 100644 index 000000000..35c37d99e --- /dev/null +++ b/spring-batch-integration/.classpath @@ -0,0 +1,10 @@ + + + + + + + + + + diff --git a/spring-batch-integration/.project b/spring-batch-integration/.project new file mode 100644 index 000000000..135d74a02 --- /dev/null +++ b/spring-batch-integration/.project @@ -0,0 +1,29 @@ + + + spring-integration-batch + + + + + + org.eclipse.jdt.core.javabuilder + + + + + org.devzuz.q.maven.jdt.core.mavenIncrementalBuilder + + + + + org.springframework.ide.eclipse.core.springbuilder + + + + + + org.springframework.ide.eclipse.core.springnature + org.eclipse.jdt.core.javanature + org.devzuz.q.maven.jdt.core.mavenNature + + diff --git a/spring-batch-integration/.settings/org.eclipse.jdt.core.prefs b/spring-batch-integration/.settings/org.eclipse.jdt.core.prefs new file mode 100644 index 000000000..f0eda8040 --- /dev/null +++ b/spring-batch-integration/.settings/org.eclipse.jdt.core.prefs @@ -0,0 +1,12 @@ +#Mon Jun 02 08:35:20 BST 2008 +eclipse.preferences.version=1 +org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled +org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.5 +org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve +org.eclipse.jdt.core.compiler.compliance=1.5 +org.eclipse.jdt.core.compiler.debug.lineNumber=generate +org.eclipse.jdt.core.compiler.debug.localVariable=generate +org.eclipse.jdt.core.compiler.debug.sourceFile=generate +org.eclipse.jdt.core.compiler.problem.assertIdentifier=error +org.eclipse.jdt.core.compiler.problem.enumIdentifier=error +org.eclipse.jdt.core.compiler.source=1.5 diff --git a/spring-batch-integration/.settings/org.springframework.ide.eclipse.core.prefs b/spring-batch-integration/.settings/org.springframework.ide.eclipse.core.prefs new file mode 100644 index 000000000..99192453d --- /dev/null +++ b/spring-batch-integration/.settings/org.springframework.ide.eclipse.core.prefs @@ -0,0 +1,49 @@ +#Mon Jun 02 08:59:31 BST 2008 +eclipse.preferences.version=1 +org.springframework.ide.eclipse.core.builders.enable.aopreferencemodelbuilder=true +org.springframework.ide.eclipse.core.builders.enable.beanmetadatabuilder=false +org.springframework.ide.eclipse.core.builders.enable.osgibundleupdater=true +org.springframework.ide.eclipse.core.enable.project.preferences=false +org.springframework.ide.eclipse.core.validator.enable.com.springsource.sts.bestpractices.beansvalidator=true +org.springframework.ide.eclipse.core.validator.enable.org.springframework.ide.eclipse.beans.core.beansvalidator=true +org.springframework.ide.eclipse.core.validator.enable.org.springframework.ide.eclipse.core.springvalidator=true +org.springframework.ide.eclipse.core.validator.enable.org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.com.springsource.sts.bestpractices.com.springsource.sts.bestpractices.AvoidDriverManagerDataSource-com.springsource.sts.bestpractices.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.com.springsource.sts.bestpractices.com.springsource.sts.bestpractices.ImportElementsAtTopRulee-com.springsource.sts.bestpractices.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.com.springsource.sts.bestpractices.com.springsource.sts.bestpractices.ParentBeanSpecifiesAbstractClassRule-com.springsource.sts.bestpractices.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.com.springsource.sts.bestpractices.com.springsource.sts.bestpractices.RefElementRule-com.springsource.sts.bestpractices.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.com.springsource.sts.bestpractices.com.springsource.sts.bestpractices.TooManyBeansInFileRule-com.springsource.sts.bestpractices.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.com.springsource.sts.bestpractices.com.springsource.sts.bestpractices.UnnecessaryValueElementRule-com.springsource.sts.bestpractices.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.com.springsource.sts.bestpractices.com.springsource.sts.bestpractices.UseBeanInheritance-com.springsource.sts.bestpractices.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.com.springsource.sts.bestpractices.legacyxmlusage.jndiobjectfactory-com.springsource.sts.bestpractices.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.beans.core.beanAlias-org.springframework.ide.eclipse.beans.core.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.beans.core.beanClass-org.springframework.ide.eclipse.beans.core.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.beans.core.beanConstructorArgument-org.springframework.ide.eclipse.beans.core.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.beans.core.beanDefinition-org.springframework.ide.eclipse.beans.core.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.beans.core.beanDefinitionHolder-org.springframework.ide.eclipse.beans.core.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.beans.core.beanFactory-org.springframework.ide.eclipse.beans.core.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.beans.core.beanInitDestroyMethod-org.springframework.ide.eclipse.beans.core.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.beans.core.beanProperty-org.springframework.ide.eclipse.beans.core.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.beans.core.beanReference-org.springframework.ide.eclipse.beans.core.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.beans.core.methodOverride-org.springframework.ide.eclipse.beans.core.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.beans.core.parsingProblems-org.springframework.ide.eclipse.beans.core.beansvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.beans.core.requiredProperty-org.springframework.ide.eclipse.beans.core.beansvalidator=false +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.core.springClasspath-org.springframework.ide.eclipse.core.springvalidator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.action-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.actionstate-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.attribute-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.attributemapper-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.beanaction-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.evaluationaction-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.evaluationresult-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.exceptionhandler-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.import-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.inputattribute-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.mapping-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.outputattribute-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.set-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.state-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.subflowstate-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.transition-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.variable-org.springframework.ide.eclipse.webflow.core.validator=true +org.springframework.ide.eclipse.core.validator.rule.enable.org.springframework.ide.eclipse.webflow.core.validation.webflowstate-org.springframework.ide.eclipse.webflow.core.validator=true diff --git a/spring-batch-integration/.springBeans b/spring-batch-integration/.springBeans new file mode 100644 index 000000000..13e46bd54 --- /dev/null +++ b/spring-batch-integration/.springBeans @@ -0,0 +1,20 @@ + + + 1 + + + + + + + src/test/resources/integration-context.xml + src/test/resources/job-execution-context.xml + src/test/resources/org/springframework/integration/batch/launch/JobLaunchingMessageHandlerIntegrationTests-context.xml + src/test/resources/org/springframework/integration/batch/item/MessageChannelItemWriterIntegrationTests-context.xml + src/test/resources/org/springframework/integration/batch/job/MessageOrientedStepIntegrationTests-context.xml + src/test/resources/org/springframework/integration/batch/file/ResourceSplitterIntegrationTests-context.xml + src/test/resources/simple-job-launcher-context.xml + + + + diff --git a/spring-batch-integration/pom.xml b/spring-batch-integration/pom.xml new file mode 100644 index 000000000..f19873fbf --- /dev/null +++ b/spring-batch-integration/pom.xml @@ -0,0 +1,126 @@ + + + + 4.0.0 + org.springframework.batch + spring-batch-integration + jar + 1.1.0.DEV-SNAPSHOT + + true + 2.5.4.A + 1.0.0.M4 + + + + javax.jms + jms + 1.1 + compile + + + junit + junit + 4.4 + test + + + log4j + log4j + 1.2.12 + test + + + cglib + cglib-nodep + 2.1_3 + test + + + org.springframework.integration + org.springframework.integration + ${spring.integration.version} + compile + + + org.springframework.integration + org.springframework.integration.adapter + ${spring.integration.version} + compile + + + org.springframework + org.springframework.context + ${spring.framework.version} + compile + + + org.springframework + org.springframework.aop + ${spring.framework.version} + compile + + + org.springframework + org.springframework.jms + ${spring.framework.version} + compile + + + org.springframework + org.springframework.transaction + ${spring.framework.version} + compile + + + org.springframework + org.springframework.test + ${spring.framework.version} + test + + + org.springframework.batch + spring-batch-core + 1.1.0.DEV-SNAPSHOT + compile + + + org.springframework + spring-context + + + org.springframework + spring-beans + + + org.springframework + spring-aop + + + + + + + springsource-external + http://repository.springsource.com/maven/bundles/external + + + springsource-release + http://repository.springsource.com/maven/bundles/release + + + springsource-milestone + http://repository.springsource.com/maven/bundles/milestone + + + + springsource-snapshot + http://repository.springsource.com/maven/bundles/snapshot + + + diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/AsynchronousFailureException.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/AsynchronousFailureException.java new file mode 100644 index 000000000..69a00f084 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/AsynchronousFailureException.java @@ -0,0 +1,37 @@ +package org.springframework.integration.batch.chunk; + +import org.springframework.batch.item.ItemWriterException; + +/** + * Exception indicating that a failure or early completion condition was + * detected in a remote worker. + * + * @author Dave Syer + * + */ +public class AsynchronousFailureException extends ItemWriterException { + + /** + * Create a new {@link AsynchronousFailureException} based on a message and + * another exception. + * + * @param message + * the message for this exception + * @param cause + * the other exception + */ + public AsynchronousFailureException(String message, Throwable cause) { + super(message, cause); + } + + /** + * Create a new {@link AsynchronousFailureException} based on a message. + * + * @param message + * the message for this exception + */ + public AsynchronousFailureException(String message) { + super(message); + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ChunkHandler.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ChunkHandler.java new file mode 100644 index 000000000..3f496d10c --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ChunkHandler.java @@ -0,0 +1,8 @@ +package org.springframework.integration.batch.chunk; + + +public interface ChunkHandler { + + ChunkResponse handleChunk(ChunkRequest chunk); + +} \ No newline at end of file diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ChunkMessageChannelItemWriter.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ChunkMessageChannelItemWriter.java new file mode 100644 index 000000000..2d7f7b61c --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ChunkMessageChannelItemWriter.java @@ -0,0 +1,255 @@ +package org.springframework.integration.batch.chunk; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.listener.StepExecutionListenerSupport; +import org.springframework.batch.item.ClearFailedException; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.FlushFailedException; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemStreamException; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.database.HibernateAwareItemWriter; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.batch.repeat.RepeatContext; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.util.Assert; + +public class ChunkMessageChannelItemWriter extends StepExecutionListenerSupport implements ItemWriter, ItemStream { + + private static final Log logger = LogFactory.getLog(ChunkMessageChannelItemWriter.class); + + /** + * Key for items processed in the current transaction {@link RepeatContext}. + */ + private static final String ITEMS_PROCESSED = HibernateAwareItemWriter.class.getName() + ".ITEMS_PROCESSED"; + + static final String ACTUAL = "ACTUAL"; + + static final String EXPECTED = "EXPECTED"; + + private static final long DEFAULT_THROTTLE_LIMIT = 6; + + private MessageChannel requestChannel; + + private MessageChannel replyChannel; + + // TODO: abstract the state or make a factory for this writer? + private LocalState localState = new LocalState(); + + private long throttleLimit = DEFAULT_THROTTLE_LIMIT; + + public void setReplyChannel(MessageChannel replyChannel) { + this.replyChannel = replyChannel; + } + + public void setRequestChannel(MessageChannel requestChannel) { + this.requestChannel = requestChannel; + } + + public void write(Object item) throws Exception { + bindTransactionResources(); + getProcessed().add(item); + logger.debug("Added item to chunk: " + item); + } + + /** + * Flush the buffer. + * + * @see org.springframework.batch.item.ItemWriter#flush() + */ + public void flush() throws FlushFailedException { + + bindTransactionResources(); // in case we are called outside a + // transaction + + // Block until expecting < throttle limit - can Spring + // Integration do that for me? + while (localState.getExpecting() > throttleLimit) { + getNextResult(100); + } + + List processed = getProcessed(); + + if (!processed.isEmpty()) { + + logger.debug("Dispatching chunk: " + processed); + ChunkRequest request = new ChunkRequest(processed, localState.getJobId(), localState.getSkipCount()); + GenericMessage message = new GenericMessage(request ); + requestChannel.send(message); + localState.expected++; + + } + + // Short little timeout to look for an immediate reply. + getNextResult(1); + + unbindTransactionResources(); + + } + + @Override + public void beforeStep(StepExecution stepExecution) { + localState.setStepExecution(stepExecution); + } + + @Override + public ExitStatus afterStep(StepExecution stepExecution) { + if (!(stepExecution.getStatus() == BatchStatus.COMPLETED)) { + return ExitStatus.CONTINUABLE; + } + long expecting = localState.getExpecting(); + boolean timedOut; + try { + logger.debug("Waiting for results in step listener..."); + timedOut = !waitForResults(); + logger.debug("Finished waiting for results in step listener."); + } + catch (RuntimeException e) { + logger.debug("Detected failure waiting for results in step listener."); + stepExecution.setStatus(BatchStatus.FAILED); + return ExitStatus.FAILED.addExitDescription(e.getClass().getName() + ": " + e.getMessage()); + } + if (timedOut) { + stepExecution.setStatus(BatchStatus.FAILED); + throw new ItemStreamException("Timed out waiting for back log at end of step"); + } + return ExitStatus.FINISHED.addExitDescription("Waited for " + expecting + " results."); + } + + public void close(ExecutionContext executionContext) throws ItemStreamException { + localState.reset(); + } + + public void open(ExecutionContext executionContext) throws ItemStreamException { + if (executionContext.containsKey(EXPECTED)) { + localState.expected = executionContext.getLong(EXPECTED); + localState.actual = executionContext.getLong(ACTUAL); + if (!waitForResults()) { + throw new ItemStreamException("Timed out waiting for back log on open"); + } + } + } + + public void update(ExecutionContext executionContext) throws ItemStreamException { + executionContext.putLong(EXPECTED, localState.expected); + executionContext.putLong(ACTUAL, localState.actual); + } + + /** + * Wait until all the results that are in the pipeline come back to the + * reply channel. + * + * @return true if successfully received a result, false if timed out + */ + private boolean waitForResults() { + // TODO: cumulative timeout, or throw an exception? + int count = 0; + while (localState.getExpecting() > 0 && count++ < 10) { + getNextResult(100); + } + return count < 10; + } + + /** + * Get the next result if it is available within the timeout specified, + * otherwise return null. + */ + private void getNextResult(long timeout) { + Message message = replyChannel.receive(timeout); + if (message != null) { + ChunkResponse payload = (ChunkResponse) message.getPayload(); + Long jobInstanceId = payload.getJobId(); + Assert.state(jobInstanceId!=null, "Message did not contain job instance id."); + Assert.state(jobInstanceId.equals(localState.getJobId()), "Message contained wrong job instance id ["+jobInstanceId+"] should have been ["+localState.getJobId()+"]."); + localState.actual++; + ExitStatus result = payload.getExitStatus(); + // TODO: check it can never be ExitStatus.FINISHED? + if (!result.isContinuable()) { + throw new AsynchronousFailureException("Failure or early completion detected in handler: " + result); + } + } + } + + /** + * Accessor for the list of processed items in this transaction. + * + * @return the processed + */ + @SuppressWarnings("unchecked") + private List getProcessed() { + Assert.state(TransactionSynchronizationManager.hasResource(ITEMS_PROCESSED), + "Processed items not bound to transaction."); + List processed = (List) TransactionSynchronizationManager.getResource(ITEMS_PROCESSED); + return processed; + } + + /** + * Set up the {@link RepeatContext} as a transaction resource. + * + * @param context the context to set + */ + private void bindTransactionResources() { + if (TransactionSynchronizationManager.hasResource(ITEMS_PROCESSED)) { + return; + } + TransactionSynchronizationManager.bindResource(ITEMS_PROCESSED, new ArrayList()); + } + + /** + * Remove the transaction resource associated with this context. + */ + private void unbindTransactionResources() { + if (!TransactionSynchronizationManager.hasResource(ITEMS_PROCESSED)) { + return; + } + TransactionSynchronizationManager.unbindResource(ITEMS_PROCESSED); + } + + /** + * Clear the buffer. + * + * @see org.springframework.batch.item.ItemWriter#clear() + */ + public void clear() throws ClearFailedException { + unbindTransactionResources(); + } + + private static class LocalState { + private long actual; + + private long expected; + + private StepExecution stepExecution; + + public long getExpecting() { + return expected - actual; + } + + public int getSkipCount() { + // TODO Auto-generated method stub + return stepExecution.getSkipCount(); + } + + public Long getJobId() { + return stepExecution.getJobExecution().getJobId(); + } + + public void setStepExecution(StepExecution stepExecution) { + this.stepExecution = stepExecution; + } + + public void reset() { + expected = actual = 0; + } + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ChunkRequest.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ChunkRequest.java new file mode 100644 index 000000000..bdb9b406a --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ChunkRequest.java @@ -0,0 +1,30 @@ +package org.springframework.integration.batch.chunk; + +import java.io.Serializable; +import java.util.Collection; + +public class ChunkRequest implements Serializable { + + private final int skipCount; + private final Long jobId; + private final Collection items; + + public ChunkRequest(Collection items, Long jobId, int skipCount) { + this.items = items; + this.jobId = jobId; + this.skipCount = skipCount; + } + + public int getSkipCount() { + return skipCount; + } + + public Long getJobId() { + return jobId; + } + + public Collection getItems() { + return items; + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ChunkResponse.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ChunkResponse.java new file mode 100644 index 000000000..6ee3fea02 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ChunkResponse.java @@ -0,0 +1,31 @@ +package org.springframework.integration.batch.chunk; + +import java.io.Serializable; + +import org.springframework.batch.repeat.ExitStatus; + +public class ChunkResponse implements Serializable { + + private final int skipCount; + private final Long jobId; + private final ExitStatus exitStatus; + + public ChunkResponse(ExitStatus exitStatus, Long jobId, int skipCount) { + this.exitStatus = exitStatus; + this.jobId = jobId; + this.skipCount = skipCount; + } + + public int getSkipCount() { + return skipCount; + } + + public Long getJobId() { + return jobId; + } + + public ExitStatus getExitStatus() { + return exitStatus; + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ItemWriterChunkHandler.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ItemWriterChunkHandler.java new file mode 100644 index 000000000..d0e21c629 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ItemWriterChunkHandler.java @@ -0,0 +1,85 @@ +package org.springframework.integration.batch.chunk; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.SkipListener; +import org.springframework.batch.core.listener.CompositeSkipListener; +import org.springframework.batch.core.step.skip.ItemSkipPolicy; +import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.integration.annotation.Handler; +import org.springframework.transaction.annotation.Transactional; + +public class ItemWriterChunkHandler implements ChunkHandler { + + private static final Log logger = LogFactory.getLog(ItemWriterChunkHandler.class); + + private ItemWriter itemWriter; + + private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy(); + + private CompositeSkipListener skipListener = new CompositeSkipListener(); + + public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) { + this.itemSkipPolicy = itemSkipPolicy; + } + + public void setItemWriter(ItemWriter itemWriter) { + this.itemWriter = itemWriter; + } + + public void registerSkipListener(SkipListener listener) { + skipListener.register(listener); + } + + public void setSkipListeners(SkipListener[] skipListeners) { + for (SkipListener listener : skipListeners) { + registerSkipListener(listener); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.integration.batch.slave.ChunkHandler#handleChunk(java.util.Collection) + */ + @Handler + @Transactional + public ChunkResponse handleChunk(ChunkRequest chunk) { + + logger.debug("Handling chunk: " + chunk); + + int parentSkipCount = chunk.getSkipCount(); + int skipCount = 0; + + try { + for (Object item : chunk.getItems()) { + try { + itemWriter.write(item); + } + catch (Exception e) { + if (itemSkipPolicy.shouldSkip(e, parentSkipCount + skipCount)) { + logger.debug("Skipping item on exception", e); + skipCount++; + skipListener.onSkipInWrite(item, e); + } else { + logger.debug("Cannot skip, re-throwing"); + throw e; + } + } + } + itemWriter.flush(); + } + catch (Exception e) { + logger.debug("Failed chunk", e); + itemWriter.clear(); + // TODO: need to force rollback as well + return new ChunkResponse(ExitStatus.FAILED.addExitDescription(e.getClass().getName() + ": " + + e.getMessage()), chunk.getJobId(), skipCount); + } + + logger.debug("Completed chunk handling with " + skipCount + " skips"); + return new ChunkResponse(ExitStatus.CONTINUABLE, chunk.getJobId(), skipCount); + + } +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/file/FileToMessagesJobFactoryBean.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/file/FileToMessagesJobFactoryBean.java new file mode 100644 index 000000000..824f5da53 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/file/FileToMessagesJobFactoryBean.java @@ -0,0 +1,185 @@ +/* + * 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.integration.batch.file; + +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecutionListener; +import org.springframework.batch.core.job.SimpleJob; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.resource.StepExecutionResourceProxy; +import org.springframework.batch.core.step.item.SimpleStepFactoryBean; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.file.FlatFileItemReader; +import org.springframework.batch.item.xml.StaxEventItemReader; +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.FactoryBean; +import org.springframework.beans.factory.annotation.Required; +import org.springframework.core.io.Resource; +import org.springframework.integration.batch.item.MessageChannelItemWriter; +import org.springframework.integration.batch.launch.JobLaunchingMessageHandler; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.dispatcher.DirectChannel; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.util.Assert; + +/** + * A FactoryBean for a {@link Job} with a single step which just pumps messages + * from a file into a channel. The channel has to be a + * {@link SynchronousChannel} to ensure that failures propagate up to the step + * and fail the job execution. Normally this job will be used in conjunction + * with a {@link JobLaunchingMessageHandler} and a + * {@link ResourcePayloadAsJobParameterStrategy}, so that the user can just + * send a message to a request channel listing the files to be processed, and + * everything else just happens by magic. After a failure the job will be + * restarted just by sending it the same message. + * + * @author Dave Syer + * + */ +public class FileToMessagesJobFactoryBean implements FactoryBean, BeanNameAware { + + private String name = "fileToMessageJob"; + + private ItemReader itemReader; + + private MessageChannel channel; + + private PlatformTransactionManager transactionManager; + + private JobRepository jobRepository; + + /* + * (non-Javadoc) + * @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String) + */ + public void setBeanName(String name) { + this.name = name; + } + + /** + * Public setter for the {@link ItemReader}. Must be either a + * {@link FlatFileItemReader} or a {@link StaxEventItemReader}. In either + * case there is no need to set the resource property as it will be set by + * this factory. + * + * @param itemReader the itemReader to set + */ + @Required + public void setItemReader(ItemReader itemReader) { + this.itemReader = itemReader; + } + + /** + * Public setter for the channel. Each item from the item reader will be + * sent to this channel. + * @param channel the channel to set + */ + @Required + public void setChannel(MessageChannel channel) { + this.channel = channel; + } + + /** + * Public setter for the {@link JobRepository}. + * @param jobRepository the job repository to set + */ + @Required + public void setJobRepository(JobRepository jobRepository) { + this.jobRepository = jobRepository; + } + + /** + * Public setter for the {@link PlatformTransactionManager}. + * @param transactionManager the transaction manager to set + */ + @Required + public void setTransactionManager(PlatformTransactionManager transactionManager) { + this.transactionManager = transactionManager; + } + + /** + * Creates a {@link Job} that can process a flat file or XML file into + * messages. To launch the job will require only a {@link JobParameters} + * instance with a resource location as a URL. + * + * @see org.springframework.beans.factory.FactoryBean#getObject() + */ + public Object getObject() throws Exception { + + SimpleJob job = new SimpleJob(); + job.setName(name); + job.setJobRepository(jobRepository); + + SimpleStepFactoryBean stepFactory = new SimpleStepFactoryBean(); + stepFactory.setBeanName("step"); + + Assert.state((itemReader instanceof FlatFileItemReader) || (itemReader instanceof StaxEventItemReader), + "ItemReader must be either a FlatFileItemReader or a StaxEventItemReader"); + StepExecutionResourceProxy resourceProxy = new StepExecutionResourceProxy(); + resourceProxy.setFilePattern("%" + ResourcePayloadAsJobParameterStrategy.FILE_INPUT_PATH + "%"); + stepFactory.setListeners(new StepExecutionListener[] { resourceProxy }); + setResource(itemReader, resourceProxy); + stepFactory.setItemReader(itemReader); + + Assert.notNull(channel, "A channel must be provided"); + Assert.state(channel instanceof DirectChannel, + "The channel must be a DirectChannel (otherwise failures can not be recovered from)"); + MessageChannelItemWriter itemWriter = new MessageChannelItemWriter(); + itemWriter.setChannel(channel); + stepFactory.setItemWriter(itemWriter); + + Assert.notNull(transactionManager, "A transaction manager must be provided"); + stepFactory.setTransactionManager(transactionManager); + + Assert.notNull(jobRepository, "A job repository must be provided"); + stepFactory.setJobRepository(jobRepository); + + job.addStep((Step) stepFactory.getObject()); + return job; + } + + /** + * @param itemReader + * @param resource + */ + private void setResource(ItemReader itemReader, Resource resource) { + if (itemReader instanceof FlatFileItemReader) { + ((FlatFileItemReader) itemReader).setResource(resource); + } + else { + ((StaxEventItemReader) itemReader).setResource(resource); + } + } + + /** + * Always returns {@link Job}. + * @see org.springframework.beans.factory.FactoryBean#getObjectType() + */ + public Class getObjectType() { + return Job.class; + } + + /** + * Always true. TODO: should it be false? + * @see org.springframework.beans.factory.FactoryBean#isSingleton() + */ + public boolean isSingleton() { + return true; + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/file/ResourcePayloadAsJobParameterStrategy.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/file/ResourcePayloadAsJobParameterStrategy.java new file mode 100644 index 000000000..5780eddf8 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/file/ResourcePayloadAsJobParameterStrategy.java @@ -0,0 +1,56 @@ +/* + * 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.integration.batch.file; + +import java.io.IOException; + +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.JobParametersBuilder; +import org.springframework.batch.item.ItemStreamException; +import org.springframework.core.io.Resource; +import org.springframework.integration.batch.launch.MessageToJobParametersStrategy; +import org.springframework.integration.message.Message; + +/** + * @author Dave Syer + * + */ +public class ResourcePayloadAsJobParameterStrategy implements MessageToJobParametersStrategy { + + /** + * The key name for the job parameter that will be a URL for the input file + */ + public static final String FILE_INPUT_PATH = "input.file.path"; + + /** + * Convert a message payload which is a {@link Resource} to its URL + * representation and load that into a job parameter. + * + * @see org.springframework.integration.batch.launch.MessageToJobParametersStrategy#getJobParameters(org.springframework.integration.message.Message) + */ + public JobParameters getJobParameters(Message message) { + JobParametersBuilder builder = new JobParametersBuilder(); + Resource resource = (Resource) message.getPayload(); + try { + builder.addString(FILE_INPUT_PATH, resource.getURL().toExternalForm()); + } + catch (IOException e) { + throw new ItemStreamException("Could not create URL for resource: [" + resource + "]", e); + } + return builder.toJobParameters(); + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/item/MessageChannelItemWriter.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/item/MessageChannelItemWriter.java new file mode 100644 index 000000000..d58168eea --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/item/MessageChannelItemWriter.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.integration.batch.item; + +import org.springframework.batch.item.AbstractItemWriter; +import org.springframework.beans.factory.annotation.Required; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.GenericMessage; + +/** + * @author Dave Syer + * + */ +public class MessageChannelItemWriter extends AbstractItemWriter { + + private MessageChannel channel; + + /** + * Public setter for the channel. + * @param channel the channel to set + */ + @Required + public void setChannel(MessageChannel channel) { + this.channel = channel; + } + + /* + * (non-Javadoc) + * @see org.springframework.batch.item.ItemWriter#write(java.lang.Object) + */ + public void write(Object item) throws Exception { + channel.send(new GenericMessage(item)); + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/job/JobExecutionReply.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/job/JobExecutionReply.java new file mode 100644 index 000000000..e731fb883 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/job/JobExecutionReply.java @@ -0,0 +1,24 @@ +/* + * 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.integration.batch.job; + +/** + * @author dsyer + * + */ +public class JobExecutionReply { + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/job/JobExecutionRequest.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/job/JobExecutionRequest.java new file mode 100644 index 000000000..b1889be73 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/job/JobExecutionRequest.java @@ -0,0 +1,107 @@ +/* + * 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.integration.batch.job; + +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.core.AttributeAccessorSupport; + +/** + * Encapsulation of a request to execute a job execution through a message flow + * consisting of step handlers. A handler should pass the message on as it is, + * modifying the request properties as necessary. Generally a handler will + * execute a step as part of the {@link JobExecution} passed in, and should + * change the status to {@link BatchStatus#COMPLETED} if the step is successful + * (generally a handler cannot determine if the whole job execution is complete, + * so this is just information about the step).
+ * + * If the incoming status is {@link BatchStatus#FAILED}, + * {@link BatchStatus#STOPPED} or {@link BatchStatus#STOPPING} the request + * should be ignored by handlers (passed on without modification). + * + * @author Dave Syer + * + */ +public class JobExecutionRequest extends AttributeAccessorSupport { + + private JobExecution jobExecution; + + private BatchStatus status; + + private Throwable throwable; + + /** + * @param jobExecution + */ + public JobExecutionRequest(JobExecution jobExecution) { + this.jobExecution = jobExecution; + status = jobExecution.getStatus(); + } + + /** + * @return + */ + public Long getJobId() { + return this.jobExecution.getJobId(); + } + + /** + * @return + */ + public BatchStatus getStatus() { + return status; + } + + /** + * Public setter for the status. + * @param status the status to set + */ + public void setStatus(BatchStatus status) { + this.status = status; + } + + /** + * @return + */ + public boolean hasErrors() { + return throwable != null; + } + + /** + * Public getter for the throwable. + * @return the throwable + */ + public Throwable getLastThrowable() { + return throwable; + } + + /** + * Public setter for the throwable. + * @param throwable the throwable to set + */ + public void registerThrowable(Throwable throwable) { + this.throwable = throwable; + } + + /** + * Public getter for the jobExecution. + * @return the jobExecution + */ + public JobExecution getJobExecution() { + return jobExecution; + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/job/MessageOrientedStep.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/job/MessageOrientedStep.java new file mode 100644 index 000000000..6747e40d3 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/job/MessageOrientedStep.java @@ -0,0 +1,165 @@ +/* + * 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.integration.batch.job; + +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.step.AbstractStep; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.beans.factory.annotation.Required; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.util.Assert; + +/** + * @author Dave Syer + * + */ +public class MessageOrientedStep extends AbstractStep { + + /** + * Key in execution context for flag to say we are waiting. + */ + public static final String WAITING = MessageOrientedStep.class.getName() + ".WAITING"; + + private MessageChannel requestChannel; + + private MessageChannel replyChannel; + + /** + * Public setter for the requestChannel. + * @param requestChannel the requestChannel to set + */ + @Required + public void setRequestChannel(MessageChannel requestChannel) { + this.requestChannel = requestChannel; + } + + /** + * Public setter for the replyChannel. + * @param replyChannel the replyChannel to set + */ + @Required + public void setReplyChannel(MessageChannel replyChannel) { + this.replyChannel = replyChannel; + } + + /* + * (non-Javadoc) + * @see org.springframework.batch.core.step.AbstractStep#execute(org.springframework.batch.core.StepExecution) + */ + @Override + public ExitStatus doExecute(StepExecution stepExecution) throws JobInterruptedException, + UnexpectedJobExecutionException { + + JobExecutionRequest request = new JobExecutionRequest(stepExecution.getJobExecution()); + + ExecutionContext executionContext = stepExecution.getExecutionContext(); + + if (executionContext.containsKey(WAITING)) { + // restart scenario: we are still waiting for a response + waitForReply(request.getJobId()); + } + else { + executionContext.putString(WAITING, "true"); + // TODO: need these two lines to be atomic + getJobRepository().saveOrUpdate(stepExecution); + requestChannel.send(new GenericMessage(request)); + waitForReply(request.getJobId()); + } + + return ExitStatus.FINISHED; + + } + + /** + * Do nothing. + * + * @see org.springframework.batch.core.step.AbstractStep#open(org.springframework.batch.item.ExecutionContext) + */ + @Override + protected void open(ExecutionContext ctx) throws Exception { + } + + /** + * Do nothing. + * + * @see org.springframework.batch.core.step.AbstractStep#close(org.springframework.batch.item.ExecutionContext) + */ + @Override + protected void close(ExecutionContext ctx) throws Exception { + } + + /** + * @param expectedJobId + */ + private void waitForReply(Long expectedJobId) { + // TODO: promote timeout to field and calculate count + long timeout = 5; + int count = 0; + + // TODO: use a ReponseCorrelator?, or just a SynchronousChannel + while (count++ < 100) { + + Message message = replyChannel.receive(timeout); + + if (message != null) { + + JobExecutionRequest payload = (JobExecutionRequest) message.getPayload(); + Long jobInstanceId = payload.getJobId(); + Assert.state(jobInstanceId != null, "Message did not contain job instance id."); + Assert.state(jobInstanceId.equals(expectedJobId), "Message contained wrong job instance id [" + + jobInstanceId + "] should have been [" + expectedJobId + "]."); + + if (payload.getStatus() == BatchStatus.COMPLETED) { + // One of the steps decided we were finished + // TODO: wait for all the other steps that might be + // executing concurrently? + // TODO: maybe *any* reply on this channel should + // mean the end of the step? + break; + } + + if (payload.hasErrors()) { + rethrow(payload.getLastThrowable()); + } + + } + } + + if (count >= 100) { + throw new StepExecutionTimeoutException("Timed out waiting for steps to execute."); + } + } + + /** + * @param lastThrowable + */ + private static void rethrow(Throwable t) throws RuntimeException { + if (t instanceof RuntimeException) { + throw (RuntimeException) t; + } + if (t instanceof Exception) { + throw new UnexpectedJobExecutionException("Unexpected checked exception thrown by step.", t); + } + throw (Error) t; + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/job/StepExecutionMessageHandler.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/job/StepExecutionMessageHandler.java new file mode 100644 index 000000000..a6c17381b --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/job/StepExecutionMessageHandler.java @@ -0,0 +1,238 @@ +/* + * 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.integration.batch.job; + +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobExecutionException; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.StartLimitExceededException; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.beans.factory.annotation.Required; +import org.springframework.integration.annotation.Handler; + +/** + * @author Dave Syer + * + */ +public class StepExecutionMessageHandler { + + private Step step; + + private JobRepository jobRepository; + + private String[] inputKeys = new String[0]; + + private String[] outputKeys = new String[0]; + + /** + * Public setter for the {@link Step}. + * @param step the step to set + */ + @Required + public void setStep(Step step) { + this.step = step; + } + + /** + * Public setter for the input keys. Attributes from the incoming request + * will be added to the execution context for the step. + * @param inputKeys the inputKeys to set + */ + public void setInputKeys(String[] inputKeys) { + this.inputKeys = inputKeys; + } + + /** + * Public setter for the output keys. Attributes from a successful step + * execution context will be added to the request before it is handed on to + * the next handler. + * @param outputKeys the outputKeys to set + */ + public void setOutputKeys(String[] outputKeys) { + this.outputKeys = outputKeys; + } + + /** + * Public setter for the {@link JobRepository} that is needed to manage the + * state of the batch meta domain (jobs, steps, executions) during the life + * of a job. + * + * @param jobRepository + */ + @Required + public void setJobRepository(JobRepository jobRepository) { + this.jobRepository = jobRepository; + } + + @Handler + public JobExecutionRequest handle(JobExecutionRequest request) { + + + // Hand off immediately if the job has already failed + if (isComplete(request)) { + return request; + } + + JobExecution jobExecution = request.getJobExecution(); + JobInstance jobInstance = jobExecution.getJobInstance(); + + StepExecution stepExecution = jobExecution.createStepExecution(step); + try { + + StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step); + + // Even if it completed successfully we want to pass on the output + // attributes, so set up the execution context here if it is + // available. + if (lastStepExecution != null) { + stepExecution.setExecutionContext(lastStepExecution.getExecutionContext()); + } + + // If it is already complete and not restartable it will simply be + // skipped + if (shouldStart(lastStepExecution, step)) { + + boolean isRestart = (jobRepository.getStepExecutionCount(jobInstance, step) > 0 && !lastStepExecution + .getExitStatus().equals(ExitStatus.FINISHED)) ? true : false; + + if (!isRestart || lastStepExecution == null) { + stepExecution.setExecutionContext(getExecutionContextWithInputs(request)); + } + + step.execute(stepExecution); + + } + + // (the job might actually not be complete, but the stage is). + request.setStatus(BatchStatus.COMPLETED); + + } + catch (Exception e) { + handleFailure(request, e); + } + catch (Error e) { + handleFailure(request, e); + throw e; + } + + // TODO: could a failure here could cause job to be in inconsistent + // state? + getMessageWithOutputs(request, stepExecution.getExecutionContext()); + return request; + + } + + /** + * Clear the message header of the input attributes and add in output + * attributes from the execution context. + * @param request + * @param executionContext + * @return + */ + private JobExecutionRequest getMessageWithOutputs(JobExecutionRequest request, ExecutionContext executionContext) { + for (int i = 0; i < inputKeys.length; i++) { + String key = inputKeys[i]; + request.removeAttribute(key); + } + for (int i = 0; i < outputKeys.length; i++) { + String key = outputKeys[i]; + request.setAttribute(key, executionContext.get(key)); + } + // TODO: is this safe, or should we build a new message? + return request; + } + + /** + * Generate a new execution context with all the attributes requested from + * the message (if they exist). + * + * @return an {@link ExecutionContext} + */ + private ExecutionContext getExecutionContextWithInputs(JobExecutionRequest request) { + ExecutionContext executionContext = new ExecutionContext(); + for (int i = 0; i < inputKeys.length; i++) { + String key = inputKeys[i]; + Object value = request.getAttribute(key); + executionContext.put(key, value); + } + return executionContext; + } + + /** + * @param request + * @return + */ + private boolean isComplete(JobExecutionRequest request) { + return request.getStatus() == BatchStatus.FAILED || request.getStatus() == BatchStatus.STOPPED + || request.getStatus() == BatchStatus.STOPPING; + } + + /** + * @param request + * @param e + */ + private void handleFailure(JobExecutionRequest request, Throwable e) { + request.registerThrowable(e); + request.setStatus(BatchStatus.FAILED); + } + + /* + * TODO: merge this with SimpleJob implementation. + * + * Given a step and configuration, return true if the step should start, + * false if it should not, and throw an exception if the job should finish. + */ + private boolean shouldStart(StepExecution lastStepExecution, Step step) throws JobExecutionException { + + BatchStatus stepStatus; + // if the last execution is null, the step has never been executed. + if (lastStepExecution == null) { + return true; + } + else { + stepStatus = lastStepExecution.getStatus(); + } + + if (stepStatus == BatchStatus.UNKNOWN) { + throw new JobExecutionException("Cannot restart step from UNKNOWN status. " + + "The last execution ended with a failure that could not be rolled back, " + + "so it may be dangerous to proceed. " + "Manual intervention is probably necessary."); + } + + if (stepStatus == BatchStatus.COMPLETED && step.isAllowStartIfComplete() == false) { + // step is complete, false should be returned, indicating that the + // step should not be started + return false; + } + + if (jobRepository.getStepExecutionCount(lastStepExecution.getJobExecution().getJobInstance(), step) < step + .getStartLimit()) { + // step start count is less than start max, return true + return true; + } + else { + // start max has been exceeded, throw an exception. + throw new StartLimitExceededException("Maximum start limit exceeded for step: " + step.getName() + + "StartMax: " + step.getStartLimit()); + } + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/job/StepExecutionTimeoutException.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/job/StepExecutionTimeoutException.java new file mode 100644 index 000000000..a98b83ab2 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/job/StepExecutionTimeoutException.java @@ -0,0 +1,36 @@ +/* + * 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.integration.batch.job; + +import org.springframework.batch.core.UnexpectedJobExecutionException; + +/** + * @author Dave Syer + * + */ +public class StepExecutionTimeoutException extends UnexpectedJobExecutionException { + + /** + * Constructs a new instance. + * + * @param msg the exception message. + * + */ + public StepExecutionTimeoutException(String msg) { + super(msg); + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/JobLaunchingMessageHandler.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/JobLaunchingMessageHandler.java new file mode 100644 index 000000000..85dcfdf68 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/JobLaunchingMessageHandler.java @@ -0,0 +1,55 @@ +package org.springframework.integration.batch.launch; + +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobExecutionException; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.launch.JobLauncher; +import org.springframework.integration.annotation.Handler; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageHandlingException; + +/** + * Message handler which uses strategies to convert a Message into a job and a + * set of job parameters + * @author Jonas Partner + * @author Dave Syer + * + */ +public class JobLaunchingMessageHandler { + + private MessageToJobStrategy messageToJobStrategy; + + private MessageToJobParametersStrategy messageToJobParametersStrategy = new MessagePropertiesToJobParametersStrategy(); + + private JobLauncher jobLauncher; + + public JobLaunchingMessageHandler(JobLauncher jobLauncher, MessageToJobStrategy messageToJobStrategy) { + super(); + this.jobLauncher = jobLauncher; + this.messageToJobStrategy = messageToJobStrategy; + } + + @Handler + public JobExecution handle(Message message) { + Job job = messageToJobStrategy.getJob(message); + JobParameters jobParameters = messageToJobParametersStrategy.getJobParameters(message); + + try { + JobExecution execution = jobLauncher.run(job, jobParameters); + if (message.getHeader().getReturnAddress() != null) { + return execution; + } + return null; + } + catch (JobExecutionException e) { + throw new MessageHandlingException(message, "Exception executing job "); + } + } + + public void setMessageToJobParametersStrategy(MessageToJobParametersStrategy messageToJobParametersStrategy) { + this.messageToJobParametersStrategy = messageToJobParametersStrategy; + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/JobLaunchingPostReceiveChannelInterceptor.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/JobLaunchingPostReceiveChannelInterceptor.java new file mode 100644 index 000000000..5a6981eec --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/JobLaunchingPostReceiveChannelInterceptor.java @@ -0,0 +1,58 @@ +package org.springframework.integration.batch.launch; + +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobExecutionException; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.launch.JobLauncher; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageHandlingException; + +/** + * Channel interceptor which launches the configured job after the message has + * been received + * @author Jonas Partner + * + */ +public class JobLaunchingPostReceiveChannelInterceptor extends ChannelInterceptorAdapter { + + private JobLauncher jobLauncher; + + private Job job; + + private MessageToJobParametersStrategy messageToJobParametersStrategy = new MessagePropertiesToJobParametersStrategy(); + + /** + * + * @param job The job to launch + * @param jobLauncher + * @param errorHandler + */ + public JobLaunchingPostReceiveChannelInterceptor(Job job, JobLauncher jobLauncher) { + super(); + this.job = job; + this.jobLauncher = jobLauncher; + } + + public MessageToJobParametersStrategy getMessageToJobParametersStrategy() { + return messageToJobParametersStrategy; + } + + public void setMessageToJobParametersStrategy(MessageToJobParametersStrategy messageToJobParametersStrategy) { + this.messageToJobParametersStrategy = messageToJobParametersStrategy; + } + + @Override + public void postReceive(Message message, MessageChannel channel) { + JobParameters parameters = messageToJobParametersStrategy.getJobParameters(message); + try { + jobLauncher.run(job, parameters); + } + catch (JobExecutionException e) { + throw new MessageHandlingException(message, "Excpetion executing job ", e); + } + + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/MessagePropertiesToJobParametersStrategy.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/MessagePropertiesToJobParametersStrategy.java new file mode 100644 index 000000000..fd0720a40 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/MessagePropertiesToJobParametersStrategy.java @@ -0,0 +1,25 @@ +package org.springframework.integration.batch.launch; + +import java.util.Set; + +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.JobParametersBuilder; +import org.springframework.integration.message.Message; + +/** + * Builds an instance of JobParameters from the properties in the message header + * @author Jonas Partner + * + */ +public class MessagePropertiesToJobParametersStrategy implements MessageToJobParametersStrategy { + + public JobParameters getJobParameters(Message message) { + JobParametersBuilder parametersBuilder = new JobParametersBuilder(); + Set propertyNames = message.getHeader().getPropertyNames(); + for (String key : propertyNames) { + parametersBuilder.addString(key, message.getHeader().getProperty(key)); + } + return parametersBuilder.toJobParameters(); + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/MessageToJobParametersStrategy.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/MessageToJobParametersStrategy.java new file mode 100644 index 000000000..d092cd253 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/MessageToJobParametersStrategy.java @@ -0,0 +1,16 @@ +package org.springframework.integration.batch.launch; + +import org.springframework.batch.core.JobParameters; +import org.springframework.integration.message.Message; + + +/** + * + * @author Jonas Partner + * + */ +public interface MessageToJobParametersStrategy { + + public JobParameters getJobParameters(Message message); + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/MessageToJobStrategy.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/MessageToJobStrategy.java new file mode 100644 index 000000000..d1d26d5db --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/MessageToJobStrategy.java @@ -0,0 +1,16 @@ +package org.springframework.integration.batch.launch; + +import org.springframework.batch.core.Job; +import org.springframework.integration.message.Message; + + +/** + * Interface for strategy implementations which convert from a Message to a Spring batch Job + * @author Jonas Partner + * + */ +public interface MessageToJobStrategy{ + + public Job getJob(Message message); + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/StringPayloadAsJobNameStrategy.java b/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/StringPayloadAsJobNameStrategy.java new file mode 100644 index 000000000..58a255f82 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/StringPayloadAsJobNameStrategy.java @@ -0,0 +1,32 @@ +package org.springframework.integration.batch.launch; + +import org.springframework.batch.core.Job; +import org.springframework.batch.core.configuration.JobLocator; +import org.springframework.batch.core.repository.NoSuchJobException; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageHandlingException; + +/** + * Takes the string payload of a message and delegates to a JobLocator + * @author Jonas Partner + * + */ +public class StringPayloadAsJobNameStrategy implements MessageToJobStrategy{ + + private JobLocator jobLocator; + + public StringPayloadAsJobNameStrategy(JobLocator jobLocator){ + this.jobLocator = jobLocator; + } + + public Job getJob(Message message) { + String name = (String)message.getPayload(); + try { + return jobLocator.getJob(name); + } + catch (NoSuchJobException e) { + throw new MessageHandlingException(message, "Could not find job with name " + name, e); + } + } + +} diff --git a/spring-batch-integration/src/site/apt/index.apt b/spring-batch-integration/src/site/apt/index.apt new file mode 100644 index 000000000..f0c51b2e2 --- /dev/null +++ b/spring-batch-integration/src/site/apt/index.apt @@ -0,0 +1,10 @@ + ------ + Spring Integration Batch + ------ + Dave Syer + ------ + March 2008 + +Overview of the Spring Integration Batch Module + + Many of the 2.0 Features we identified for Spring Batch look like they might be efficiently and concisely implemented in Spring Integration. These are features that can extend Spring Batch, or use Spring batch features in the context of Spring Integration. diff --git a/spring-batch-integration/src/site/site.xml b/spring-batch-integration/src/site/site.xml new file mode 100644 index 000000000..4aac57281 --- /dev/null +++ b/spring-batch-integration/src/site/site.xml @@ -0,0 +1,17 @@ + + + + + + + + + + + + + + + + + diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/JobRepositorySupport.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/JobRepositorySupport.java new file mode 100644 index 000000000..33113d810 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/JobRepositorySupport.java @@ -0,0 +1,75 @@ +/* + * 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.integration.batch; + +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; +import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.JobRestartException; + +/** + * @author Dave Syer + * + */ +public class JobRepositorySupport implements JobRepository { + + /* (non-Javadoc) + * @see org.springframework.batch.core.repository.JobRepository#createJobExecution(org.springframework.batch.core.Job, org.springframework.batch.core.JobParameters) + */ + public JobExecution createJobExecution(Job job, JobParameters jobParameters) + throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException { + return new JobExecution(new JobInstance(0L, jobParameters, job.getName())); + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.repository.JobRepository#getLastStepExecution(org.springframework.batch.core.JobInstance, org.springframework.batch.core.Step) + */ + public StepExecution getLastStepExecution(JobInstance jobInstance, Step step) { + return null; + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.repository.JobRepository#getStepExecutionCount(org.springframework.batch.core.JobInstance, org.springframework.batch.core.Step) + */ + public int getStepExecutionCount(JobInstance jobInstance, Step step) { + return 0; + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.repository.JobRepository#saveOrUpdate(org.springframework.batch.core.JobExecution) + */ + public void saveOrUpdate(JobExecution jobExecution) { + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.repository.JobRepository#saveOrUpdate(org.springframework.batch.core.StepExecution) + */ + public void saveOrUpdate(StepExecution stepExecution) { + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.repository.JobRepository#saveOrUpdateExecutionContext(org.springframework.batch.core.StepExecution) + */ + public void saveOrUpdateExecutionContext(StepExecution stepExecution) { + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/JobSupport.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/JobSupport.java new file mode 100644 index 000000000..813193423 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/JobSupport.java @@ -0,0 +1,36 @@ +package org.springframework.integration.batch; + +import java.util.List; + +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobExecutionException; + +public class JobSupport implements Job { + + String name; + + public JobSupport(String name){ + this.name = name; + } + + public void execute(JobExecution execution) throws JobExecutionException { + // TODO Auto-generated method stub + } + + public String getName() { + return name; + } + + @SuppressWarnings("unchecked") + public List getSteps() { + // TODO Auto-generated method stub + return null; + } + + public boolean isRestartable() { + // TODO Auto-generated method stub + return false; + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/SmokeTests.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/SmokeTests.java new file mode 100644 index 000000000..722baf0f4 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/SmokeTests.java @@ -0,0 +1,57 @@ +package org.springframework.integration.batch; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.annotation.Handler; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +@ContextConfiguration(locations = "/integration-context.xml") +@RunWith(SpringJUnit4ClassRunner.class) +@MessageEndpoint(input = "smokein", output = "smokeout") +public class SmokeTests { + + @Autowired + @Qualifier("smokein") + private MessageChannel smokein; + + @Autowired + @Qualifier("smokeout") + private MessageChannel smokeout; + + // This has to be static because the MessageBus registers the handler + // more than once (every time a test instance is created), but only one of + // them will get the message. + private volatile static int count = 0; + + @Handler + public String process(String message) { + count++; + String result = message + ": " + count; + return result; + } + + @Test + public void testDummyWithSimpleAssert() throws Exception { + assertTrue(true); + } + + @Test + public void testVanillaSendAndReceive() throws Exception { + smokein.send(new GenericMessage("foo")); + Message message = smokeout.receive(100); + String result = (String) (message == null ? null : message.getPayload()); + assertEquals("foo: 1", result); + assertEquals(1, count); + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/StepSupport.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/StepSupport.java new file mode 100644 index 000000000..bac3efac8 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/StepSupport.java @@ -0,0 +1,74 @@ +/* + * 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.integration.batch; + +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; + +/** + * @author Dave Syer + * + */ +public class StepSupport implements Step { + + private String name; + private int startLimit; + + /** + * @param name + */ + public StepSupport(String name) { + super(); + this.name = name; + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.Step#execute(org.springframework.batch.core.StepExecution) + */ + public void execute(StepExecution stepExecution) throws JobInterruptedException { + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.Step#getName() + */ + public String getName() { + return name; + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.Step#getStartLimit() + */ + public int getStartLimit() { + return startLimit; + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.Step#isAllowStartIfComplete() + */ + public boolean isAllowStartIfComplete() { + return false; + } + + /** + * Public setter for the startLimit. + * @param startLimit the startLimit to set + */ + public void setStartLimit(int startLimit) { + this.startLimit = startLimit; + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/chunk/ChunkMessageItemWriterIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/chunk/ChunkMessageItemWriterIntegrationTests.java new file mode 100644 index 000000000..38d5d92d7 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/chunk/ChunkMessageItemWriterIntegrationTests.java @@ -0,0 +1,330 @@ +package org.springframework.integration.batch.chunk; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.Arrays; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParametersBuilder; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.UnexpectedJobExecutionException; +import org.springframework.batch.core.job.SimpleJob; +import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; +import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; +import org.springframework.batch.core.repository.JobRestartException; +import org.springframework.batch.core.repository.dao.MapJobExecutionDao; +import org.springframework.batch.core.repository.dao.MapJobInstanceDao; +import org.springframework.batch.core.repository.dao.MapStepExecutionDao; +import org.springframework.batch.core.repository.support.SimpleJobRepository; +import org.springframework.batch.core.step.item.SimpleStepFactoryBean; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.support.ListItemReader; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.batch.support.transaction.ResourcelessTransactionManager; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.StringUtils; + +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class ChunkMessageItemWriterIntegrationTests { + + private ChunkMessageChannelItemWriter writer = new ChunkMessageChannelItemWriter(); + + @Autowired + @Qualifier("requests") + private MessageChannel requests; + + @Autowired + @Qualifier("replies") + private MessageChannel replies; + + private SimpleStepFactoryBean factory; + + private SimpleJobRepository jobRepository; + + private static long jobCounter; + + @Before + public void setUp() { + + factory = new SimpleStepFactoryBean(); + jobRepository = new SimpleJobRepository(new MapJobInstanceDao(), + new MapJobExecutionDao(), new MapStepExecutionDao()); + factory.setJobRepository(jobRepository); + factory.setTransactionManager(new ResourcelessTransactionManager()); + factory.setBeanName("step"); + factory.setItemWriter(writer); + factory.setCommitInterval(4); + + writer.setReplyChannel(replies); + writer.setRequestChannel(requests); + + TestItemWriter.count = 0; + + // Drain queues + Message message = requests.receive(10); + while (message!=null) { + System.err.println(message); + message = requests.receive(10); + } + message = replies.receive(10); + while (message!=null) { + System.err.println(message); + message = replies.receive(10); + } + + } + + @After + public void tearDown() { + while(requests.receive(10L)!=null) {} + while(replies.receive(10L)!=null) {} + } + + @Test + public void testOpenWithNoState() throws Exception { + writer.open(new ExecutionContext()); + } + + @Test + public void testUpdateAndOpenWithState() throws Exception { + ExecutionContext executionContext = new ExecutionContext(); + writer.update(executionContext); + writer.open(executionContext); + assertEquals(0, executionContext + .getLong(ChunkMessageChannelItemWriter.EXPECTED)); + assertEquals(0, executionContext + .getLong(ChunkMessageChannelItemWriter.ACTUAL)); + } + + @Test + public void testVanillaIteration() throws Exception { + + factory.setItemReader(new ListItemReader(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("1,2,3,4,5,6")))); + + Step step = (Step) factory.getObject(); + + StepExecution stepExecution = getStepExecution(step); + step.execute(stepExecution); + + waitForResults(6, 10); + + assertEquals(6, TestItemWriter.count); + assertEquals(6, stepExecution.getItemCount().intValue()); + + } + + @Test + public void testSimulatedRestart() throws Exception { + + factory.setItemReader(new ListItemReader(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("1,2,3,4,5,6")))); + + Step step = (Step) factory.getObject(); + + StepExecution stepExecution = getStepExecution(step); + + // Set up context with two messages (chunks) in the backlog + stepExecution.getExecutionContext().putLong( + ChunkMessageChannelItemWriter.EXPECTED, 6); + stepExecution.getExecutionContext().putLong( + ChunkMessageChannelItemWriter.ACTUAL, 4); + // And make the back log real + requests.send(getSimpleMessage("foo", stepExecution.getJobExecution().getJobId())); + requests.send(getSimpleMessage("bar", stepExecution.getJobExecution().getJobId())); + step.execute(stepExecution); + + waitForResults(8, 10); + + assertEquals(8, TestItemWriter.count); + assertEquals(6, stepExecution.getItemCount().intValue()); + + } + + @Test + public void testSimulatedRestartWithBadMessagesFromAnotherJob() throws Exception { + + factory.setItemReader(new ListItemReader(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("1,2,3,4,5,6")))); + + Step step = (Step) factory.getObject(); + + StepExecution stepExecution = getStepExecution(step); + + // Set up context with two messages (chunks) in the backlog + stepExecution.getExecutionContext().putLong( + ChunkMessageChannelItemWriter.EXPECTED, 3); + stepExecution.getExecutionContext().putLong( + ChunkMessageChannelItemWriter.ACTUAL, 2); + // And make the back log real + requests.send(getSimpleMessage("foo", new Long(4321))); + try { + step.execute(stepExecution); + fail("Expected UnexpectedJobExecutionException"); + } catch (UnexpectedJobExecutionException e) { + String message = e.getCause().getMessage(); + assertTrue("Message does not contain 'wrong job': "+message, message.contains("wrong job")); + } + + waitForResults(1, 10); + + assertEquals(1, TestItemWriter.count); + assertEquals(0, stepExecution.getItemCount().intValue()); + + } + + /** + * @param jobId + * @param string + * @return + */ + @SuppressWarnings("unchecked") + private GenericMessage getSimpleMessage(String string, Long jobId) { + ChunkRequest chunk = new ChunkRequest(StringUtils + .commaDelimitedListToSet(string), jobId, 0); + GenericMessage message = new GenericMessage(chunk); + return message; + } + + @Test + public void testEarlyCompletionSignalledInHandler() throws Exception { + + factory.setItemReader(new ListItemReader(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("1,bad,3,4,5,6")))); + factory.setCommitInterval(2); + + Step step = (Step) factory.getObject(); + + StepExecution stepExecution = getStepExecution(step); + try { + step.execute(stepExecution); + fail("Expected AsynchronousFailureException"); + } catch (AsynchronousFailureException e) { + assertTrue(e.getMessage().contains("bad")); + } + + waitForResults(2, 10); + + // The number of items processed is actually between 1 and 6, because + // the one that failed might have been processed out of order. + assertTrue(1 <= TestItemWriter.count); + assertTrue(6 >= TestItemWriter.count); + // But it should fail the step in any case + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + + } + + @Test + public void testSimulatedRestartWithNoBacklog() throws Exception { + + factory.setItemReader(new ListItemReader(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("1,2,3,4,5,6")))); + + Step step = (Step) factory.getObject(); + + StepExecution stepExecution = getStepExecution(step); + + // Set up expectation of three messages (chunks) in the backlog + stepExecution.getExecutionContext().putLong( + ChunkMessageChannelItemWriter.EXPECTED, 6); + stepExecution.getExecutionContext().putLong( + ChunkMessageChannelItemWriter.ACTUAL, 3); + /* + * With no backlog we process all the items, but the listener can't + * reconcile the expected number of items with the actual. An infinite + * loop would be bad, so the best we can do is fail as fast as possible. + */ + try { + step.execute(stepExecution); + fail("Expected UnexpectedJobExecutionException"); + } catch (UnexpectedJobExecutionException e) { + String message = e.getCause().getMessage(); + assertTrue("Message did not contain 'timed out': " + message, + message.toLowerCase().contains("timed out")); + } + + assertEquals(0, TestItemWriter.count); + assertEquals(0, stepExecution.getItemCount().intValue()); + + } + + /** + * This one is flakey - we try to force it to wait until after the step to + * finish processing just by waiting for long enough. + * + * @throws Exception + */ + @Test + public void testFailureInStepListener() throws Exception { + + factory.setItemReader(new ListItemReader(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("wait,bad,3,4,5,6")))); + + Step step = (Step) factory.getObject(); + + StepExecution stepExecution = getStepExecution(step); + step.execute(stepExecution); + + waitForResults(2, 10); + + // The number of items processed is actually between 1 and 6, because + // the one that failed might have been processed out of order. + assertTrue(1 <= TestItemWriter.count); + assertTrue(6 >= TestItemWriter.count); + + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + assertEquals(ExitStatus.FAILED.getExitCode(), stepExecution + .getExitStatus().getExitCode()); + + String exitDescription = stepExecution.getExitStatus() + .getExitDescription(); + assertTrue("Exit description does not contain exception type name: " + + exitDescription, exitDescription + .contains(AsynchronousFailureException.class.getName())); + + } + + // TODO: test failure in chunk handler + // TODO : test non-dispatch of empty chunk + + /** + * @param expected + * @param maxWait + * @throws InterruptedException + */ + private void waitForResults(int expected, int maxWait) throws InterruptedException { + int count = 0; + while (TestItemWriter.count[] { ItemReader.class }); + assertNotNull(method); + Annotation[] annotations = AnnotationUtils.getAnnotations(method); + assertEquals(1, annotations.length); + assertEquals(Required.class, annotations[0].annotationType()); + } + + /** + * Test method for + * {@link org.springframework.integration.batch.file.FileToMessagesJobFactoryBean#setChannel(org.springframework.integration.channel.MessageChannel)}. + */ + @Test + public void testSetChannel() { + Method method = ReflectionUtils.findMethod(FileToMessagesJobFactoryBean.class, "setChannel", + new Class[] { MessageChannel.class }); + assertNotNull(method); + Annotation[] annotations = AnnotationUtils.getAnnotations(method); + assertEquals(1, annotations.length); + assertEquals(Required.class, annotations[0].annotationType()); + } + + /** + * Test method for + * {@link org.springframework.integration.batch.file.FileToMessagesJobFactoryBean#setJobRepository(org.springframework.batch.core.repository.JobRepository)}. + */ + @Test + public void testSetJobRepository() { + Method method = ReflectionUtils.findMethod(FileToMessagesJobFactoryBean.class, "setJobRepository", + new Class[] { JobRepository.class }); + assertNotNull(method); + Annotation[] annotations = AnnotationUtils.getAnnotations(method); + assertEquals(1, annotations.length); + assertEquals(Required.class, annotations[0].annotationType()); + } + + /** + * Test method for + * {@link org.springframework.integration.batch.file.FileToMessagesJobFactoryBean#setTransactionManager(org.springframework.transaction.PlatformTransactionManager)}. + */ + @Test + public void testSetTransactionManager() { + Method method = ReflectionUtils.findMethod(FileToMessagesJobFactoryBean.class, "setTransactionManager", + new Class[] { PlatformTransactionManager.class }); + assertNotNull(method); + Annotation[] annotations = AnnotationUtils.getAnnotations(method); + assertEquals(1, annotations.length); + assertEquals(Required.class, annotations[0].annotationType()); + } + + /** + * Test method for + * {@link org.springframework.integration.batch.file.FileToMessagesJobFactoryBean#getObject()}. + * @throws Exception + */ + @Test + public void testGetObjectNotBroken() throws Exception { + assertNotNull(factory.getObject()); + } + + /** + * Test method for + * {@link org.springframework.integration.batch.file.FileToMessagesJobFactoryBean#getObjectType()}. + */ + @Test + public void testGetObjectType() { + FileToMessagesJobFactoryBean factory = new FileToMessagesJobFactoryBean(); + assertEquals(Job.class, factory.getObjectType()); + } + + /** + * Test method for + * {@link org.springframework.integration.batch.file.FileToMessagesJobFactoryBean#isSingleton()}. + */ + @Test + public void testIsSingleton() { + FileToMessagesJobFactoryBean factory = new FileToMessagesJobFactoryBean(); + assertEquals(true, factory.isSingleton()); + } + + @SuppressWarnings("unchecked") + @Test + public void testVanillaJobExecution() throws Exception { + + Job job = (Job) factory.getObject(); + JobParameters jobParameters = new JobParametersBuilder().addString(FILE_INPUT_PATH, "classpath:/log4j.properties").toJobParameters(); + JobExecution jobExecution = jobRepository.createJobExecution(job, jobParameters); + + job.execute(jobExecution); + assertNotNull(jobExecution); + assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); + + FieldSet payload; + Message
message; + + // first line from properties file + message = (Message
) receiver.receive(100L); + assertNotNull(message); + payload = message.getPayload(); + assertNotNull(payload); + // second line from properties file + message = (Message
) receiver.receive(100L); + assertNotNull(message); + payload = message.getPayload(); + assertNotNull(payload); + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/file/ResourcePayloadAsJobParameterStrategyTests.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/file/ResourcePayloadAsJobParameterStrategyTests.java new file mode 100644 index 000000000..81467334a --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/file/ResourcePayloadAsJobParameterStrategyTests.java @@ -0,0 +1,64 @@ +/* + * 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.integration.batch.file; + +import static org.junit.Assert.*; + +import org.junit.Test; +import org.springframework.batch.core.JobParameters; +import org.springframework.core.io.ClassPathResource; +import org.springframework.core.io.Resource; +import org.springframework.integration.batch.file.ResourcePayloadAsJobParameterStrategy; +import org.springframework.integration.message.GenericMessage; + +/** + * @author Dave Syer + * + */ +public class ResourcePayloadAsJobParameterStrategyTests { + + /** + * + */ + private static final String INPUT_FILE_PATH = ResourcePayloadAsJobParameterStrategy.FILE_INPUT_PATH; + + /** + * Test method for {@link org.springframework.integration.batch.file.ResourcePayloadAsJobParameterStrategy#getJobParameters(org.springframework.integration.message.Message)}. + */ + @Test + public void testGetJobParameters() { + ResourcePayloadAsJobParameterStrategy strategy = new ResourcePayloadAsJobParameterStrategy(); + JobParameters parameters = strategy.getJobParameters(new GenericMessage(new ClassPathResource("log4j.properties"))); + assertTrue(parameters.getParameters().containsKey(INPUT_FILE_PATH)); + } + + /** + * Test method for {@link org.springframework.integration.batch.file.ResourcePayloadAsJobParameterStrategy#getJobParameters(org.springframework.integration.message.Message)}. + */ + @Test + public void testGetJobParametersWithWrongPayload() { + ResourcePayloadAsJobParameterStrategy strategy = new ResourcePayloadAsJobParameterStrategy(); + try { + strategy.getJobParameters(new GenericMessage("log4j.properties")); + fail("Expected ClassCastException"); + } catch (ClassCastException e) { + String message = e.getMessage(); + assertTrue("Wrong message: "+message, message.contains("String cannot be cast")); + } + + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/file/ResourceSplitterIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/file/ResourceSplitterIntegrationTests.java new file mode 100644 index 000000000..f30223446 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/file/ResourceSplitterIntegrationTests.java @@ -0,0 +1,76 @@ +/* + * 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.integration.batch.file; + +import static org.junit.Assert.assertNotNull; + +import java.util.Arrays; +import java.util.List; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.core.io.Resource; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.Splitter; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Dave Syer + * + */ +@ContextConfiguration() +@RunWith(SpringJUnit4ClassRunner.class) +@MessageEndpoint(input = "resources", output = "requests") +public class ResourceSplitterIntegrationTests { + + @Autowired + @Qualifier("resources") + private MessageChannel resources; + + @Autowired + @Qualifier("requests") + private MessageChannel requests; + + /* + * This is so cool (but see INT-190)...
+ * + * The incoming message is a Resource pattern, and it is converted to the + * correct payload type with Spring's default strategy + */ + @Splitter + public Resource[] handle(Resource[] message) { + List list = Arrays.asList(message); + System.err.println(list); + return message; + } + + @SuppressWarnings("unchecked") + @Test + public void testVanillaConversion() throws Exception { + resources.send(new GenericMessage("classpath:*-context.xml")); + Message message = (Message) requests.receive(100L); + assertNotNull(message); + message = (Message) requests.receive(100L); + assertNotNull(message); + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/item/MessageChannelItemWriterIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/item/MessageChannelItemWriterIntegrationTests.java new file mode 100644 index 000000000..4a2b409be --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/item/MessageChannelItemWriterIntegrationTests.java @@ -0,0 +1,51 @@ +/* + * 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.integration.batch.item; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.item.ItemWriter; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.Message; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Dave Syer + * + */ +@ContextConfiguration() +@RunWith(SpringJUnit4ClassRunner.class) +public class MessageChannelItemWriterIntegrationTests { + + @Autowired + private MessageChannel channel; + + @Autowired + private ItemWriter itemWriter; + + @Test + public void testSend() throws Exception { + itemWriter.write("foo"); + Message message = channel.receive(10); + assertNotNull(message); + assertEquals("foo", message.getPayload()); + } +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/item/MessageChannelItemWriterTests.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/item/MessageChannelItemWriterTests.java new file mode 100644 index 000000000..e5731273d --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/item/MessageChannelItemWriterTests.java @@ -0,0 +1,126 @@ +/* + * 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.integration.batch.item; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.fail; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; + +import org.junit.Test; +import org.springframework.beans.factory.annotation.Required; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.ThreadLocalChannel; +import org.springframework.integration.dispatcher.DirectChannel; +import org.springframework.integration.endpoint.HandlerEndpoint; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.Target; +import org.springframework.integration.util.ErrorHandler; +import org.springframework.util.ReflectionUtils; + +/** + * @author Dave Syer + * + */ +public class MessageChannelItemWriterTests { + + /** + * Test method for {@link org.springframework.integration.batch.item.MessageChannelItemWriter#setChannel(org.springframework.integration.channel.MessageChannel)}. + */ + @Test + public void testSetChannel() { + Method method = ReflectionUtils.findMethod(MessageChannelItemWriter.class, "setChannel", new Class[] {MessageChannel.class}); + assertNotNull(method); + Annotation[] annotations = AnnotationUtils.getAnnotations(method); + assertEquals(1, annotations.length); + assertEquals(Required.class, annotations[0].annotationType()); + } + + /** + * Test method for {@link org.springframework.integration.batch.item.MessageChannelItemWriter#write(java.lang.Object)}. + * @throws Exception + */ + @Test + public void testWrite() throws Exception { + DirectChannel channel = new DirectChannel(); + ThreadLocalChannel receiver = new ThreadLocalChannel(); + channel.subscribe(receiver); + MessageChannelItemWriter writer = new MessageChannelItemWriter(); + writer.setChannel(channel); + writer.write("foo"); + Message message = receiver.receive(10); + assertNotNull(message); + assertEquals("foo", message.getPayload()); + } + + /** + * Test method for {@link org.springframework.integration.batch.item.MessageChannelItemWriter#write(java.lang.Object)}. + * @throws Exception + */ + @Test + public void testWriteWithRollback() throws Exception { + DirectChannel channel = new DirectChannel(); + channel.subscribe(new Target() { + public boolean send(Message message) { + throw new RuntimeException("Planned failure"); + } + }); + MessageChannelItemWriter writer = new MessageChannelItemWriter(); + writer.setChannel(channel); + try { + writer.write("foo"); + fail("Expected RuntimeException"); + } + catch (RuntimeException e) { + assertEquals("Planned failure", e.getMessage()); + } + } + + /** + * Test method for {@link org.springframework.integration.batch.item.MessageChannelItemWriter#write(java.lang.Object)}. + * @throws Exception + */ + @Test + public void testWriteWithRollbackOnEndpoint() throws Exception { + DirectChannel channel = new DirectChannel(); + HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() { + public Message handle(Message message) { + throw new RuntimeException("Planned failure"); + } + }); + // INT-184: this shouldn't be necessary? + endpoint.setErrorHandler(new ErrorHandler() { + public void handle(Throwable t) { + throw (RuntimeException)t; + } + }); + channel.subscribe(endpoint); + endpoint.start(); + MessageChannelItemWriter writer = new MessageChannelItemWriter(); + writer.setChannel(channel); + try { + writer.write("foo"); + fail("Expected RuntimeException"); + } + catch (RuntimeException e) { + assertEquals("Planned failure", e.getMessage()); + } + } +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/job/MessageOrientedStepIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/job/MessageOrientedStepIntegrationTests.java new file mode 100644 index 000000000..8efbd40fa --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/job/MessageOrientedStepIntegrationTests.java @@ -0,0 +1,51 @@ +/* + * 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.integration.batch.job; + +import static org.junit.Assert.assertEquals; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.configuration.JobLocator; +import org.springframework.batch.core.launch.JobLauncher; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Dave Syer + * + */ +@ContextConfiguration() +@RunWith(SpringJUnit4ClassRunner.class) +public class MessageOrientedStepIntegrationTests { + + @Autowired + private JobLauncher jobLauncher; + + @Autowired + private JobLocator jobLocator; + + @Test + public void testLaunchJob() throws Exception { + JobExecution jobExecution = jobLauncher.run(jobLocator.getJob("job"), new JobParameters()); + assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/job/MessageOrientedStepTests.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/job/MessageOrientedStepTests.java new file mode 100644 index 000000000..f6a430fe8 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/job/MessageOrientedStepTests.java @@ -0,0 +1,165 @@ +/* + * 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.integration.batch.job; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.StepExecution; +import org.springframework.beans.factory.annotation.Required; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.batch.JobRepositorySupport; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.ThreadLocalChannel; +import org.springframework.integration.dispatcher.DirectChannel; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.Target; +import org.springframework.util.ReflectionUtils; + +/** + * @author Dave Syer + * + */ +public class MessageOrientedStepTests { + + private MessageOrientedStep step = new MessageOrientedStep(); + + private JobExecution jobExecution; + + private DirectChannel requestChannel; + + private MessageChannel replyChannel; + + /** + * @return + */ + @Before + public void createStep() { + replyChannel = new ThreadLocalChannel(); + requestChannel = new DirectChannel(); + step.setName("step"); + step.setRequestChannel(requestChannel); + step.setReplyChannel(replyChannel); + step.setStartLimit(10); + step.setJobRepository(new JobRepositorySupport()); + JobInstance jobInstance = new JobInstance(0L, new JobParameters(), "job"); + jobExecution = new JobExecution(jobInstance); + } + + /** + * Test method for + * {@link org.springframework.integration.batch.job.MessageOrientedStep#setRequestChannel(org.springframework.integration.channel.MessageChannel)}. + */ + @Test + public void testSetRequestChannel() { + Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setRequestChannel", + new Class[] { MessageChannel.class }); + assertNotNull(method); + Annotation[] annotations = AnnotationUtils.getAnnotations(method); + assertEquals(1, annotations.length); + assertEquals(Required.class, annotations[0].annotationType()); + } + + /** + * Test method for + * {@link org.springframework.integration.batch.job.MessageOrientedStep#setReplyChannel(org.springframework.integration.channel.MessageChannel)}. + */ + @Test + public void testSetReplyChannel() { + Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setReplyChannel", + new Class[] { MessageChannel.class }); + assertNotNull(method); + Annotation[] annotations = AnnotationUtils.getAnnotations(method); + assertEquals(1, annotations.length); + assertEquals(Required.class, annotations[0].annotationType()); + } + + /** + * Test method for + * {@link org.springframework.integration.batch.job.MessageOrientedStep#execute(org.springframework.batch.core.StepExecution)}. + * @throws Exception + * @throws + */ + @Test + public void testExecuteWithTimeout() throws Exception { + try { + step.execute(jobExecution.createStepExecution(step)); + fail("Expected StepExecutionTimeoutException"); + } + catch (StepExecutionTimeoutException e) { + // expected + String message = e.getMessage(); + assertTrue("Wrong message: " + message, message.contains("waiting for steps")); + } + } + + @Test + public void testVanillaExecute() throws Exception { + requestChannel.subscribe(new Target() { + public boolean send(Message message) { + JobExecutionRequest jobExecution = (JobExecutionRequest) message.getPayload(); + jobExecution.setStatus(BatchStatus.COMPLETED); + return replyChannel.send(message); + } + }); + step.execute(jobExecution.createStepExecution(step)); + } + + @Test + public void testExecuteWithFailure() throws Exception { + requestChannel.subscribe(new Target() { + public boolean send(Message message) { + JobExecutionRequest jobExecution = (JobExecutionRequest) message.getPayload(); + jobExecution.registerThrowable(new RuntimeException("Planned failure")); + return replyChannel.send(message); + } + }); + try { + step.execute(jobExecution.createStepExecution(step)); + fail("Expected RuntimeException"); + } + catch (RuntimeException e) { + // expected + String message = e.getMessage(); + assertEquals("Wrong message: " + message, "Planned failure", message); + } + } + + @Test + public void testExecuteOnRestart() throws Exception { + JobExecutionRequest jobExecutionRequest = new JobExecutionRequest(jobExecution); + jobExecutionRequest.setStatus(BatchStatus.COMPLETED); + // Send a message to the reply channel to simulate step that we were + // waiting for when we failed on the last execution. + replyChannel.send(new GenericMessage(jobExecutionRequest)); + StepExecution stepExecution = jobExecution.createStepExecution(step); + stepExecution.getExecutionContext().putString(MessageOrientedStep.WAITING, "true"); + step.execute(stepExecution); + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/job/StepExecutionMessageHandlerTests.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/job/StepExecutionMessageHandlerTests.java new file mode 100644 index 000000000..c276a7163 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/job/StepExecutionMessageHandlerTests.java @@ -0,0 +1,247 @@ +/* + * 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.integration.batch.job; + +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 java.lang.annotation.Annotation; +import java.lang.reflect.Method; + +import org.junit.Test; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.StartLimitExceededException; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.JobRestartException; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.support.PropertiesConverter; +import org.springframework.beans.factory.annotation.Required; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.batch.JobRepositorySupport; +import org.springframework.integration.batch.JobSupport; +import org.springframework.integration.batch.StepSupport; +import org.springframework.util.ReflectionUtils; + +/** + * @author Dave Syer + * + */ +public class StepExecutionMessageHandlerTests { + + /** + * Test method for + * {@link org.springframework.integration.batch.job.StepExecutionMessageHandler#setStep(org.springframework.batch.core.Step)}. + */ + @Test + public void testSetStep() { + Method method = ReflectionUtils.findMethod(StepExecutionMessageHandler.class, "setStep", + new Class[] { Step.class }); + assertNotNull(method); + Annotation[] annotations = AnnotationUtils.getAnnotations(method); + assertEquals(1, annotations.length); + assertEquals(Required.class, annotations[0].annotationType()); + } + + /** + * Test method for + * {@link org.springframework.integration.batch.job.StepExecutionMessageHandler#setJobRepository(org.springframework.batch.core.repository.JobRepository)}. + */ + @Test + public void testSetJobRepository() { + Method method = ReflectionUtils.findMethod(StepExecutionMessageHandler.class, "setJobRepository", + new Class[] { JobRepository.class }); + assertNotNull(method); + Annotation[] annotations = AnnotationUtils.getAnnotations(method); + assertEquals(1, annotations.length); + assertEquals(Required.class, annotations[0].annotationType()); + } + + /** + * Test method for + * {@link org.springframework.integration.batch.job.StepExecutionMessageHandler#handle(org.springframework.integration.message.Message)}. + * @throws Exception + * @throws JobRestartException + * @throws JobExecutionAlreadyRunningException + */ + @SuppressWarnings("unchecked") + @Test + public void testVanillaHandle() throws Exception { + JobRepositorySupport jobRepository = new JobRepositorySupport(); + StepExecutionMessageHandler handler = createHandler(jobRepository); + JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobRepository.createJobExecution( + new JobSupport("job"), new JobParameters()))); + assertEquals(1, message.getJobExecution().getStepExecutions().size()); + assertEquals(BatchStatus.COMPLETED, message.getStatus()); + } + + @SuppressWarnings("unchecked") + @Test + public void testHandleWithInputs() throws Exception { + JobRepositorySupport jobRepository = new JobRepositorySupport(); + StepExecutionMessageHandler handler = createHandler(jobRepository); + handler.setInputKeys(new String[] { "foo" }); + JobExecutionRequest jobExecutionRequest = new JobExecutionRequest(jobRepository.createJobExecution( + new JobSupport("job"), new JobParameters())); + jobExecutionRequest.setAttribute("foo", "bar"); + JobExecutionRequest message = handler.handle(jobExecutionRequest); + assertEquals(1, message.getJobExecution().getStepExecutions().size()); + StepExecution stepExecution = (StepExecution) message.getJobExecution().getStepExecutions().iterator().next(); + assertTrue(stepExecution.getExecutionContext().containsKey("foo")); + } + + @SuppressWarnings("unchecked") + @Test + public void testHandleWithInputsAndOutputs() throws Exception { + JobRepositorySupport jobRepository = new JobRepositorySupport(); + StepExecutionMessageHandler handler = createHandler(jobRepository); + handler.setInputKeys(new String[] { "foo" }); + handler.setOutputKeys(new String[] { "bar" }); + JobExecutionRequest jobExecutionRequest = new JobExecutionRequest(jobRepository.createJobExecution( + new JobSupport("job"), new JobParameters())); + jobExecutionRequest.setAttribute("foo", "bar"); + // The step has to add the output attribute to the context + handler.setStep(new StepSupport("step") { + @Override + public void execute(StepExecution stepExecution) throws JobInterruptedException { + stepExecution.getExecutionContext().putString("bar", "spam"); + } + }); + handler.handle(jobExecutionRequest); + assertFalse(jobExecutionRequest.hasAttribute("foo")); + assertTrue(jobExecutionRequest.hasAttribute("bar")); + } + + @SuppressWarnings("unchecked") + @Test + public void testHandleFailedJob() throws Exception { + JobRepositorySupport jobRepository = new JobRepositorySupport(); + StepExecutionMessageHandler handler = createHandler(jobRepository); + JobExecution jobExecution = jobRepository.createJobExecution(new JobSupport("job"), new JobParameters()); + jobExecution.setStatus(BatchStatus.FAILED); + JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobExecution)); + assertEquals(0, message.getJobExecution().getStepExecutions().size()); + } + + @SuppressWarnings("unchecked") + @Test + public void testHandleRestart() throws Exception { + JobRepositorySupport jobRepository = new JobRepositorySupport() { + @Override + public StepExecution getLastStepExecution(JobInstance jobInstance, Step step) { + StepExecution stepExecution = new StepExecution(step.getName(), new JobExecution(jobInstance)); + stepExecution.setStatus(BatchStatus.FAILED); + stepExecution.setExecutionContext(new ExecutionContext(PropertiesConverter + .stringToProperties("foo=bar"))); + return stepExecution; + } + + /* + * (non-Javadoc) + * @see org.springframework.integration.batch.JobRepositorySupport#getStepExecutionCount(org.springframework.batch.core.JobInstance, + * org.springframework.batch.core.Step) + */ + @Override + public int getStepExecutionCount(JobInstance jobInstance, Step step) { + return 1; + } + }; + StepExecutionMessageHandler handler = createHandler(jobRepository); + JobExecution jobExecution = jobRepository.createJobExecution(new JobSupport("job"), new JobParameters()); + JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobExecution)); + assertNotNull(message); + assertEquals(1, jobExecution.getStepExecutions().size()); + StepExecution stepExecution = (StepExecution) jobExecution.getStepExecutions().iterator().next(); + assertTrue(stepExecution.getExecutionContext().containsKey("foo")); + } + + @SuppressWarnings("unchecked") + @Test + public void testHandleRestartAlreadyComplete() throws Exception { + JobRepositorySupport jobRepository = new JobRepositorySupport() { + @Override + public StepExecution getLastStepExecution(JobInstance jobInstance, Step step) { + StepExecution stepExecution = new StepExecution(step.getName(), new JobExecution(jobInstance)); + stepExecution.setStatus(BatchStatus.COMPLETED); + stepExecution.setExecutionContext(new ExecutionContext(PropertiesConverter + .stringToProperties("foo=bar"))); + return stepExecution; + } + }; + StepExecutionMessageHandler handler = createHandler(jobRepository); + JobExecution jobExecution = jobRepository.createJobExecution(new JobSupport("job"), new JobParameters()); + JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobExecution)); + assertNotNull(message); + assertEquals(1, jobExecution.getStepExecutions().size()); + StepExecution stepExecution = (StepExecution) jobExecution.getStepExecutions().iterator().next(); + assertEquals(BatchStatus.STARTING, stepExecution.getStatus()); + // We expect to get the context from the previous execution, even if we + // do not execute + assertTrue(stepExecution.getExecutionContext().containsKey("foo")); + } + + @SuppressWarnings("unchecked") + @Test + public void testHandleRestartStartLimitExceeded() throws Exception { + JobRepositorySupport jobRepository = new JobRepositorySupport() { + @Override + public StepExecution getLastStepExecution(JobInstance jobInstance, Step step) { + return new StepExecution(step.getName(), new JobExecution(jobInstance)); + } + + @Override + public int getStepExecutionCount(JobInstance jobInstance, Step step) { + // sufficiently high restart count + return 100; + } + }; + StepExecutionMessageHandler handler = createHandler(jobRepository); + JobExecution jobExecution = jobRepository.createJobExecution(new JobSupport("job"), new JobParameters()); + JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobExecution)); + assertNotNull(message); + assertEquals(1, jobExecution.getStepExecutions().size()); + JobExecutionRequest payload = message; + assertEquals(BatchStatus.FAILED, payload.getStatus()); + assertTrue(payload.hasErrors()); + Throwable error = payload.getLastThrowable(); + assertTrue(error instanceof StartLimitExceededException); + String text = error.getMessage(); + assertTrue("Wrong exit description: " + text, text.toLowerCase().contains("start limit")); + } + + /** + * @param jobRepository + * @return + * + */ + public StepExecutionMessageHandler createHandler(JobRepositorySupport jobRepository) { + StepExecutionMessageHandler handler = new StepExecutionMessageHandler(); + StepSupport step = new StepSupport("step"); + step.setStartLimit(10); + handler.setStep(step); + handler.setJobRepository(jobRepository); + return handler; + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/job/TestTasklet.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/job/TestTasklet.java new file mode 100644 index 000000000..9dc78b039 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/job/TestTasklet.java @@ -0,0 +1,34 @@ +/* + * 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.integration.batch.job; + +import org.springframework.batch.core.step.tasklet.Tasklet; +import org.springframework.batch.repeat.ExitStatus; + +/** + * @author Dave Syer + * + */ +public class TestTasklet implements Tasklet { + + /* (non-Javadoc) + * @see org.springframework.batch.core.step.tasklet.Tasklet#execute() + */ + public ExitStatus execute() throws Exception { + return ExitStatus.FINISHED; + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/launch/JobLaunchingMessageHandlerIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/launch/JobLaunchingMessageHandlerIntegrationTests.java new file mode 100644 index 000000000..26df95a52 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/launch/JobLaunchingMessageHandlerIntegrationTests.java @@ -0,0 +1,58 @@ +package org.springframework.integration.batch.launch; + +import static org.junit.Assert.*; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.core.JobExecution; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +@ContextConfiguration() +@RunWith(SpringJUnit4ClassRunner.class) +public class JobLaunchingMessageHandlerIntegrationTests { + + @Autowired @Qualifier("requests") + private MessageChannel requestChannel; + + @Autowired @Qualifier("response") + private MessageChannel responseChannel; + + @Before + public void setUp(){ + requestChannel.purge(null); + responseChannel.purge(null); + } + + + @Test @DirtiesContext @SuppressWarnings("unchecked") + public void testNoReply(){ + requestChannel.send(new StringMessage("testJob")); + Message executionMessage = (Message)responseChannel.receive(1000); + + assertNull("JobExecution message received when no return address set", executionMessage); + } + + + @SuppressWarnings("unchecked") + @Test @DirtiesContext + public void testReply(){ + StringMessage trigger = new StringMessage("testJob"); + trigger.getHeader().setProperty("dontclash", "12"); + trigger.getHeader().setReturnAddress("response"); + requestChannel.send(trigger); + Message executionMessage = (Message)responseChannel.receive(1000); + + assertNotNull("No response received", executionMessage); + JobExecution execution = executionMessage.getPayload(); + assertNotNull("JobExectuion not returned", execution); + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/launch/JobLaunchingMessageHandlerTests.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/launch/JobLaunchingMessageHandlerTests.java new file mode 100644 index 000000000..c67844cf5 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/launch/JobLaunchingMessageHandlerTests.java @@ -0,0 +1,82 @@ +package org.springframework.integration.batch.launch; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicLong; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.launch.JobLauncher; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.batch.JobSupport; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests; + +@ContextConfiguration(locations = { "/job-execution-context.xml" }) +public class JobLaunchingMessageHandlerTests extends AbstractJUnit4SpringContextTests { + + JobLaunchingMessageHandler messageHandler; + + StubJobLauncher jobLauncher; + + + + // @Autowired + // @Qualifier("jobs") TODO: Qualifier seems to be broken here why ????? + public AbstractMessageChannel jobsChannel; + + @Autowired + public MessageBus messageBus; + + @Before + public void setUp() { + jobLauncher = new StubJobLauncher(); + messageHandler = new JobLaunchingMessageHandler(jobLauncher, new StubMessageToJobStrategy()); + jobsChannel = (AbstractMessageChannel) applicationContext.getBean("jobs"); + } + + @Test + public void testSimpleDelivery() throws Exception{ + messageHandler.handle(new StringMessage("testjob")); + + assertEquals("Wrong job count", 1, jobLauncher.jobs.size()); + assertEquals("Wrong job name", jobLauncher.jobs.get(0).getName(), "testjob"); + + } + + private static class StubJobLauncher implements JobLauncher { + + List jobs = new ArrayList(); + + List parameters = new ArrayList(); + + AtomicLong jobId = new AtomicLong(); + + public JobExecution run(Job job, JobParameters jobParameters){ + jobs.add(job); + parameters.add(jobParameters); + return new JobExecution(new JobInstance(jobId.getAndIncrement(), jobParameters, job.getName())); + } + + } + + private static class StubMessageToJobStrategy implements MessageToJobStrategy { + + public Job getJob(Message message) { + String name = (String) message.getPayload(); + return new JobSupport(name); + } + + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/launch/JobLaunchingPostReceiveChannelAdapterTests.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/launch/JobLaunchingPostReceiveChannelAdapterTests.java new file mode 100644 index 000000000..0c8d58fd1 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/launch/JobLaunchingPostReceiveChannelAdapterTests.java @@ -0,0 +1,83 @@ +package org.springframework.integration.batch.launch; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.launch.JobLauncher; +import org.springframework.integration.batch.JobSupport; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.message.StringMessage; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +@ContextConfiguration(locations = { "/job-execution-context.xml" }) +@RunWith(SpringJUnit4ClassRunner.class) +public class JobLaunchingPostReceiveChannelAdapterTests extends AbstractJUnit4SpringContextTests { + + JobLaunchingPostReceiveChannelInterceptor interceptor; + + StubJobLauncher jobLauncher; + + JobSupport job; + +// @Autowired + // @Qualifier("jobs") TODO: Qualifier seems to be broken here why ????? + public AbstractMessageChannel jobsChannel; + + @Before + public void setUp() { + job = new JobSupport(getClass().getName()); + jobLauncher = new StubJobLauncher(); + interceptor = new JobLaunchingPostReceiveChannelInterceptor(job, jobLauncher); + jobsChannel = (AbstractMessageChannel) applicationContext.getBean("jobs"); + jobsChannel.addInterceptor(interceptor); + } + + @Test + public void testJobPassedToLauncherCalled() { + StringMessage message = new StringMessage("test payload"); + jobsChannel.send(message); + assertTrue("Job launcher called before recevie", (jobLauncher.jobs.size() == 0)); + jobsChannel.receive(); + assertEquals(job, jobLauncher.jobs.get(0)); + } + + @Test + public void testMessagePropertiesPassedAsJobParameters() { + StringMessage message = new StringMessage("test payload"); + message.getHeader().setProperty("testOne", "a"); + message.getHeader().setProperty("testTwo", "b"); + jobsChannel.send(message); + jobsChannel.receive(); + JobParameters parameters = jobLauncher.parameters.get(0); + assertEquals("a", parameters.getString("testOne")); + assertEquals("b", parameters.getString("testTwo")); + + } + + private static class StubJobLauncher implements JobLauncher { + + List jobs = new ArrayList(); + + List parameters = new ArrayList(); + + + public JobExecution run(Job job, JobParameters jobParameters){ + jobs.add(job); + parameters.add(jobParameters); + return null; + } + + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/integration/batch/retry/PollableSourceRetryTests.java b/spring-batch-integration/src/test/java/org/springframework/integration/batch/retry/PollableSourceRetryTests.java new file mode 100644 index 000000000..c0c9e33b7 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/integration/batch/retry/PollableSourceRetryTests.java @@ -0,0 +1,462 @@ +package org.springframework.integration.batch.retry; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.Executors; + +import org.aopalliance.aop.Advice; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.Test; +import org.springframework.batch.item.ItemKeyGenerator; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemRecoverer; +import org.springframework.batch.item.support.ListItemReader; +import org.springframework.batch.repeat.interceptor.RepeatOperationsInterceptor; +import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; +import org.springframework.batch.repeat.support.RepeatTemplate; +import org.springframework.batch.retry.interceptor.StatefulRetryOperationsInterceptor; +import org.springframework.batch.support.PropertiesConverter; +import org.springframework.batch.support.transaction.ResourcelessTransactionManager; +import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; +import org.springframework.context.Lifecycle; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.dispatcher.DirectChannel; +import org.springframework.integration.endpoint.SourceEndpoint; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.Source; +import org.springframework.integration.message.Target; +import org.springframework.integration.scheduling.MessagingTaskScheduler; +import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; +import org.springframework.integration.util.ErrorHandler; +import org.springframework.transaction.interceptor.TransactionInterceptor; +import org.springframework.util.StringUtils; + +public class PollableSourceRetryTests { + + private Log logger = LogFactory.getLog(getClass()); + + private List processed = new ArrayList(); + + protected List recovered = new ArrayList(); + + public void add(String str) { + logger.debug("Adding: " + str); + processed.add(str); + } + + ItemKeyGenerator itemKeyGenerator = new ItemKeyGenerator() { + @SuppressWarnings("unchecked") + public Object getKey(Object item) { + if (item == null) { + return "NULL"; + } + if (item.getClass().isArray()) { + item = ((Object[]) item)[0]; + } + return ((Message) item).getPayload(); + } + }; + + @SuppressWarnings("unchecked") + @Test + public void testSimpleTransactionalPolling() throws Exception { + + List list = TransactionAwareProxyFactory.createTransactionalList(); + list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,c,d,e,f,g,h,j,k"))); + int beforeCount = list.size(); + + Target handler = new Target() { + public boolean send(Message message) { + Object payload = message.getPayload(); + logger.debug("Handling: " + payload); + return processed.add((String) payload); + } + }; + Source source = getPollableSource(list); + DirectChannel channel = getChannel(handler, source); + SourceEndpoint endpoint = getSourceEndpoint(source, channel); + endpoint.setDispatchAdviceChain(Arrays.asList(new Advice[] { getTransactionInterceptor() })); + endpoint.initializeTask(); + MessagingTaskScheduler scheduler = getSchedulerWithErrorHandler(endpoint); + + waitForResults(scheduler, 2, 40); + + assertEquals(2, processed.size()); + + assertEquals(beforeCount - list.size(), processed.size()); + assertEquals("a", processed.get(0)); + + } + + @SuppressWarnings("unchecked") + @Test + public void testNonTransactionalPollingWithRollback() throws Exception { + + List list = TransactionAwareProxyFactory.createTransactionalList(); + list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,c,d,e,f,g,h,j,k"))); + int beforeCount = list.size(); + + Target handler = new Target() { + public boolean send(Message message) { + Object payload = message.getPayload(); + logger.debug("Handling: " + payload); + processed.add((String) payload); + throw new RuntimeException("Planned failure: " + payload); + } + }; + Source source = getPollableSource(list); + DirectChannel channel = getChannel(handler, source); + SourceEndpoint endpoint = getSourceEndpoint(source, channel); + MessagingTaskScheduler scheduler = getSchedulerWithErrorHandler(endpoint); + + waitForResults(scheduler, 2, 20); + + assertEquals(2, processed.size()); + + // None rolled back because there was no transaction + assertEquals(beforeCount - list.size(), 2); + assertEquals("a", processed.get(0)); + assertEquals("b", processed.get(1)); + + } + + @SuppressWarnings("unchecked") + @Test + public void testTransactionalHandlingWithUnconditionalRollback() throws Exception { + + List list = TransactionAwareProxyFactory.createTransactionalList(); + list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,c,d,e,f,g,h,j,k"))); + int beforeCount = list.size(); + + Target handler = new Target() { + public boolean send(Message message) { + Object payload = message.getPayload(); + logger.debug("Handling: " + payload); + processed.add((String) payload); + throw new RuntimeException("Planned failure: " + payload); + } + }; + Source source = getPollableSource(list); + DirectChannel channel = getChannel(handler, source); + SourceEndpoint endpoint = getSourceEndpoint(source, channel); + endpoint.setTaskAdviceChain(Arrays.asList(new Advice[] { getTransactionInterceptor() })); + endpoint.initializeTask(); + MessagingTaskScheduler scheduler = getSchedulerWithErrorHandler(endpoint); + + waitForResults(scheduler, 2, 20); + + assertEquals(2, processed.size()); + + // TODO: this would fail if exception not propagated: INT-184. + // All rolled back + assertEquals(beforeCount - list.size(), 0); + assertEquals("a", processed.get(0)); + // processed twice and rolled back both times + assertEquals("a", processed.get(1)); + + } + + @SuppressWarnings("unchecked") + @Test + public void testTransactionalHandlingWithRollback() throws Exception { + + List list = TransactionAwareProxyFactory.createTransactionalList(); + list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k"))); + int beforeCount = list.size(); + + Target handler = new Target() { + public boolean send(Message message) { + Object payload = message.getPayload(); + logger.debug("Handling: " + payload); + boolean result = processed.add((String) payload); + if ("fail".equals(payload)) { + throw new RuntimeException("Planned failure: " + payload); + } + return result; + } + }; + + Source source = getPollableSource(list); + DirectChannel channel = getChannel(handler, source); + SourceEndpoint endpoint = getSourceEndpoint(source, channel); + endpoint.setTaskAdviceChain(Arrays.asList(new Advice[] { getTransactionInterceptor() })); + endpoint.initializeTask(); + MessagingTaskScheduler scheduler = getSchedulerWithErrorHandler(endpoint); + + waitForResults(scheduler, 5, 20); + + assertEquals(5, processed.size()); + assertFalse("No messages got to processor", processed.isEmpty()); + // First two TX succeed, and the rest rolled back so list has had two + // elements popped off + assertEquals(beforeCount - 2, list.size()); + assertEquals("a", processed.get(0)); + assertEquals("b", processed.get(1)); + // stuck in effectively an infinite loop - it fails every time... + assertEquals("fail", processed.get(2)); + assertEquals("fail", processed.get(3)); + assertEquals("fail", processed.get(4)); + + } + + @SuppressWarnings("unchecked") + @Test + public void testTransactionalHandlingWithRepeat() throws Exception { + + List list = TransactionAwareProxyFactory.createTransactionalList(); + list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k"))); + int beforeCount = list.size(); + + Target handler = new Target() { + public boolean send(Message message) { + Object payload = message.getPayload(); + logger.debug("Handling: " + payload); + boolean result = processed.add((String) payload); + if ("fail".equals(payload)) { + throw new RuntimeException("Planned failure: " + payload); + } + return result; + } + }; + + Source source = getPollableSource(list); + DirectChannel channel = getChannel(handler, source); + SourceEndpoint endpoint = getSourceEndpoint(source, channel); + endpoint.setTaskAdviceChain(Arrays.asList(new Advice[] { getTransactionInterceptor(), + getRepeatOperationsInterceptor(3) })); + endpoint.initializeTask(); + MessagingTaskScheduler scheduler = getSchedulerWithErrorHandler(endpoint); + + waitForResults(scheduler, 6, 100); + + assertEquals(6, processed.size()); + assertFalse("No messages got to processor", processed.isEmpty()); + // Two TX rolled back so list is same size as when it started + assertEquals(beforeCount, list.size()); + assertEquals("a", processed.get(0)); + assertEquals("b", processed.get(1)); + // stuck in effectively an infinite loop - it fails every time with the + // same 3 records... + assertEquals("fail", processed.get(2)); + assertEquals("a", processed.get(3)); + assertEquals("b", processed.get(4)); + + } + + @SuppressWarnings("unchecked") + @Test + public void testTransactionalHandlingWithRetry() throws Exception { + + List list = TransactionAwareProxyFactory.createTransactionalList(); + list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k"))); + int beforeCount = list.size(); + + Target handler = new Target() { + public boolean send(Message message) { + if (message == null) { + return false; + } + Object payload = message.getPayload(); + logger.debug("Handling: " + payload); + boolean result = processed.add((String) payload); + // INT-184 this won't work if it is a "real" handler that throws + // MessageHandlingException + if ("fail".equals(payload)) { + throw new RuntimeException("Planned failure: " + payload); + } + return result; + } + }; + + Source source = getPollableSource(list); + MessageChannel channel = getChannel(handler, source); + SourceEndpoint endpoint = getSourceEndpoint(source, channel); + endpoint.setTaskAdviceChain(Arrays.asList(new Advice[] { getTransactionInterceptor() })); + endpoint + .setDispatchAdviceChain(Arrays.asList(new Advice[] { getRetryOperationsInterceptor(itemKeyGenerator) })); + endpoint.initializeTask(); + MessagingTaskScheduler scheduler = getSchedulerWithErrorHandler(endpoint); + + waitForResults(scheduler, 4, 20); + + assertEquals(4, processed.size()); + assertEquals(1, recovered.size()); + assertFalse("No messages got to processor", processed.isEmpty()); + // 4 items from list should have been processed (with no repeats, since + // the failed item was recovered with no retry - NeverRetryPolicy) + assertEquals(beforeCount - 4, list.size()); + assertEquals("a", processed.get(0)); + assertEquals("b", processed.get(1)); + // retry makes it fail once then recover... + assertEquals("fail", processed.get(2)); + assertEquals("d", processed.get(3)); + + } + + @SuppressWarnings("unchecked") + @Test + public void testTransactionalHandlingWithRepeatAndRetry() throws Exception { + + List list = TransactionAwareProxyFactory.createTransactionalList(); + list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,fail,c,d,e,f,g,h,j,k"))); + int beforeCount = list.size(); + + Target handler = new Target() { + public boolean send(Message message) { + Object payload = message.getPayload(); + logger.debug("Handling: " + payload); + boolean result = processed.add((String) payload); + if ("fail".equals(payload)) { + throw new RuntimeException("Planned failure: " + payload); + } + return result; + } + }; + + Source source = getPollableSource(list); + MessageChannel channel = getChannel(handler, source); + SourceEndpoint endpoint = getSourceEndpoint(source, channel); + endpoint.setTaskAdviceChain(Arrays.asList(new Advice[] { getTransactionInterceptor(), + getRepeatOperationsInterceptor(3) })); + endpoint + .setDispatchAdviceChain(Arrays.asList(new Advice[] { getRetryOperationsInterceptor(itemKeyGenerator) })); + endpoint.initializeTask(); + MessagingTaskScheduler scheduler = getSchedulerWithErrorHandler(endpoint); + + waitForResults(scheduler, 6, 100); + System.err.println(processed); + System.err.println(list); + + assertEquals(6, processed.size()); + assertFalse("No messages got to processor", processed.isEmpty()); + // One roll back and then start again with a,b,d,e + assertEquals(beforeCount - 5, list.size()); + assertEquals("a", processed.get(0)); + assertEquals("fail", processed.get(1)); + // retry makes it fail once then recover... + assertEquals("a", processed.get(2)); + assertEquals("c", processed.get(3)); + assertEquals("d", processed.get(4)); + + } + + /** + * @param source + * @param channel + * @return + */ + private SourceEndpoint getSourceEndpoint(Source source, MessageChannel channel) { + PollingSchedule schedule = new PollingSchedule(50); + schedule.setFixedRate(true); // used to be the default + return new SourceEndpoint(source, channel, schedule); + } + + /** + * @param handler + * @param source + * @return + */ + private DirectChannel getChannel(Target handler, Source source) { + DirectChannel channel = new DirectChannel(source); + channel.setName("input"); + channel.subscribe(handler); + return channel; + } + + private void waitForResults(Lifecycle lifecycle, int count, int maxTries) throws InterruptedException { + lifecycle.start(); + int timeout = 0; + while (processed.size() < count && timeout++ < maxTries) { + Thread.sleep(10); + } + lifecycle.stop(); + } + + private Source getPollableSource(List list) { + final ItemReader reader = new ListItemReader(list) { + public Object read() { + Object item = super.read(); + logger.debug("Reading: " + item); + return item; + } + }; + Source source = new Source() { + public Message receive() { + try { + return new GenericMessage(reader.read()); + } + catch (RuntimeException e) { + throw e; + } + catch (Exception e) { + throw new IllegalStateException(e); + } + } + }; + return source; + } + + // Workaround for INT-182 + private MessagingTaskScheduler getSchedulerWithErrorHandler(Runnable task) { + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(Executors + .newSingleThreadScheduledExecutor()); + scheduler.setErrorHandler(new ErrorHandler() { + public void handle(Throwable t) { + logger.error("Exception in scheduler", t); + // throw (RuntimeException)t; + } + }); + scheduler.schedule(task); + return scheduler; + } + + /** + * @param itemKeyGenerator + * @return + */ + private StatefulRetryOperationsInterceptor getRetryOperationsInterceptor(ItemKeyGenerator itemKeyGenerator) { + StatefulRetryOperationsInterceptor advice = new StatefulRetryOperationsInterceptor(); + advice.setRecoverer(new ItemRecoverer() { + @SuppressWarnings("unchecked") + public Object recover(Object data, Throwable cause) { + if (data == null) { + return false; + } + if (data.getClass().isArray()) { + data = ((Object[]) data)[0]; + } + recovered.add(((Message) data).getPayload()); + return true; + } + }); + advice.setKeyGenerator(itemKeyGenerator); + return advice; + } + + /** + * @return + */ + private TransactionInterceptor getTransactionInterceptor() { + return new TransactionInterceptor(new ResourcelessTransactionManager(), PropertiesConverter + .stringToProperties("*=PROPAGATION_REQUIRED")); + } + + /** + * @param commitInterval + * @return + */ + private RepeatOperationsInterceptor getRepeatOperationsInterceptor(int commitInterval) { + RepeatOperationsInterceptor advice = new RepeatOperationsInterceptor(); + RepeatTemplate repeatTemplate = new RepeatTemplate(); + repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(commitInterval)); + advice.setRepeatOperations(repeatTemplate); + return advice; + } +} diff --git a/spring-batch-integration/src/test/resources/integration-context.xml b/spring-batch-integration/src/test/resources/integration-context.xml new file mode 100644 index 000000000..0d876a36d --- /dev/null +++ b/spring-batch-integration/src/test/resources/integration-context.xml @@ -0,0 +1,17 @@ + + + + + + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/job-execution-context.xml b/spring-batch-integration/src/test/resources/job-execution-context.xml new file mode 100644 index 000000000..10798967a --- /dev/null +++ b/spring-batch-integration/src/test/resources/job-execution-context.xml @@ -0,0 +1,14 @@ + + + + + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/log4j.properties b/spring-batch-integration/src/test/resources/log4j.properties new file mode 100644 index 000000000..b5a00f56b --- /dev/null +++ b/spring-batch-integration/src/test/resources/log4j.properties @@ -0,0 +1,10 @@ +log4j.rootCategory=WARN, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%d %5p %t [%c] - <%m>%n + +log4j.logger.org.springframework.integration.batch=DEBUG +log4j.logger.org.springframework.batch=DEBUG +log4j.category.org.springframework.integration=DEBUG +log4j.category.org.springframework.transaction=DEBUG \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/integration/batch/chunk/ChunkMessageItemWriterIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/integration/batch/chunk/ChunkMessageItemWriterIntegrationTests-context.xml new file mode 100644 index 000000000..4c1bc61e9 --- /dev/null +++ b/spring-batch-integration/src/test/resources/org/springframework/integration/batch/chunk/ChunkMessageItemWriterIntegrationTests-context.xml @@ -0,0 +1,33 @@ + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/integration/batch/file/ResourceSplitterIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/integration/batch/file/ResourceSplitterIntegrationTests-context.xml new file mode 100644 index 000000000..1a3f63769 --- /dev/null +++ b/spring-batch-integration/src/test/resources/org/springframework/integration/batch/file/ResourceSplitterIntegrationTests-context.xml @@ -0,0 +1,18 @@ + + + + + + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/integration/batch/item/MessageChannelItemWriterIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/integration/batch/item/MessageChannelItemWriterIntegrationTests-context.xml new file mode 100644 index 000000000..a0d338d14 --- /dev/null +++ b/spring-batch-integration/src/test/resources/org/springframework/integration/batch/item/MessageChannelItemWriterIntegrationTests-context.xml @@ -0,0 +1,19 @@ + + + + + + + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/integration/batch/job/MessageOrientedStepIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/integration/batch/job/MessageOrientedStepIntegrationTests-context.xml new file mode 100644 index 000000000..e292418c4 --- /dev/null +++ b/spring-batch-integration/src/test/resources/org/springframework/integration/batch/job/MessageOrientedStepIntegrationTests-context.xml @@ -0,0 +1,47 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/integration/batch/launch/JobLaunchingMessageHandlerIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/integration/batch/launch/JobLaunchingMessageHandlerIntegrationTests-context.xml new file mode 100644 index 000000000..a30b456c5 --- /dev/null +++ b/spring-batch-integration/src/test/resources/org/springframework/integration/batch/launch/JobLaunchingMessageHandlerIntegrationTests-context.xml @@ -0,0 +1,44 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/simple-job-launcher-context.xml b/spring-batch-integration/src/test/resources/simple-job-launcher-context.xml new file mode 100644 index 000000000..30911ad32 --- /dev/null +++ b/spring-batch-integration/src/test/resources/simple-job-launcher-context.xml @@ -0,0 +1,80 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file