diff --git a/spring-batch-integration/.settings/org.springframework.ide.eclipse.beans.core.prefs b/spring-batch-integration/.settings/org.springframework.ide.eclipse.beans.core.prefs new file mode 100644 index 000000000..a03e4a5ac --- /dev/null +++ b/spring-batch-integration/.settings/org.springframework.ide.eclipse.beans.core.prefs @@ -0,0 +1,3 @@ +#Wed May 06 08:20:22 BST 2009 +eclipse.preferences.version=1 +org.springframework.ide.eclipse.beans.core.ignoreMissingNamespaceHandler=false diff --git a/spring-batch-integration/.springBeans b/spring-batch-integration/.springBeans index d5759e916..8acabca68 100644 --- a/spring-batch-integration/.springBeans +++ b/spring-batch-integration/.springBeans @@ -1,7 +1,7 @@ 1 - + @@ -18,6 +18,8 @@ src/test/resources/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests-context.xml src/test/resources/org/springframework/batch/integration/SmokeTests-context.xml src/test/resources/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests-context.xml + src/test/resources/org/springframework/batch/integration/item/MessagingGatewayIntegrationTests-context.xml + src/test/resources/org/springframework/batch/integration/partition/VanillaIntegrationTests-context.xml diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/BeanFactoryStepLocator.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/BeanFactoryStepLocator.java new file mode 100644 index 000000000..262b450a7 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/BeanFactoryStepLocator.java @@ -0,0 +1,47 @@ +package org.springframework.batch.integration.partition; + +import java.util.Arrays; +import java.util.Collection; + +import org.springframework.batch.core.Step; +import org.springframework.batch.core.step.StepLocator; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.util.Assert; + +/** + * A {@link StepLocator} implementation that just looks in its enclosing bean + * factory for components of type {@link Step}. + * + * @author Dave Syer + * + */ +public class BeanFactoryStepLocator implements StepLocator, BeanFactoryAware { + + private BeanFactory beanFactory; + + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.beanFactory = beanFactory; + } + + /** + * Look up a bean with the provided name of type {@link Step}. + * @see StepLocator#getStep(String) + */ + public Step getStep(String stepName) { + return (Step) beanFactory.getBean(stepName, Step.class); + } + + /** + * Look in the bean factory for all beans of type {@link Step}. + * @throws IllegalStateException if the {@link BeanFactory} is not listable + * @see StepLocator#getStepNames() + */ + public Collection getStepNames() { + Assert.state(beanFactory instanceof ListableBeanFactory, "BeanFactory is not listable."); + return Arrays.asList(((ListableBeanFactory) beanFactory).getBeanNamesForType(Step.class)); + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandler.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandler.java new file mode 100644 index 000000000..b5ce67a11 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandler.java @@ -0,0 +1,138 @@ +package org.springframework.batch.integration.partition; + +import java.util.Collection; +import java.util.List; +import java.util.Set; + +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.partition.PartitionHandler; +import org.springframework.batch.core.partition.StepExecutionSplitter; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.integration.annotation.Aggregator; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.gateway.MessagingGateway; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.util.Assert; + +/** + * A {@link PartitionHandler} that uses {@link MessageChannel} instances to send + * instructions to remote workers and receive their responses. The + * {@link MessageChannel} provides a nice abstraction so that the location of + * the workers and the transport used to communicate with them can be changed at + * run time. The communication with the remote workers does not need to be + * transactional or have guaranteed delivery, so a local thread pool based + * implementation works as well as a remote web service or JMS implementation. + * If a remote worker fails or doesn't send a reply message, the job will fail + * and can be restarted to pick up missing messages and processing. The remote + * workers need access to the Spring Batch {@link JobRepository} so that the + * shared state across those restarts can be managed centrally. + * + * @author Dave Syer + * + */ +@MessageEndpoint +public class MessageChannelPartitionHandler implements PartitionHandler { + + private int gridSize = 1; + + private MessagingGateway messagingGateway; + + private String stepName; + + public void afterPropertiesSet() throws Exception { + Assert.notNull(stepName, "A step name must be provided for the remote workers."); + Assert.state(messagingGateway != null, "The MessagingGateway must be set"); + } + + /** + * A pre-configured gateway for sending and receiving messages to the remote + * workers. Using this property allows a large degree of control over the + * timeouts and other properties of the send. It should have channels set up + * internally: + * + * The timeout for the repoy should be set sufficiently long that the remote + * steps have time to complete. + * + * @param messagingGateway the {@link MessagingGateway} to set + */ + public void setMessagingGateway(MessagingGateway messagingGateway) { + this.messagingGateway = messagingGateway; + } + + /** + * Passed to the {@link StepExecutionSplitter} in the + * {@link #handle(StepExecutionSplitter, StepExecution)} method, instructing + * it how many {@link StepExecution} instances are required, ideally. The + * {@link StepExecutionSplitter} is allowed to ignore the grid size in the + * case of a restart, since the input data partitions must be preserved. + * + * @param gridSize the number of step executions that will be created + */ + public void setGridSize(int gridSize) { + this.gridSize = gridSize; + } + + /** + * The name of the {@link Step} that will be used to execute the partitioned + * {@link StepExecution}. This is a regular Spring Batch step, with all the + * business logic required to complete an execution based on the input + * parameters in its {@link StepExecution} context. The name will be + * translated into a {@link Step} instance by the remote worker. + * + * @param stepName the name of the {@link Step} instance to execute business + * logic + */ + public void setStepName(String stepName) { + this.stepName = stepName; + } + + /** + * @param messages the messages to be aggregated + * @return the list as it was passed in + */ + @Aggregator(sendPartialResultsOnTimeout = true) + public List aggregate(List messages) { + return messages; + } + + /** + * Sends {@link StepExecutionRequest} objects to the request channel of the + * {@link MessagingGateway}, and then receives the result back as a list of + * {@link StepExecution} on a reply channel. Use the + * {@link #aggregate(List)} method as an aggregator of the individual remote + * replies. The receive timeout needs to be set realistically in the + * {@link MessagingGateway} and the aggregator, so that there is a + * good chance of all work being done. + * + * @see PartitionHandler#handle(StepExecutionSplitter, StepExecution) + */ + public Collection handle(StepExecutionSplitter stepExecutionSplitter, + StepExecution masterStepExecution) throws Exception { + + Set split = stepExecutionSplitter.split(masterStepExecution, gridSize); + int count = 0; + for (StepExecution stepExecution : split) { + messagingGateway.send(createMessage(count++, split.size(), new StepExecutionRequest(stepName, stepExecution + .getJobExecutionId(), stepExecution.getId()))); + } + + @SuppressWarnings("unchecked") + Collection result = (Collection) messagingGateway.receive(); + return result; + + } + + private Message createMessage(int sequenceNumber, int sequenceSize, + StepExecutionRequest stepExecutionRequest) { + return MessageBuilder.withPayload(stepExecutionRequest).setSequenceNumber(sequenceNumber).setSequenceSize( + sequenceSize).setCorrelationId( + stepExecutionRequest.getJobExecutionId() + ":" + stepExecutionRequest.getStepName()).build(); + } +} diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/StepExecutionRequest.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/StepExecutionRequest.java new file mode 100644 index 000000000..b412a6b73 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/StepExecutionRequest.java @@ -0,0 +1,37 @@ +package org.springframework.batch.integration.partition; + +import java.io.Serializable; + +public class StepExecutionRequest implements Serializable { + + private final Long stepExecutionId; + + private final String stepName; + + private final Long jobExecutionId; + + public StepExecutionRequest(String stepName, Long jobExecutionId, Long stepExecutionId) { + this.stepName = stepName; + this.jobExecutionId = jobExecutionId; + this.stepExecutionId = stepExecutionId; + } + + public Long getJobExecutionId() { + return jobExecutionId; + } + + public Long getStepExecutionId() { + return stepExecutionId; + } + + public String getStepName() { + return stepName; + } + + @Override + public String toString() { + return String.format("StepExecutionRequest: [jobExecutionId=%d, stepExecutionId=%d, stepName=%s]", + jobExecutionId, stepExecutionId, stepName); + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/StepExecutionRequestHandler.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/StepExecutionRequestHandler.java new file mode 100644 index 000000000..e50bcf3c5 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/StepExecutionRequestHandler.java @@ -0,0 +1,81 @@ +package org.springframework.batch.integration.partition; + +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.explore.JobExplorer; +import org.springframework.batch.core.step.StepLocator; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.ServiceActivator; + +/** + * A {@link MessageEndpoint} that can handle a {@link StepExecutionRequest} and + * return a {@link StepExecution} as the result. Typically these need to be + * aggregated into a response to a partition handler. + * + * @author Dave Syer + * + */ +@MessageEndpoint +public class StepExecutionRequestHandler { + + private JobExplorer jobExplorer; + + private StepLocator stepLocator; + + /** + * Used to locate a {@link Step} to execute for each request. + * @param stepLocator a {@link StepLocator} + */ + public void setStepLocator(StepLocator stepLocator) { + this.stepLocator = stepLocator; + } + + /** + * An explorer that should be used to check for {@link StepExecution} + * completion. + * + * @param jobExplorer a {@link JobExplorer} that is linked to the shared + * repository used by all remote workers. + */ + public void setJobExplorer(JobExplorer jobExplorer) { + this.jobExplorer = jobExplorer; + } + + @ServiceActivator + public StepExecution handle(StepExecutionRequest request) { + + Long jobExecutionId = request.getJobExecutionId(); + Long stepExecutionId = request.getStepExecutionId(); + StepExecution stepExecution = jobExplorer.getStepExecution(jobExecutionId, stepExecutionId); + if (stepExecution == null) { + // TODO: create new Exception + throw new IllegalStateException("No StepExecution could be located for this request: " + request); + } + + String stepName = request.getStepName(); + Step step = stepLocator.getStep(stepName); + if (step == null) { + // TODO: create new Exception + throw new IllegalStateException(String.format("No Step with name [%s] could be located.", stepName)); + } + + try { + step.execute(stepExecution); + } + catch (JobInterruptedException e) { + stepExecution.setStatus(BatchStatus.STOPPED); + // TODO: maybe update stepExecution in repository + } + catch (Throwable e) { + stepExecution.addFailureException(e); + stepExecution.setStatus(BatchStatus.FAILED); + // TODO: maybe update stepExecution in repository + } + + return stepExecution; + + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/BeanFactoryStepLocatorTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/BeanFactoryStepLocatorTests.java new file mode 100644 index 000000000..c50aa27ba --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/BeanFactoryStepLocatorTests.java @@ -0,0 +1,59 @@ +package org.springframework.batch.integration.partition; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import org.junit.Test; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.beans.factory.support.DefaultListableBeanFactory; + + +public class BeanFactoryStepLocatorTests { + + private BeanFactoryStepLocator stepLocator = new BeanFactoryStepLocator(); + private DefaultListableBeanFactory beanFactory = new DefaultListableBeanFactory(); + + @Test + public void testGetStep() throws Exception { + beanFactory.registerSingleton("foo", new StubStep("foo")); + stepLocator.setBeanFactory(beanFactory); + assertNotNull(stepLocator.getStep("foo")); + } + + @Test + public void testGetStepNames() throws Exception { + beanFactory.registerSingleton("foo", new StubStep("foo")); + beanFactory.registerSingleton("bar", new StubStep("bar")); + stepLocator.setBeanFactory(beanFactory); + assertEquals(2, stepLocator.getStepNames().size()); + } + + private static final class StubStep implements Step { + + private String name; + + public StubStep(String name) { + this.name = name; + } + + public void execute(StepExecution stepExecution) throws JobInterruptedException { + } + + public String getName() { + return name; + } + + public int getStartLimit() { + // TODO Auto-generated method stub + return 0; + } + + public boolean isAllowStartIfComplete() { + // TODO Auto-generated method stub + return false; + } + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/ExampleItemReader.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/ExampleItemReader.java new file mode 100644 index 000000000..c9b1c299d --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/ExampleItemReader.java @@ -0,0 +1,57 @@ +package org.springframework.batch.integration.partition; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemStreamException; + +/** + * {@link ItemReader} with hard-coded input data. + */ +public class ExampleItemReader implements ItemReader, ItemStream { + + private Log logger = LogFactory.getLog(getClass()); + + private String[] input = { "Hello", "world!", "Go", "on", "punk", "make", "my", "day!" }; + + private int index = 0; + + public static volatile boolean fail = false; + + /** + * Reads next record from input + */ + public String read() throws Exception { + if (index >= input.length) { + return null; + } + logger.info(String.format("Processing input index=%s, item=%s, in (%s)", index, input[index], this)); + if (fail && index == 4) { + synchronized (ExampleItemReader.class) { + if (fail) { + // Only fail once per flag setting... + fail = false; + logger.info(String.format("Throwing exception index=%s, item=%s, in (%s)", index, input[index], + this)); + index++; + throw new RuntimeException("Planned failure"); + } + } + } + return input[index++]; + } + + public void close() throws ItemStreamException { + } + + public void open(ExecutionContext executionContext) throws ItemStreamException { + index = (int) executionContext.getLong("POSITION", 0); + } + + public void update(ExecutionContext executionContext) throws ItemStreamException { + executionContext.putLong("POSITION", index); + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/ExampleItemReaderTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/ExampleItemReaderTests.java new file mode 100644 index 000000000..3f2a7f96e --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/ExampleItemReaderTests.java @@ -0,0 +1,70 @@ +package org.springframework.batch.integration.partition; + +import static org.junit.Assert.*; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.item.ExecutionContext; + +public class ExampleItemReaderTests { + + private ExampleItemReader reader = new ExampleItemReader(); + + @Before + @After + public void ensureFailFlagUnset() { + ExampleItemReader.fail = false; + } + + @Test + public void testRead() throws Exception { + int count = 0; + while (reader.read()!=null) { + count++; + } + assertEquals(8, count); + } + + @Test + public void testOpen() throws Exception { + ExecutionContext context = new ExecutionContext(); + for (int i=0; i<4; i++) { + reader.read(); + } + reader.update(context); + reader.open(context); + int count = 0; + while (reader.read()!=null) { + count++; + } + assertEquals(4, count); + } + + @Test + public void testFailAndRestart() throws Exception { + ExecutionContext context = new ExecutionContext(); + ExampleItemReader.fail = true; + for (int i=0; i<4; i++) { + reader.read(); + reader.update(context); + } + try { + reader.read(); + reader.update(context); + fail("Expected Exception"); + } + catch (Exception e) { + // expected + assertEquals("Planned failure", e.getMessage()); + } + assertFalse(ExampleItemReader.fail); + reader.open(context); + int count = 0; + while (reader.read()!=null) { + count++; + } + assertEquals(4, count); + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/ExampleItemWriter.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/ExampleItemWriter.java new file mode 100644 index 000000000..fd3965190 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/ExampleItemWriter.java @@ -0,0 +1,23 @@ +package org.springframework.batch.integration.partition; + +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.item.ItemWriter; + +/** + * Dummy {@link ItemWriter} which only logs data it receives. + */ +public class ExampleItemWriter implements ItemWriter { + + private static final Log log = LogFactory.getLog(ExampleItemWriter.class); + + /** + * @see ItemWriter#write(List) + */ + public void write(List data) throws Exception { + log.info(data); + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/VanillaIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/VanillaIntegrationTests.java new file mode 100644 index 000000000..e4484528b --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/VanillaIntegrationTests.java @@ -0,0 +1,70 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.integration.partition; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.util.List; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.core.BatchStatus; +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.explore.JobExplorer; +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 VanillaIntegrationTests { + + @Autowired + private JobLauncher jobLauncher; + + @Autowired + private Job job; + + @Autowired + private JobExplorer jobExplorer; + + @Test + public void testSimpleProperties() throws Exception { + assertNotNull(jobLauncher); + } + + @Test + public void testLaunchJob() throws Exception { + int before = jobExplorer.getJobInstances(job.getName(), 0, 100).size(); + assertNotNull(jobLauncher.run(job, new JobParameters())); + List jobInstances = jobExplorer.getJobInstances(job.getName(), 0, 100); + int after = jobInstances.size(); + assertEquals(1, after-before); + JobExecution jobExecution = jobExplorer.getJobExecutions(jobInstances.get(jobInstances.size()-1)).get(0); + assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); + assertEquals(3, jobExecution.getStepExecutions().size()); + } + +} diff --git a/spring-batch-integration/src/test/resources/log4j.properties b/spring-batch-integration/src/test/resources/log4j.properties index 2f533c778..8c5be0336 100644 --- a/spring-batch-integration/src/test/resources/log4j.properties +++ b/spring-batch-integration/src/test/resources/log4j.properties @@ -4,6 +4,6 @@ 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.batch=DEBUG +log4j.logger.org.springframework.batch.core=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/batch/integration/SmokeTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/SmokeTests-context.xml index 98df0f185..09184a31f 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/SmokeTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/SmokeTests-context.xml @@ -10,10 +10,11 @@ http://www.springframework.org/schema/context/spring-context-2.5.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd"> - + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/partition/VanillaIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/partition/VanillaIntegrationTests-context.xml new file mode 100644 index 000000000..2f6d0b4fb --- /dev/null +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/partition/VanillaIntegrationTests-context.xml @@ -0,0 +1,96 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +