diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/StepListenerFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/StepListenerFactoryBean.java index 462850238..724170c9a 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/StepListenerFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/StepListenerFactoryBean.java @@ -78,7 +78,7 @@ public class StepListenerFactoryBean implements FactoryBean, InitializingBean{ Set> listenerInterfaces = new HashSet>(); - //For every entry in th emap, try and find a method by interface, name, or annotation. If the same + //For every entry in the map, try and find a method by interface, name, or annotation. If the same for(Entry entry : metaDataMap.entrySet()){ StepListenerMetaData metaData = StepListenerMetaData.fromPropertyName(entry.getKey()); Set invokers = new NullIgnoringSet(); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/MultiResourcePartitioner.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/MultiResourcePartitioner.java new file mode 100644 index 000000000..a77a02a30 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/MultiResourcePartitioner.java @@ -0,0 +1,89 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.core.partition.support; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +import org.springframework.batch.item.ExecutionContext; +import org.springframework.core.io.Resource; +import org.springframework.util.Assert; + +/** + * Implementation of {@link Partitioner} that locates multiple resources and + * associates their file names with execution context keys. Creates an + * {@link ExecutionContext} per resource, and labels them as + * {partition0, partition1, ..., partitionN}. The grid size is + * ignored. + * + * @author Dave Syer + * + */ +public class MultiResourcePartitioner implements Partitioner { + + private static final String DEFAULT_KEY_NAME = "fileName"; + + private static final String PARTITION_KEY = "partition"; + + private Resource[] resources = new Resource[0]; + + private String keyName = DEFAULT_KEY_NAME; + + /** + * The resources to assign to each partition. In Spring configuration you + * can use a pattern to select multiple resources. + * @param resources the resources to use + */ + public void setResources(Resource[] resources) { + this.resources = resources; + } + + /** + * The name of the key for the file name in each {@link ExecutionContext}. + * Defaults to "fileName". + * @param keyName the value of the key + */ + public void setKeyName(String keyName) { + this.keyName = keyName; + } + + /** + * Assign the filename of each of the injected resources to an + * {@link ExecutionContext}. + * + * @see Partitioner#partition(int) + */ + public Map partition(int gridSize) { + Map map = new HashMap(gridSize); + int i = 0; + for (Resource resource : resources) { + ExecutionContext context = new ExecutionContext(); + Assert.state(resource.exists(), "Resource does not exist: "+resource); + try { + context.putString(keyName, resource.getURL().toExternalForm()); + } + catch (IOException e) { + throw new IllegalArgumentException("File could not be located for: "+resource, e); + } + map.put(PARTITION_KEY + i, context); + i++; + } + return map; + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/MultiResourcePartitionerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/MultiResourcePartitionerTests.java new file mode 100644 index 000000000..34a323d97 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/MultiResourcePartitionerTests.java @@ -0,0 +1,54 @@ +package org.springframework.batch.core.partition.support; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.Map; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.core.io.FileSystemResource; +import org.springframework.core.io.Resource; +import org.springframework.core.io.UrlResource; +import org.springframework.core.io.support.ResourceArrayPropertyEditor; + +public class MultiResourcePartitionerTests { + + private MultiResourcePartitioner partitioner = new MultiResourcePartitioner(); + + @Before + public void setUp() { + ResourceArrayPropertyEditor editor = new ResourceArrayPropertyEditor(); + editor.setAsText("classpath:log4j*"); + partitioner.setResources((Resource[]) editor.getValue()); + } + + @Test(expected = IllegalStateException.class) + public void testMissingResource() { + partitioner.setResources(new Resource[] { new FileSystemResource("does-not-exist") }); + partitioner.partition(0); + } + + @Test + public void testPartitionSizeAndKey() { + Map partition = partitioner.partition(0); + assertEquals(1, partition.size()); + assertTrue(partition.containsKey("partition0")); + } + + @Test + public void testReadFile() throws Exception { + Map partition = partitioner.partition(0); + String url = partition.get("partition0").getString("fileName"); + assertTrue(new UrlResource(url).exists()); + } + + @Test + public void testSetKeyName() { + partitioner.setKeyName("foo"); + Map partition = partitioner.partition(0); + assertTrue(partition.get("partition0").containsKey("foo")); + } + +} diff --git a/spring-batch-samples/.springBeans b/spring-batch-samples/.springBeans index a0ae44302..64eb692b5 100644 --- a/spring-batch-samples/.springBeans +++ b/spring-batch-samples/.springBeans @@ -48,6 +48,26 @@ src/main/resources/jobs/iosample/jpa.xml src/main/resources/jobs/iosample/ibatis.xml src/test/resources/org/springframework/batch/sample/JobOperatorFunctionalTests-context.xml + src/test/resources/org/springframework/batch/sample/BeanWrapperMapperSampleJobFunctionalTests-context.xml + src/test/resources/org/springframework/batch/sample/CompositeItemWriterSampleFunctionalTests-context.xml + src/test/resources/org/springframework/batch/sample/CustomerFilterJobFunctionalTests-context.xml + src/test/resources/org/springframework/batch/sample/DatabaseShutdownFunctionalTests-context.xml + src/test/resources/org/springframework/batch/sample/DelegatingJobFunctionalTests-context.xml + src/test/resources/org/springframework/batch/sample/FootballJobFunctionalTests-context.xml + src/test/resources/org/springframework/batch/sample/GracefulShutdownFunctionalTests-context.xml + src/test/resources/org/springframework/batch/sample/HeaderFooterSampleFunctionalTests-context.xml + src/test/resources/org/springframework/batch/sample/HibernateFailureJobFunctionalTests-context.xml + src/test/resources/org/springframework/batch/sample/domain/trade/internal/JdbcCustomerDebitDaoTests-context.xml + src/main/resources/jobs/loopFlowSample.xml + src/test/resources/org/springframework/batch/sample/MultilineJobFunctionalTests-context.xml + src/test/resources/org/springframework/batch/sample/MultilineOrderJobFunctionalTests-context.xml + src/test/resources/org/springframework/batch/sample/ParallelJobFunctionalTests-context.xml + src/main/resources/jobs/partitionJob.xml + src/test/resources/org/springframework/batch/sample/RestartFunctionalTests-context.xml + src/test/resources/org/springframework/batch/sample/RetrySampleFunctionalTests-context.xml + src/test/resources/org/springframework/batch/sample/common/StagingItemReaderTests-context.xml + src/test/resources/org/springframework/batch/sample/common/StagingItemWriterTests-context.xml + src/test/resources/org/springframework/batch/sample/TradeJobFunctionalTests-context.xml diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/OutputFileListener.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/OutputFileListener.java new file mode 100644 index 000000000..a04e154cf --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/OutputFileListener.java @@ -0,0 +1,50 @@ +/* + * 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.sample.common; + +import org.apache.commons.io.FilenameUtils; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.annotation.BeforeStep; +import org.springframework.batch.item.ExecutionContext; + +/** + * @author Dave Syer + * + */ +public class OutputFileListener { + + private String outputKeyName = "outputFile"; + + private String inputKeyName = "fileName"; + + public void setOutputKeyName(String outputKeyName) { + this.outputKeyName = outputKeyName; + } + + public void setInputKeyName(String inputKeyName) { + this.inputKeyName = inputKeyName; + } + + @BeforeStep + public void CreateOutputNameFromInput(StepExecution stepExecution) { + ExecutionContext executionContext = stepExecution.getExecutionContext(); + if (executionContext.containsKey(inputKeyName) && !executionContext.containsKey(outputKeyName)) { + String inputName = executionContext.getString(inputKeyName); + executionContext.putString(outputKeyName, "file:./target/output/" + FilenameUtils.getBaseName(inputName) + ".csv"); + } + } + +} diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/OutputFileNameListener.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/OutputFileNameListener.java new file mode 100644 index 000000000..64c0c4054 --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/OutputFileNameListener.java @@ -0,0 +1,5 @@ +package org.springframework.batch.sample.common; + +public class OutputFileNameListener { + +} diff --git a/spring-batch-samples/src/main/resources/jobs/partitionJob.xml b/spring-batch-samples/src/main/resources/jobs/partitionJob.xml new file mode 100644 index 000000000..d75f4fe59 --- /dev/null +++ b/spring-batch-samples/src/main/resources/jobs/partitionJob.xml @@ -0,0 +1,98 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/PartitionJobFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/PartitionJobFunctionalTests.java new file mode 100644 index 000000000..dec80d69a --- /dev/null +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/PartitionJobFunctionalTests.java @@ -0,0 +1,114 @@ +/* + * 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.sample; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.List; + +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.item.ExecutionContext; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.sample.domain.trade.CustomerCredit; +import org.springframework.batch.sample.domain.trade.internal.CustomerCreditIncreaseProcessor; +import org.springframework.batch.test.AbstractJobTests; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration() +public class PartitionJobFunctionalTests extends AbstractJobTests { + + @Autowired + @Qualifier("inputTestReader") + private ItemReader inputReader; + + /** + * Check the resulting credits correspond to inputs increased by fixed + * amount. + */ + @Test + public void testUpdateCredit() throws Exception { + + assertTrue("Define a prototype bean called 'outputTestReader' to check the output", getApplicationContext() + .containsBeanDefinition("outputTestReader")); + + open(inputReader); + List inputs = getCredits(inputReader); + close(inputReader); + + JobExecution jobExecution = this.launchJob(); + assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); + + @SuppressWarnings("unchecked") + ItemReader outputReader = (ItemReader) getApplicationContext().getBean( + "outputTestReader"); + open(outputReader); + List outputs = getCredits(outputReader); + close(outputReader); + + assertEquals(inputs.size(), outputs.size()); + int itemCount = inputs.size(); + assertTrue(itemCount > 0); + + for (int i = 0; i < itemCount; i++) { + assertEquals(inputs.get(i).getCredit().add(CustomerCreditIncreaseProcessor.FIXED_AMOUNT).intValue(), + outputs.get(i).getCredit().intValue()); + } + + } + + /** + * Read all credits using the provided reader. + */ + private List getCredits(ItemReader reader) throws Exception { + CustomerCredit credit; + List result = new ArrayList(); + while ((credit = reader.read()) != null) { + result.add(credit); + } + return result; + + } + + /** + * Open the reader if applicable. + */ + private void open(ItemReader reader) { + if (reader instanceof ItemStream) { + ((ItemStream) reader).open(new ExecutionContext()); + } + } + + /** + * Close the reader if applicable. + */ + private void close(ItemReader reader) { + if (reader instanceof ItemStream) { + ((ItemStream) reader).close(); + } + } + +} diff --git a/spring-batch-samples/src/test/resources/org/springframework/batch/sample/PartitionJobFunctionalTests-context.xml b/spring-batch-samples/src/test/resources/org/springframework/batch/sample/PartitionJobFunctionalTests-context.xml new file mode 100644 index 000000000..44dbf80b8 --- /dev/null +++ b/spring-batch-samples/src/test/resources/org/springframework/batch/sample/PartitionJobFunctionalTests-context.xml @@ -0,0 +1,10 @@ + + + + + + + diff --git a/spring-batch-test/src/main/java/org/springframework/batch/test/AbstractJobTests.java b/spring-batch-test/src/main/java/org/springframework/batch/test/AbstractJobTests.java index 881e24b21..691d829f2 100644 --- a/spring-batch-test/src/main/java/org/springframework/batch/test/AbstractJobTests.java +++ b/spring-batch-test/src/main/java/org/springframework/batch/test/AbstractJobTests.java @@ -32,8 +32,10 @@ import org.springframework.batch.core.job.SimpleJob; import org.springframework.batch.core.job.flow.FlowJob; import org.springframework.batch.core.launch.JobLauncher; import org.springframework.batch.core.repository.JobRepository; +import org.springframework.beans.BeansException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; /** *

@@ -62,7 +64,7 @@ import org.springframework.context.ApplicationContext; * @author Dan Garrette * @since 2.0 */ -public abstract class AbstractJobTests { +public abstract class AbstractJobTests implements ApplicationContextAware { /** Logger */ protected final Log logger = LogFactory.getLog(getClass()); @@ -78,6 +80,23 @@ public abstract class AbstractJobTests { private StepRunner stepRunner; + private ApplicationContext applicationContext; + + /** + * {@inheritDoc} + */ + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; + } + + /** + * @return the applicationContext + */ + protected ApplicationContext getApplicationContext() { + return applicationContext; + } + /** * @return the job repository which is autowired by type */