diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java index e2b6d727a..f56ae1559 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java @@ -185,7 +185,7 @@ public class FaultTolerantStepBuilder extends SimpleStepBuilder { registerSkipListeners(); ChunkProvider chunkProvider = createChunkProvider(); ChunkProcessor chunkProcessor = createChunkProcessor(); - ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet(chunkProvider, chunkProcessor); + ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet<>(chunkProvider, chunkProcessor); tasklet.setBuffering(!isReaderTransactionalQueue()); return tasklet; } @@ -452,7 +452,7 @@ public class FaultTolerantStepBuilder extends SimpleStepBuilder { chunkProvider.setMaxSkipsOnRead(Math.max(getChunkSize(), FaultTolerantChunkProvider.DEFAULT_MAX_SKIPS_ON_READ)); chunkProvider.setSkipPolicy(readSkipPolicy); chunkProvider.setRollbackClassifier(getRollbackClassifier()); - ArrayList listeners = new ArrayList(getItemListeners()); + ArrayList listeners = new ArrayList<>(getItemListeners()); listeners.addAll(skipListeners); chunkProvider.setListeners(listeners); @@ -464,7 +464,7 @@ public class FaultTolerantStepBuilder extends SimpleStepBuilder { BatchRetryTemplate batchRetryTemplate = createRetryOperations(); - FaultTolerantChunkProcessor chunkProcessor = new FaultTolerantChunkProcessor(getProcessor(), + FaultTolerantChunkProcessor chunkProcessor = new FaultTolerantChunkProcessor<>(getProcessor(), getWriter(), batchRetryTemplate); chunkProcessor.setBuffering(!isReaderTransactionalQueue()); chunkProcessor.setProcessorTransactional(processorTransactional); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java index 0892e9ec1..f25f454d9 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java @@ -20,6 +20,7 @@ import java.util.ArrayList; import java.util.HashSet; import java.util.LinkedHashSet; import java.util.Set; +import java.util.function.Function; import org.springframework.batch.core.ChunkListener; import org.springframework.batch.core.ItemProcessListener; @@ -46,6 +47,7 @@ import org.springframework.batch.item.ItemProcessor; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.function.FunctionItemProcessor; import org.springframework.batch.repeat.CompletionPolicy; import org.springframework.batch.repeat.RepeatOperations; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; @@ -74,13 +76,15 @@ public class SimpleStepBuilder extends AbstractTaskletStepBuilder processor; + private Function itemProcessorFunction; + private int chunkSize = 0; private RepeatOperations chunkOperations; private CompletionPolicy completionPolicy; - private Set itemListeners = new LinkedHashSet(); + private Set itemListeners = new LinkedHashSet<>(); private boolean readerTransactionalQueue = false; @@ -111,7 +115,7 @@ public class SimpleStepBuilder extends AbstractTaskletStepBuilder faultTolerant() { - return new FaultTolerantStepBuilder(this); + return new FaultTolerantStepBuilder<>(this); } /** @@ -154,11 +158,11 @@ public class SimpleStepBuilder extends AbstractTaskletStepBuilder chunkProvider = new SimpleChunkProvider(reader, repeatOperations); - SimpleChunkProcessor chunkProcessor = new SimpleChunkProcessor(processor, writer); - chunkProvider.setListeners(new ArrayList(itemListeners)); - chunkProcessor.setListeners(new ArrayList(itemListeners)); - ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet(chunkProvider, chunkProcessor); + SimpleChunkProvider chunkProvider = new SimpleChunkProvider<>(getReader(), repeatOperations); + SimpleChunkProcessor chunkProcessor = new SimpleChunkProcessor<>(getProcessor(), getWriter()); + chunkProvider.setListeners(new ArrayList<>(itemListeners)); + chunkProcessor.setListeners(new ArrayList<>(itemListeners)); + ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet<>(chunkProvider, chunkProcessor); tasklet.setBuffering(!readerTransactionalQueue); return tasklet; } @@ -229,6 +233,19 @@ public class SimpleStepBuilder extends AbstractTaskletStepBuilder processor(Function function) { + this.itemProcessorFunction = function; + return this; + } + /** * Sets a flag to say that the reader is transactional (usually a queue), which is to say that failed items might be * rolled back and re-presented in a subsequent transaction. Default is false, meaning that the items are read @@ -339,6 +356,10 @@ public class SimpleStepBuilder extends AbstractTaskletStepBuilder getProcessor() { + if(this.itemProcessorFunction != null) { + this.processor = new FunctionItemProcessor<>(this.itemProcessorFunction); + } + return processor; } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/builder/StepBuilderTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/builder/StepBuilderTests.java index 37ba8dba5..1cf6e2fb6 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/builder/StepBuilderTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/builder/StepBuilderTests.java @@ -17,13 +17,13 @@ package org.springframework.batch.core.step.builder; import java.util.ArrayList; import java.util.List; +import java.util.function.Function; import org.junit.Test; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.JobParameters; -import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.StepExecutionListener; import org.springframework.batch.core.annotation.AfterChunk; @@ -39,12 +39,10 @@ import org.springframework.batch.core.annotation.BeforeWrite; import org.springframework.batch.core.configuration.xml.DummyItemWriter; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; -import org.springframework.batch.core.scope.context.ChunkContext; -import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.support.ListItemReader; +import org.springframework.batch.item.support.ListItemWriter; import org.springframework.batch.item.support.PassThroughItemProcessor; -import org.springframework.batch.repeat.RepeatStatus; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.transaction.PlatformTransactionManager; @@ -66,13 +64,7 @@ public class StepBuilderTests { jobRepository.add(execution); PlatformTransactionManager transactionManager = new ResourcelessTransactionManager(); TaskletStepBuilder builder = new StepBuilder("step").repository(jobRepository) - .transactionManager(transactionManager).tasklet(new Tasklet() { - @Override - public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) - throws Exception { - return null; - } - }); + .transactionManager(transactionManager).tasklet((contribution, chunkContext) -> null); builder.build().execute(execution); assertEquals(BatchStatus.COMPLETED, execution.getStatus()); } @@ -88,13 +80,7 @@ public class StepBuilderTests { .transactionManager(transactionManager) .listener(new InterfaceBasedStepExecutionListener()) .listener(new AnnotationBasedStepExecutionListener()) - .tasklet(new Tasklet() { - @Override - public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) - throws Exception { - return null; - } - }); + .tasklet((contribution, chunkContext) -> null); builder.build().execute(execution); assertEquals(BatchStatus.COMPLETED, execution.getStatus()); assertEquals(1, InterfaceBasedStepExecutionListener.beforeStepCount); @@ -142,6 +128,41 @@ public class StepBuilderTests { assertEquals(2, AnnotationBasedStepExecutionListener.afterChunkCount); } + @Test + public void testFunctions() throws Exception { + JobRepository jobRepository = new MapJobRepositoryFactoryBean().getObject(); + StepExecution execution = jobRepository.createJobExecution("foo", new JobParameters()).createStepExecution("step"); + jobRepository.add(execution); + PlatformTransactionManager transactionManager = new ResourcelessTransactionManager(); + + List items = new ArrayList() {{ + add(1L); + add(2L); + add(3L); + }}; + + ItemReader reader = new ListItemReader<>(items); + + ListItemWriter itemWriter = new ListItemWriter<>(); + @SuppressWarnings("unchecked") + SimpleStepBuilder builder = new StepBuilder("step") + .repository(jobRepository) + .transactionManager(transactionManager) + .chunk(3) + .reader(reader) + .processor((Function) s -> s.toString()) + .writer(itemWriter) + .listener(new AnnotationBasedStepExecutionListener()); + builder.build().execute(execution); + + assertEquals(BatchStatus.COMPLETED, execution.getStatus()); + + List writtenItems = itemWriter.getWrittenItems(); + assertEquals("1", writtenItems.get(0)); + assertEquals("2", writtenItems.get(1)); + assertEquals("3", writtenItems.get(2)); + } + public static class InterfaceBasedStepExecutionListener implements StepExecutionListener { static int beforeStepCount = 0; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/function/FunctionItemProcessor.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/function/FunctionItemProcessor.java new file mode 100644 index 000000000..fccb3f58d --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/function/FunctionItemProcessor.java @@ -0,0 +1,45 @@ +/* + * Copyright 2017 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.item.function; + +import java.util.function.Function; + +import org.springframework.batch.item.ItemProcessor; +import org.springframework.util.Assert; + +/** + * An {@link ItemProcessor} implementation that delegates to a {@link Function} + * + * @author Michael Minella + * @since 4.0 + */ +public class FunctionItemProcessor implements ItemProcessor{ + + private final Function function; + + /** + * @param function the delegate. Must not be null + */ + public FunctionItemProcessor(Function function) { + Assert.notNull(function, "A function is required"); + this.function = function; + } + + @Override + public O process(I item) throws Exception { + return this.function.apply(item); + } +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/function/FunctionItemProcessorTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/function/FunctionItemProcessorTests.java new file mode 100644 index 000000000..bb65daa65 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/function/FunctionItemProcessorTests.java @@ -0,0 +1,57 @@ +/* + * Copyright 2017 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.item.function; + +import java.util.function.Function; + +import org.junit.Before; +import org.junit.Test; + +import org.springframework.batch.item.ItemProcessor; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +/** + * @author Michael Minella + */ +public class FunctionItemProcessorTests { + + private Function function; + + @Before + public void setUp() { + this.function = o -> o.toString(); + } + + @Test + public void testConstructorValidation() { + try { + new FunctionItemProcessor<>(null); + fail("null should not be accepted as a constructor arg"); + } + catch (IllegalArgumentException iae) {} + } + + @Test + public void testFunctionItemProcessor() throws Exception { + ItemProcessor itemProcessor = + new FunctionItemProcessor<>(this.function); + + assertEquals("1", itemProcessor.process(1L)); + assertEquals("foo", itemProcessor.process("foo")); + } +}