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

Half-hearted attempt to fix broken integration tests

Browse Source
This commit is contained in:
dsyer
2008-09-02 15:45:11 +00:00
parent b79a496626
commit 40b65bafb2
14 changed files with 233 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkHandler.java
View File

@@ -1,8 +1,7 @@
package org.springframework.batch.integration.chunk;
public interface ChunkHandler<T> {
ChunkResponse handleChunk(ChunkRequest<? extends T> chunk);
ChunkResponse handleChunk(ChunkRequest<T> chunk);
}

32
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessor.java Normal file
View File

@@ -0,0 +1,32 @@
/*
* 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.integration.chunk;
import java.util.Collection;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
/**
* @author Dave Syer
*/
public interface ChunkProcessor<S> {
// This is transactional with REQUIRES_NEW because we need to force rollback
@Transactional(propagation = Propagation.REQUIRES_NEW)
int process(Collection<? extends S> items, int skipCount) throws Exception;
}

54
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java Normal file
View File

@@ -0,0 +1,54 @@
package org.springframework.batch.integration.chunk;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.annotation.Handler;
import org.springframework.util.Assert;
public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, InitializingBean {
private static final Log logger = LogFactory.getLog(ChunkProcessorChunkHandler.class);
private ChunkProcessor<S> chunkProcessor;
/* (non-Javadoc)
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
public void afterPropertiesSet() throws Exception {
Assert.notNull(chunkProcessor, "A ChunkProcessor must be provided");
}
/**
* Public setter for the {@link ChunkProcessor}.
* @param chunkProcessor the chunkProcessor to set
*/
public void setChunkProcessor(ChunkProcessor<S> chunkProcessor) {
this.chunkProcessor = chunkProcessor;
}
/*
* (non-Javadoc)
* @see org.springframework.integration.batch.slave.ChunkHandler#handleChunk(java.util.Collection)
*/
@Handler
public ChunkResponse handleChunk(ChunkRequest<S> chunkRequest) {
logger.debug("Handling chunk: " + chunkRequest);
int skipCount = 0;
try {
skipCount = chunkProcessor.process(chunkRequest.getItems(), chunkRequest.getSkipCount());
}
catch (Exception e) {
logger.debug("Failed chunk", e);
return new ChunkResponse(ExitStatus.FAILED.addExitDescription(e.getClass().getName() + ": "
+ e.getMessage()), chunkRequest.getJobId(), skipCount);
}
logger.debug("Completed chunk handling with " + skipCount + " skips");
return new ChunkResponse(ExitStatus.CONTINUABLE, chunkRequest.getJobId(), skipCount);
}
}

85
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ItemWriterChunkHandler.java
View File

@@ -1,85 +0,0 @@
package org.springframework.batch.integration.chunk;
import java.util.Collections;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.listener.CompositeSkipListener;
import org.springframework.batch.core.step.skip.ItemSkipPolicy;
import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.integration.annotation.Handler;
import org.springframework.transaction.annotation.Transactional;
public class ItemWriterChunkHandler<T> implements ChunkHandler<T> {
private static final Log logger = LogFactory.getLog(ItemWriterChunkHandler.class);
private ItemWriter<? super T> itemWriter;
private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy();
private CompositeSkipListener skipListener = new CompositeSkipListener();
public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) {
this.itemSkipPolicy = itemSkipPolicy;
}
public void setItemWriter(ItemWriter<? super T> itemWriter) {
this.itemWriter = itemWriter;
}
public void registerSkipListener(SkipListener listener) {
skipListener.register(listener);
}
public void setSkipListeners(SkipListener[] skipListeners) {
for (SkipListener listener : skipListeners) {
registerSkipListener(listener);
}
}
/*
* (non-Javadoc)
* @see org.springframework.integration.batch.slave.ChunkHandler#handleChunk(java.util.Collection)
*/
@Handler
@Transactional
public ChunkResponse handleChunk(ChunkRequest<? extends T> chunk) {
logger.debug("Handling chunk: " + chunk);
int parentSkipCount = chunk.getSkipCount();
int skipCount = 0;
try {
for (T item : chunk.getItems()) {
try {
itemWriter.write(Collections.singletonList(item));
}
catch (Exception e) {
if (itemSkipPolicy.shouldSkip(e, parentSkipCount + skipCount)) {
logger.debug("Skipping item on exception", e);
skipCount++;
skipListener.onSkipInWrite(item, e);
} else {
logger.debug("Cannot skip, re-throwing");
throw e;
}
}
}
}
catch (Exception e) {
logger.debug("Failed chunk", e);
// TODO: need to force rollback as well
return new ChunkResponse(ExitStatus.FAILED.addExitDescription(e.getClass().getName() + ": "
+ e.getMessage()), chunk.getJobId(), skipCount);
}
logger.debug("Completed chunk handling with " + skipCount + " skips");
return new ChunkResponse(ExitStatus.CONTINUABLE, chunk.getJobId(), skipCount);
}
}

78
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/SimpleChunkProcessor.java Normal file
View File

@@ -0,0 +1,78 @@
/*
* 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.integration.chunk;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
/**
* @author Dave Syer
*
*/
public class SimpleChunkProcessor<S, T> implements ChunkProcessor<S>, InitializingBean {
private ItemProcessor<? super S, ? extends T> itemProcessor;
private ItemWriter<? super T> itemWriter;
/*
* (non-Javadoc)
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
public void afterPropertiesSet() throws Exception {
Assert.notNull(itemProcessor, "An ItemProcessor must be provided");
Assert.notNull(itemWriter, "An ItemWriter must be provided");
}
/**
* @param itemWriter
*/
public void setItemWriter(ItemWriter<? super T> itemWriter) {
this.itemWriter = itemWriter;
}
/**
* Public setter for the {@link ItemProcessor}.
* @param itemProcessor the {@link ItemProcessor} to set
*/
public void setItemProcessor(ItemProcessor<? super S, ? extends T> itemProcessor) {
this.itemProcessor = itemProcessor;
}
/*
* (non-Javadoc)
* @see org.springframework.batch.integration.chunk.ChunkProcessor#process(java.util.Collection,
* int)
*/
public int process(Collection<? extends S> items, int parentSkipCount) throws Exception {
List<T> processed = new ArrayList<T>();
for (S item : items) {
processed.add(itemProcessor.process(item));
}
itemWriter.write(processed);
return 0;
}
}