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

Incomplete - task 87: Refactor KeyedItemReader

Browse Source 
Split off from ItemReader interface.
This commit is contained in:
dsyer
2008-03-01 15:24:06 +00:00
parent 886e8b4f75
commit b89a68dfeb
18 changed files with 220 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java
View File

@@ -30,8 +30,9 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.io.Skippable;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.KeyedItemReader;
import org.springframework.batch.item.exception.ResetFailedException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.dao.DataAccessException;
@@ -107,7 +108,7 @@ import org.springframework.util.StringUtils;
* @author Lucas Ward
* @author Peter Zozom
*/
public class JdbcCursorItemReader implements KeyedItemReader, InitializingBean,
public class JdbcCursorItemReader implements ItemReader, ItemKeyGenerator, InitializingBean,
ItemStream, Skippable {
private static Log log = LogFactory.getLog(JdbcCursorItemReader.class);

5
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryItemReader.java
View File

@@ -19,8 +19,9 @@ import java.util.Iterator;
import java.util.List;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.KeyedItemReader;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
@@ -46,7 +47,7 @@ import org.springframework.util.Assert;
* @author Lucas Ward
* @since 1.0
*/
public class DrivingQueryItemReader implements KeyedItemReader, InitializingBean,
public class DrivingQueryItemReader implements ItemReader, ItemKeyGenerator, InitializingBean,
ItemStream {
private boolean initialized = false;

2
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/KeyedItemReader.java → spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemKeyGenerator.java
View File

@@ -22,7 +22,7 @@ package org.springframework.batch.item;
* @author Dave Syer
*
*/
public interface KeyedItemReader extends ItemReader {
public interface ItemKeyGenerator {
/**
* Get a unique identifier for the item that can be used to cache it between

4
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemReaderRecoverer.java
View File

@@ -16,14 +16,14 @@
package org.springframework.batch.item.reader;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.item.KeyedItemReader;
import org.springframework.batch.item.ItemKeyGenerator;
/**
* @author Dave Syer
*
*/
public abstract class AbstractItemReaderRecoverer extends AbstractItemReader implements KeyedItemReader, ItemRecoverer {
public abstract class AbstractItemReaderRecoverer extends AbstractItemReader implements ItemKeyGenerator, ItemRecoverer {
public Object getKey(Object item) {
return item;
}

4
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java
View File

@@ -18,7 +18,7 @@ package org.springframework.batch.item.reader;
import org.springframework.batch.io.Skippable;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.KeyedItemReader;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
@@ -29,7 +29,7 @@ import org.springframework.util.Assert;
*
* @author Dave Syer
*/
public class DelegatingItemReader extends AbstractItemReader implements Skippable, InitializingBean, KeyedItemReader {
public class DelegatingItemReader extends AbstractItemReader implements Skippable, InitializingBean, ItemKeyGenerator {
private ItemReader itemReader;

4
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/JmsItemReader.java
View File

@@ -24,7 +24,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.FailedItemIdentifier;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.KeyedItemReader;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.item.exception.UnexpectedInputException;
import org.springframework.jms.JmsException;
import org.springframework.jms.core.JmsOperations;
@@ -40,7 +40,7 @@ import org.springframework.util.Assert;
* @author Dave Syer
*
*/
public class JmsItemReader extends AbstractItemReader implements KeyedItemReader, FailedItemIdentifier {
public class JmsItemReader extends AbstractItemReader implements ItemKeyGenerator, FailedItemIdentifier {
protected Log logger = LogFactory.getLog(getClass());

91
spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemReaderRetryCallback.java
View File

@@ -18,10 +18,10 @@ package org.springframework.batch.retry.callback;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.KeyedItemReader;
import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryContext;
import org.springframework.batch.retry.RetryPolicy;
@@ -42,27 +42,40 @@ import org.springframework.batch.retry.policy.ItemReaderRetryPolicy;
*/
public class ItemReaderRetryCallback implements RetryCallback {
private final static Log logger = LogFactory
.getLog(ItemReaderRetryCallback.class);
private final static Log logger = LogFactory.getLog(ItemReaderRetryCallback.class);
public static final String ITEM = ItemReaderRetryCallback.class.getName()
+ ".ITEM";
public static final String ITEM = ItemReaderRetryCallback.class.getName() + ".ITEM";
private KeyedItemReader provider;
private ItemReader reader;
private ItemWriter writer;
private ItemRecoverer recoverer;
public ItemReaderRetryCallback(KeyedItemReader provider,
ItemWriter writer) {
private ItemKeyGenerator keyGenerator;
private ItemKeyGenerator defaultKeyGenerator = new ItemKeyGenerator() {
public Object getKey(Object item) {
return item;
}
};
public ItemReaderRetryCallback(ItemReader reader, ItemWriter writer) {
this(reader, null, writer);
}
public ItemReaderRetryCallback(ItemReader reader, ItemKeyGenerator keyGenerator, ItemWriter writer) {
super();
this.provider = provider;
this.reader = reader;
this.writer = writer;
this.keyGenerator = keyGenerator;
}
/**
* Setter for injecting optional recovery handler.
* Setter for injecting optional recovery handler. If it is not injected but
* the reader or writer implement {@link ItemRecoverer}, one of those will
* be used instead (preferring the reader to the writer if both would be
* appropriate).
*
* @param recoveryHandler
*/
@@ -70,6 +83,17 @@ public class ItemReaderRetryCallback implements RetryCallback {
this.recoverer = recoverer;
}
/**
* Public setter for the {@link ItemKeyGenerator}. If it is not injected
* but the reader or writer implement {@link ItemKeyGenerator}, one of
* those will be used instead (preferring the reader to the writer if both
* would be appropriate).
* @param keyGenerator the keyGenerator to set
*/
public void setKeyGenerator(ItemKeyGenerator keyGenerator) {
this.keyGenerator = keyGenerator;
}
public Object doWithRetry(RetryContext context) throws Throwable {
// This requires a collaboration with the RetryPolicy...
if (!context.isExhaustedOnly()) {
@@ -82,10 +106,10 @@ public class ItemReaderRetryCallback implements RetryCallback {
Object item = context.getAttribute(ITEM);
if (item == null) {
try {
item = provider.read();
} catch (Exception e) {
throw new ExhaustedRetryException(
"Unexpected end of item provider", e);
item = reader.read();
}
catch (Exception e) {
throw new ExhaustedRetryException("Unexpected end of item provider", e);
}
if (item == null) {
// This is probably not fatal: in a batch we want to
@@ -107,9 +131,31 @@ public class ItemReaderRetryCallback implements RetryCallback {
}
/**
* Accessor for the {@link ItemRecoverer}. If the handler is null but
* the {@link ItemReader} is an instanceof {@link ItemRecoverer},
* then it will be returned instead.
* Accessor for the {@link ItemRecoverer}. If the handler is null but the
* {@link ItemReader} is an instance of {@link ItemRecoverer}, then it will
* be returned instead. If none of those strategies works then a default
* implementation of {@link ItemKeyGenerator} will be used that just returns
* the item.
*
* @return the {@link ItemRecoverer}.
*/
public ItemKeyGenerator getKeyGenerator() {
if (keyGenerator != null) {
return keyGenerator;
}
if (reader instanceof ItemKeyGenerator) {
return (ItemKeyGenerator) reader;
}
if (writer instanceof ItemKeyGenerator) {
return (ItemKeyGenerator) writer;
}
return defaultKeyGenerator;
}
/**
* Accessor for the {@link ItemRecoverer}. If the handler is null but the
* {@link ItemReader} is an instance of {@link ItemRecoverer}, then it will
* be returned instead.
*
* @return the {@link ItemRecoverer}.
*/
@@ -117,8 +163,11 @@ public class ItemReaderRetryCallback implements RetryCallback {
if (recoverer != null) {
return recoverer;
}
if (provider instanceof ItemRecoverer) {
return (ItemRecoverer) provider;
if (reader instanceof ItemRecoverer) {
return (ItemRecoverer) reader;
}
if (writer instanceof ItemRecoverer) {
return (ItemRecoverer) writer;
}
return null;
}
@@ -128,8 +177,8 @@ public class ItemReaderRetryCallback implements RetryCallback {
*
* @return the {@link ItemReader} instance.
*/
public KeyedItemReader getReader() {
return provider;
public ItemReader getReader() {
return reader;
}
}

80
spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemReaderRetryPolicy.java
View File

@@ -19,8 +19,9 @@ package org.springframework.batch.retry.policy;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.FailedItemIdentifier;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.item.KeyedItemReader;
import org.springframework.batch.repeat.synch.RepeatSynchronizationManager;
import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryContext;
@@ -32,11 +33,11 @@ import org.springframework.batch.retry.synch.RetrySynchronizationManager;
import org.springframework.util.Assert;
/**
* A {@link RetryPolicy} that detects an {@link ItemReaderRetryCallback} when
* it opens a new context, and uses it to make sure the item is in place for
* later decisions about how to retry or backoff. The callback should be an
* instance of {@link ItemReaderRetryCallback} otherwise an exception will be
* thrown when the context is created.
* A {@link RetryPolicy} that detects an {@link ItemReaderRetryCallback} when it
* opens a new context, and uses it to make sure the item is in place for later
* decisions about how to retry or backoff. The callback should be an instance
* of {@link ItemReaderRetryCallback} otherwise an exception will be thrown when
* the context is created.
*
* @author Dave Syer
*
@@ -45,9 +46,7 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy {
protected Log logger = LogFactory.getLog(getClass());
public static final String EXHAUSTED = ItemReaderRetryPolicy.class
.getName()
+ ".EXHAUSTED";
public static final String EXHAUSTED = ItemReaderRetryPolicy.class.getName() + ".EXHAUSTED";
private RetryPolicy delegate;
@@ -103,14 +102,12 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy {
*
* @see org.springframework.batch.retry.RetryPolicy#open(org.springframework.batch.retry.RetryCallback)
*
* @throws IllegalStateException
* if the callback is not of the required type.
* @throws IllegalStateException if the callback is not of the required
* type.
*/
public RetryContext open(RetryCallback callback) {
Assert.state(callback instanceof ItemReaderRetryCallback,
"Callback must be ItemProviderRetryCallback");
ItemReaderRetryContext context = new ItemReaderRetryContext(
(ItemReaderRetryCallback) callback);
Assert.state(callback instanceof ItemReaderRetryCallback, "Callback must be ItemProviderRetryCallback");
ItemReaderRetryContext context = new ItemReaderRetryContext((ItemReaderRetryCallback) callback);
context.open(callback);
return context;
}
@@ -120,10 +117,9 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy {
* implemented by subclasses), and remove the current item from the history.
*
* @see org.springframework.batch.retry.RetryPolicy#registerThrowable(org.springframework.batch.retry.RetryContext,
* java.lang.Throwable)
* java.lang.Throwable)
*/
public void registerThrowable(RetryContext context, Throwable throwable)
throws TerminatedRetryException {
public void registerThrowable(RetryContext context, Throwable throwable) throws TerminatedRetryException {
((RetryPolicy) context).registerThrowable(context, throwable);
// The throwable is stored in the delegate context.
}
@@ -137,23 +133,25 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy {
return ((RetryPolicy) context).handleRetryExhausted(context);
}
private class ItemReaderRetryContext extends RetryContextSupport
implements RetryPolicy {
private class ItemReaderRetryContext extends RetryContextSupport implements RetryPolicy {
private Object item;
// The delegate context...
private RetryContext delegateContext;
private KeyedItemReader reader;
private ItemReader reader;
private ItemRecoverer recoverer;
private ItemKeyGenerator keyGenerator;
public ItemReaderRetryContext(ItemReaderRetryCallback callback) {
super(RetrySynchronizationManager.getContext());
item = callback.next(this);
this.reader = callback.getReader();
this.recoverer = callback.getRecoverer();
this.keyGenerator = callback.getKeyGenerator();
}
public boolean canRetry(RetryContext context) {
@@ -165,9 +163,8 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy {
}
public RetryContext open(RetryCallback callback) {
if (hasFailed(reader, item)) {
this.delegateContext = retryContextCache.get(reader
.getKey(item));
if (hasFailed(reader, keyGenerator, item)) {
this.delegateContext = retryContextCache.get(keyGenerator.getKey(item));
}
if (this.delegateContext == null) {
// Only create a new context if we don't know the history of
@@ -178,37 +175,31 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy {
return null;
}
public void registerThrowable(RetryContext context, Throwable throwable)
throws TerminatedRetryException {
retryContextCache.put(reader.getKey(item), this.delegateContext);
public void registerThrowable(RetryContext context, Throwable throwable) throws TerminatedRetryException {
retryContextCache.put(keyGenerator.getKey(item), this.delegateContext);
delegate.registerThrowable(this.delegateContext, throwable);
}
public boolean isExternal() {
// Not called...
throw new UnsupportedOperationException(
"Not supported - this code should be unreachable.");
throw new UnsupportedOperationException("Not supported - this code should be unreachable.");
}
public boolean shouldRethrow(RetryContext context) {
// Not called...
throw new UnsupportedOperationException(
"Not supported - this code should be unreachable.");
throw new UnsupportedOperationException("Not supported - this code should be unreachable.");
}
public Object handleRetryExhausted(RetryContext context)
throws Exception {
public Object handleRetryExhausted(RetryContext context) throws Exception {
// If there is no going back, then we can remove the history
retryContextCache.remove(reader.getKey(item));
retryContextCache.remove(keyGenerator.getKey(item));
RepeatSynchronizationManager.setCompleteOnly();
if (recoverer != null) {
boolean success = recoverer.recover(item, context
.getLastThrowable());
boolean success = recoverer.recover(item, context.getLastThrowable());
if (!success) {
int count = context.getRetryCount();
logger.error(
"Could not recover from error after retry exhausted after ["
+ count + "] attempts.", context.getLastThrowable());
logger.error("Could not recover from error after retry exhausted after [" + count + "] attempts.",
context.getLastThrowable());
}
}
return item;
@@ -235,15 +226,16 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy {
* decision is delegated to the provider. Otherwise we just check the cache
* for the item key.
*
* @param provider
* @param reader
* @param keyGenerator
* @param item
* @return
*/
protected boolean hasFailed(KeyedItemReader provider, Object item) {
if (provider instanceof FailedItemIdentifier) {
return ((FailedItemIdentifier) provider).hasFailed(item);
protected boolean hasFailed(ItemReader reader, ItemKeyGenerator keyGenerator, Object item) {
if (reader instanceof FailedItemIdentifier) {
return ((FailedItemIdentifier) reader).hasFailed(item);
}
return retryContextCache.containsKey(provider.getKey(item));
return retryContextCache.containsKey(keyGenerator.getKey(item));
}
}