ONly increment the item count for commit interval purposes when something is written.

This commit is contained in:
dsyer
2008-08-11 21:03:06 +00:00
parent 6d15a371ec
commit b5fd803497
3 changed files with 137 additions and 16 deletions

View File

@@ -54,8 +54,8 @@ public class ItemOrientedStepHandler<T, S> implements ItemHandler {
* @param itemProcessor
* @param itemWriter
*/
public ItemOrientedStepHandler(ItemReader<? extends T> itemReader, ItemProcessor<? super T, ? extends S> itemProcessor,
ItemWriter<? super S> itemWriter) {
public ItemOrientedStepHandler(ItemReader<? extends T> itemReader,
ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter) {
super();
this.itemReader = itemReader;
this.itemProcessor = itemProcessor;
@@ -64,17 +64,23 @@ public class ItemOrientedStepHandler<T, S> implements ItemHandler {
/**
* Get the next item from {@link #read(StepContribution)} and if not null
* pass the item to {@link #write(Object, StepContribution)}.
* pass the item to {@link #write(Object, StepContribution)}. If the
* {@link ItemProcessor} returns null, the write is omitted and another
* item taken from the reader.
*
* @see org.springframework.batch.core.step.item.ItemHandler#handle(org.springframework.batch.core.StepContribution)
*/
public ExitStatus handle(StepContribution contribution) throws Exception {
T item = read(contribution);
if (item == null) {
return ExitStatus.FINISHED;
boolean processed = false;
while (!processed) {
T item = read(contribution);
if (item == null) {
return ExitStatus.FINISHED;
}
// TODO: segregate read / write / filter count
contribution.incrementItemCount();
processed = write(item, contribution);
}
contribution.incrementItemCount();
write(item, contribution);
return ExitStatus.CONTINUABLE;
}
@@ -98,21 +104,24 @@ public class ItemOrientedStepHandler<T, S> implements ItemHandler {
*
* @param item the item to write
* @param contribution current context
* @return true if the item was written (as opposed to filtered)
*/
protected void write(T item, StepContribution contribution) throws Exception {
doWrite(item);
protected boolean write(T item, StepContribution contribution) throws Exception {
return doWrite(item);
}
/**
* @param item
* @throws Exception
*/
protected final void doWrite(T item) throws Exception {
protected final boolean doWrite(T item) throws Exception {
S processed = itemProcessor.process(item);
if (processed != null) {
// TODO: increment filtered item count
itemWriter.write(processed);
return true;
}
return false;
}
/**

View File

@@ -413,11 +413,10 @@ public class SkipLimitStepFactoryBean<T,S> extends SimpleStepFactoryBean<T,S> {
* @see org.springframework.batch.core.step.item.SimpleItemHandler#write(java.lang.Object,
* org.springframework.batch.core.StepContribution)
*/
protected void write(final T item, final StepContribution contribution) throws Exception {
protected boolean write(final T item, final StepContribution contribution) throws Exception {
RecoveryRetryCallback retryCallback = new RecoveryRetryCallback(item, new RetryCallback() {
public Object doWithRetry(RetryContext context) throws Throwable {
doWrite(item);
return null;
return doWrite(item);
}
}, itemKeyGenerator != null ? itemKeyGenerator.getKey(item) : item);
retryCallback.setRecoveryCallback(new RecoveryCallback() {
@@ -432,10 +431,10 @@ public class SkipLimitStepFactoryBean<T,S> extends SimpleStepFactoryBean<T,S> {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, t);
}
}
return null;
return true;
}
});
retryOperations.execute(retryCallback);
return (Boolean) retryOperations.execute(retryCallback);
}
}

View File

@@ -0,0 +1,113 @@
/*
* 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.step.item;
import static org.junit.Assert.assertEquals;
import org.junit.Test;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.NoWorkFoundException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.item.support.AbstractItemReader;
import org.springframework.batch.item.support.AbstractItemWriter;
import org.springframework.batch.item.support.PassthroughItemProcessor;
/**
* @author Dave Syer
*
*/
public class ItemOrientedStepHandlerTests {
private StubItemReader itemReader = new StubItemReader();
private StubItemWriter itemWriter = new StubItemWriter();
@Test
public void testHandle() throws Exception {
ItemOrientedStepHandler<String, String> handler = new ItemOrientedStepHandler<String, String>(itemReader,
new PassthroughItemProcessor<String>(), itemWriter);
StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(new JobInstance(
123L, new JobParameters(), "job"))));
handler.handle(contribution);
assertEquals(1, itemReader.count);
assertEquals("1", itemWriter.values);
}
@Test
public void testHandleCompositeItem() throws Exception {
ItemOrientedStepHandler<String, String> handler = new ItemOrientedStepHandler<String, String>(itemReader,
new AgrgegateItemProcessor(), itemWriter);
StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(new JobInstance(
123L, new JobParameters(), "job"))));
handler.handle(contribution);
assertEquals(2, itemReader.count);
assertEquals("12", itemWriter.values);
}
/**
* @author Dave Syer
*
*/
private final class AgrgegateItemProcessor implements ItemProcessor<String, String> {
private int count = 0;
private String value = "";
public String process(String item) throws Exception {
value += item;
if (count++ < 1) {
return null;
}
String result = value;
value = "";
count = 0;
return result;
}
}
/**
* @author Dave Syer
*
*/
private final class StubItemWriter extends AbstractItemWriter<String> {
private String values = "";
public void write(String item) throws Exception {
values += item;
}
}
/**
* @author Dave Syer
*
*/
private final class StubItemReader extends AbstractItemReader<String> {
private int count = 0;
public String read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException {
if (count++ < 5)
return "" + count;
return null;
}
}
}