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

RESOLVED BATCH-1218: add CompositeItemStream to ChunkMonitor.

Browse Source
This commit is contained in:
dsyer
2009-05-12 10:42:02 +00:00
parent 0bd6a8fc17
commit b569b8419f
3 changed files with 8 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java
View File

@@ -21,6 +21,7 @@ import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.support.CompositeItemStream;
/**
* Manage the offset data between the last successful commit and updates made to
@@ -48,7 +49,7 @@ class ChunkMonitor implements ItemStream {
private static final String OFFSET = ChunkMonitor.class.getName() + ".OFFSET";
private ItemStream stream;
private CompositeItemStream stream = new CompositeItemStream();
private ThreadLocal<ChunkMonitorData> holder = new ThreadLocal<ChunkMonitorData>();
{
@@ -62,8 +63,8 @@ class ChunkMonitor implements ItemStream {
/**
* @param stream the stream to set
*/
public void setItemStream(ItemStream stream) {
this.stream = stream;
public void registerItemStream(ItemStream stream) {
this.stream.register(stream);
}
/**

8
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java
View File

@@ -33,7 +33,6 @@ import org.springframework.batch.core.step.skip.SkipPolicy;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.support.CompositeItemStream;
import org.springframework.batch.repeat.RepeatOperations;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.batch.retry.RetryException;
@@ -260,13 +259,12 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
*/
@Override
protected void registerStreams(TaskletStep step, ItemStream[] streams) {
CompositeItemStream composite = new CompositeItemStream();
boolean streamIsReader = false;
ItemReader<? extends T> itemReader = getItemReader();
for (ItemStream stream : streams) {
if (stream == itemReader) {
if (stream instanceof ItemReader) {
streamIsReader = true;
composite.register(stream);
chunkMonitor.registerItemStream(stream);
}
else {
step.registerStream(stream);
@@ -276,7 +274,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
// In cases where multiple nested item readers are registered,
// they all want to get the open() and close() callbacks.
if (streamIsReader) {
chunkMonitor.setItemStream(composite);
// double registration is fine
step.registerStream(chunkMonitor);
boolean concurrent = taskExecutor != null && !(taskExecutor instanceof SyncTaskExecutor);
if (!concurrent) {

2
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkMonitorTests.java
View File

@@ -51,7 +51,7 @@ public class ChunkMonitorTests {
return "" + (count++);
}
});
monitor.setItemStream(new ItemStreamSupport() {
monitor.registerItemStream(new ItemStreamSupport() {
@Override
public void close() throws ItemStreamException {
closed = true;