Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add an ItemProcessor that delegates to a java.util.function.Function

Browse Source 
This commit adds a new `ItemProcessor` that delegates to a
`java.util.function.Function`.  It also enables users to configure a
`Function` as an `ItemProcessor` via the builders (so they don't need to
worry about the `ItemProcessor` wrapper in the first place).

Resolves BATCH-2641
This commit is contained in:
Michael Minella
2017-09-14 16:07:11 -05:00
parent b180c75ce0
commit 7d2a26a136
5 changed files with 172 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java
View File

@@ -185,7 +185,7 @@ public class FaultTolerantStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
registerSkipListeners();
ChunkProvider<I> chunkProvider = createChunkProvider();
ChunkProcessor<I> chunkProcessor = createChunkProcessor();
ChunkOrientedTasklet<I> tasklet = new ChunkOrientedTasklet<I>(chunkProvider, chunkProcessor);
ChunkOrientedTasklet<I> tasklet = new ChunkOrientedTasklet<>(chunkProvider, chunkProcessor);
tasklet.setBuffering(!isReaderTransactionalQueue());
return tasklet;
}
@@ -452,7 +452,7 @@ public class FaultTolerantStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
chunkProvider.setMaxSkipsOnRead(Math.max(getChunkSize(), FaultTolerantChunkProvider.DEFAULT_MAX_SKIPS_ON_READ));
chunkProvider.setSkipPolicy(readSkipPolicy);
chunkProvider.setRollbackClassifier(getRollbackClassifier());
ArrayList<StepListener> listeners = new ArrayList<StepListener>(getItemListeners());
ArrayList<StepListener> listeners = new ArrayList<>(getItemListeners());
listeners.addAll(skipListeners);
chunkProvider.setListeners(listeners);
@@ -464,7 +464,7 @@ public class FaultTolerantStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
BatchRetryTemplate batchRetryTemplate = createRetryOperations();
FaultTolerantChunkProcessor<I, O> chunkProcessor = new FaultTolerantChunkProcessor<I, O>(getProcessor(),
FaultTolerantChunkProcessor<I, O> chunkProcessor = new FaultTolerantChunkProcessor<>(getProcessor(),
getWriter(), batchRetryTemplate);
chunkProcessor.setBuffering(!isReaderTransactionalQueue());
chunkProcessor.setProcessorTransactional(processorTransactional);

35
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java
View File

@@ -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<I, O> extends AbstractTaskletStepBuilder<SimpleSt
private ItemProcessor<? super I, ? extends O> processor;
private Function<? super I, ? extends O> itemProcessorFunction;
private int chunkSize = 0;
private RepeatOperations chunkOperations;
private CompletionPolicy completionPolicy;
private Set<StepListener> itemListeners = new LinkedHashSet<StepListener>();
private Set<StepListener> itemListeners = new LinkedHashSet<>();
private boolean readerTransactionalQueue = false;
@@ -111,7 +115,7 @@ public class SimpleStepBuilder<I, O> extends AbstractTaskletStepBuilder<SimpleSt
}
public FaultTolerantStepBuilder<I, O> faultTolerant() {
return new FaultTolerantStepBuilder<I, O>(this);
return new FaultTolerantStepBuilder<>(this);
}
/**
@@ -154,11 +158,11 @@ public class SimpleStepBuilder<I, O> extends AbstractTaskletStepBuilder<SimpleSt
Assert.state(reader != null, "ItemReader must be provided");
Assert.state(processor != null || writer != null, "ItemWriter or ItemProcessor must be provided");
RepeatOperations repeatOperations = createChunkOperations();
SimpleChunkProvider<I> chunkProvider = new SimpleChunkProvider<I>(reader, repeatOperations);
SimpleChunkProcessor<I, O> chunkProcessor = new SimpleChunkProcessor<I, O>(processor, writer);
chunkProvider.setListeners(new ArrayList<StepListener>(itemListeners));
chunkProcessor.setListeners(new ArrayList<StepListener>(itemListeners));
ChunkOrientedTasklet<I> tasklet = new ChunkOrientedTasklet<I>(chunkProvider, chunkProcessor);
SimpleChunkProvider<I> chunkProvider = new SimpleChunkProvider<>(getReader(), repeatOperations);
SimpleChunkProcessor<I, O> chunkProcessor = new SimpleChunkProcessor<>(getProcessor(), getWriter());
chunkProvider.setListeners(new ArrayList<>(itemListeners));
chunkProcessor.setListeners(new ArrayList<>(itemListeners));
ChunkOrientedTasklet<I> tasklet = new ChunkOrientedTasklet<>(chunkProvider, chunkProcessor);
tasklet.setBuffering(!readerTransactionalQueue);
return tasklet;
}
@@ -229,6 +233,19 @@ public class SimpleStepBuilder<I, O> extends AbstractTaskletStepBuilder<SimpleSt
return this;
}
/**
* A {@link Function} to be delegated to as an {@link ItemProcessor}. If this is set,
* it will take precidence over any {@code ItemProcessor} configured via
* {@link #processor(ItemProcessor)}.
*
* @param function the function to delegate item processing to
* @return this for fluent chaining
*/
public SimpleStepBuilder<I, O> processor(Function<? super I, ? extends O> 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<I, O> extends AbstractTaskletStepBuilder<SimpleSt
}
protected ItemProcessor<? super I, ? extends O> getProcessor() {
if(this.itemProcessorFunction != null) {
this.processor = new FunctionItemProcessor<>(this.itemProcessorFunction);
}
return processor;
}

57
spring-batch-core/src/test/java/org/springframework/batch/core/step/builder/StepBuilderTests.java
View File

@@ -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<Long> items = new ArrayList<Long>() {{
add(1L);
add(2L);
add(3L);
}};
ItemReader<Long> reader = new ListItemReader<>(items);
ListItemWriter<String> itemWriter = new ListItemWriter<>();
@SuppressWarnings("unchecked")
SimpleStepBuilder<Object, String> builder = new StepBuilder("step")
.repository(jobRepository)
.transactionManager(transactionManager)
.<Object, String>chunk(3)
.reader(reader)
.processor((Function<Object, String>) s -> s.toString())
.writer(itemWriter)
.listener(new AnnotationBasedStepExecutionListener());
builder.build().execute(execution);
assertEquals(BatchStatus.COMPLETED, execution.getStatus());
List<? extends String> 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;

45
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/function/FunctionItemProcessor.java Normal file
View File

@@ -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<I, O> implements ItemProcessor<I, O>{
private final Function<I, O> function;
/**
* @param function the delegate. Must not be null
*/
public FunctionItemProcessor(Function<I, O> function) {
Assert.notNull(function, "A function is required");
this.function = function;
}
@Override
public O process(I item) throws Exception {
return this.function.apply(item);
}
}

57
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/function/FunctionItemProcessorTests.java Normal file
View File

@@ -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<Object, String> 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<Object, String> itemProcessor =
new FunctionItemProcessor<>(this.function);
assertEquals("1", itemProcessor.process(1L));
assertEquals("foo", itemProcessor.process("foo"));
}
}