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

OPEN - issue BATCH-777: Parametrise RetryCallback and related interfaces

Browse Source 
Move key generator and item identifier to retry interceptor package
This commit is contained in:
dsyer
2008-08-25 16:08:19 +00:00
parent 6e36093b2e
commit d41464c176
19 changed files with 519 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files

136
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/jms/JmsItemReader.java
View File

@@ -16,27 +16,16 @@
package org.springframework.batch.item.jms;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
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.NewItemIdentifier;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.jms.JmsException;
import org.springframework.jms.core.JmsOperations;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.util.Assert;
/**
* An {@link ItemReader} for JMS using a {@link JmsTemplate}. The template
* should have a default destination, which will be used to provide items in
* {@link #read()}. If a recovery step is needed, set the error destination and
* the item will be sent there if processing fails in a stateful retry.
* {@link #read()}.<br/><br/>
*
* The implementation is thread safe after its properties are set (normal
* singleton behaviour).
@@ -44,60 +33,7 @@ import org.springframework.util.Assert;
* @author Dave Syer
*
*/
public class JmsItemReader<T> implements ItemReader<T>, ItemRecoverer<T,T>, ItemKeyGenerator,
NewItemIdentifier {
protected Log logger = LogFactory.getLog(getClass());
private JmsOperations jmsTemplate;
private Class<? extends T> itemType;
private String errorDestinationName;
private Destination errorDestination;
/**
* Set the error destination. Should not be the same as the default
* destination of the jms template.
* @param errorDestination a JMS Destination
*/
public void setErrorDestination(Destination errorDestination) {
this.errorDestination = errorDestination;
}
/**
* Set the error destination by name. Will be resolved by the destination
* resolver in the jms template.
*
* @param errorDestinationName the name of a JMS Destination
*/
public void setErrorDestinationName(String errorDestinationName) {
this.errorDestinationName = errorDestinationName;
}
/**
* Setter for jms template.
*
* @param jmsTemplate a {@link JmsOperations} instance
*/
public void setJmsTemplate(JmsOperations jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
/**
* Set the expected type of incoming message payloads. Set this to
* {@link Message} to receive the raw underlying message.
*
* @param itemType the java class of the items to be delivered. Typically
* the same as the class parameter
*
* @throws IllegalStateException if the message payload is of the wrong
* type.
*/
public void setItemType(Class<? extends T> itemType) {
this.itemType = itemType;
}
public class JmsItemReader<T> extends MessageTypeAccessor<T> implements ItemReader<T> {
@SuppressWarnings("unchecked")
public T read() {
@@ -112,72 +48,4 @@ public class JmsItemReader<T> implements ItemReader<T>, ItemRecoverer<T,T>, Item
return (T) result;
}
/**
* Send the message back to the provider using the specified error
* destination property of this reader. If the recovery is successful the
* item itself is returned, otherwise null.
*
* @see org.springframework.batch.item.ItemRecoverer#recover(Object,
* Throwable)
*/
public T recover(T item, Throwable cause) {
try {
if (errorDestination != null) {
jmsTemplate.convertAndSend(errorDestination, item);
}
else if (errorDestinationName != null) {
jmsTemplate.convertAndSend(errorDestinationName, item);
}
else {
// do nothing - it doesn't make sense to send the message back
// to the destination it came from
return null;
}
return item;
}
catch (JmsException e) {
logger.error("Could not recover because of JmsException.", e);
throw e;
}
}
/**
* If the message is a {@link Message} then returns the JMS message ID.
* Otherwise just delegate to parent class.
*
* @see org.springframework.batch.item.ItemKeyGenerator#getKey(java.lang.Object)
*
* @throws UnexpectedInputException if the JMS id cannot be determined from
* a JMS Message
*/
public Object getKey(Object item) {
if (itemType != null && itemType.isAssignableFrom(Message.class)) {
try {
return ((Message) item).getJMSMessageID();
}
catch (JMSException e) {
throw new UnexpectedInputException("Could not extract message ID", e);
}
}
return item;
}
/**
* If the item is a message, check the JMS re-delivered flag, otherwise
* return false to be on the safe side.
*
* @see org.springframework.batch.item.NewItemIdentifier#isNew(java.lang.Object)
*/
public boolean isNew(Object item) {
if (itemType != null && itemType.isAssignableFrom(Message.class)) {
try {
return !((Message) item).getJMSRedelivered();
}
catch (JMSException e) {
throw new UnexpectedInputException("Could not extract message ID", e);
}
}
return false;
}
}

64
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/jms/JmsItemWriter.java Normal file
View File

@@ -0,0 +1,64 @@
/*
* 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.item.jms;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.ItemWriter;
import org.springframework.jms.core.JmsOperations;
import org.springframework.jms.core.JmsTemplate;
/**
* An {@link ItemWriter} for JMS using a {@link JmsTemplate}. The template
* should have a default destination, which will be used to send items in
* {@link #write(List)}.<br/><br/>
*
* The implementation is thread safe after its properties are set (normal
* singleton behaviour).
*
* @author Dave Syer
*
*/
public class JmsItemWriter<T> implements ItemWriter<T> {
protected Log logger = LogFactory.getLog(getClass());
private JmsOperations jmsTemplate;
/**
* Setter for jms template.
*
* @param jmsTemplate a {@link JmsOperations} instance
*/
public void setJmsTemplate(JmsOperations jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
/**
* Send the items one-by-one to the default destination of the jms template.
*
* @see org.springframework.batch.item.ItemWriter#write(java.util.List)
*/
public void write(List<? extends T> items) throws Exception {
for (T item : items) {
jmsTemplate.convertAndSend(item);
}
}
}

61
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/jms/JmsMethodArgumentsKeyGenerator.java Normal file
View File

@@ -0,0 +1,61 @@
/*
* 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.item.jms;
import javax.jms.JMSException;
import javax.jms.Message;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.retry.interceptor.MethodArgumentsKeyGenerator;
/**
* A {@link MethodArgumentsKeyGenerator} for JMS
*
* @author Dave Syer
*
*/
public class JmsMethodArgumentsKeyGenerator implements MethodArgumentsKeyGenerator {
/**
* If the message is a {@link Message} then returns the JMS message ID.
* Otherwise just return the first argument.
*
* @see org.springframework.batch.retry.interceptor.MethodArgumentsKeyGenerator#getKey(Object[])
*
* @throws UnexpectedInputException if the JMS id cannot be determined from
* a JMS Message
* @throws IllegalArgumentException if the arguments are empty
*/
public Object getKey(Object[] items) {
for (Object item : items) {
if (item instanceof Message) {
try {
return ((Message) item).getJMSMessageID();
}
catch (JMSException e) {
throw new UnexpectedInputException("Could not extract message ID", e);
}
}
}
if (items.length == 0) {
throw new IllegalArgumentException(
"Method parameters are empty. The key generator cannot determine a unique key.");
}
return items[0];
}
}

63
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/jms/JmsMethodInvocationRecoverer.java Normal file
View File

@@ -0,0 +1,63 @@
/*
* 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.item.jms;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.retry.interceptor.MethodInvocationRecoverer;
import org.springframework.jms.JmsException;
import org.springframework.jms.core.JmsOperations;
/**
* @author Dave Syer
*
*/
public class JmsMethodInvocationRecoverer<T> implements MethodInvocationRecoverer<T> {
protected Log logger = LogFactory.getLog(getClass());
private JmsOperations jmsTemplate;
/**
* Setter for jms template.
*
* @param jmsTemplate a {@link JmsOperations} instance
*/
public void setJmsTemplate(JmsOperations jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
/**
* Send one message per item in the arguments list using the default destination of
* the jms template. If the recovery is successful null is returned.
*
* @see org.springframework.batch.retry.interceptor.MethodInvocationRecoverer#recover(Object[],
* Throwable)
*/
public T recover(Object[] items, Throwable cause) {
try {
for (Object item : items) {
jmsTemplate.convertAndSend(item);
}
return null;
}
catch (JmsException e) {
logger.error("Could not recover because of JmsException.", e);
throw e;
}
}
}

55
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/jms/JmsNewMethodArgumentsIdentifier.java Normal file
View File

@@ -0,0 +1,55 @@
/*
* 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.item.jms;
import javax.jms.JMSException;
import javax.jms.Message;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.retry.interceptor.NewMethodArgumentsIdentifier;
/**
* A {@link NewMethodArgumentsIdentifier} for JMS that looks for a message in
* the arguments and checks its delivery status.
*
* @author Dave Syer
*
*/
public class JmsNewMethodArgumentsIdentifier<T> implements NewMethodArgumentsIdentifier {
/**
* If any of the arguments is a message, check the JMS re-delivered flag and
* return it, otherwise return false to be on the safe side.
*
* @see org.springframework.batch.retry.interceptor.NewMethodArgumentsIdentifier#isNew(java.lang.Object[])
*/
public boolean isNew(Object[] args) {
for (Object item : args) {
if (item instanceof Message) {
try {
return !((Message) item).getJMSRedelivered();
}
catch (JMSException e) {
throw new UnexpectedInputException("Could not extract message ID", e);
}
}
}
return false;
}
}

68
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/jms/MessageTypeAccessor.java Normal file
View File

@@ -0,0 +1,68 @@
/*
* 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.item.jms;
import javax.jms.Message;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.jms.core.JmsOperations;
/**
* Base class for JMS concerns.
*
* @author Dave Syer
*
*/
class MessageTypeAccessor<T> {
protected Log logger = LogFactory.getLog(getClass());
protected Class<? extends T> itemType;
protected JmsOperations jmsTemplate;
/**
* Setter for jms template.
*
* @param jmsTemplate a {@link JmsOperations} instance
*/
public void setJmsTemplate(JmsOperations jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
/**
* Set the expected type of incoming message payloads. Set this to
* {@link Message} to receive the raw underlying message.
*
* @param itemType the java class of the items to be delivered. Typically
* the same as the class parameter
*
* @throws IllegalStateException if the message payload is of the wrong
* type.
*/
public void setItemType(Class<? extends T> itemType) {
this.itemType = itemType;
}
public boolean isMessageType() {
return itemType != null && itemType.isAssignableFrom(Message.class);
}
}

8
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemKeyGenerator.java → spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/MethodArgumentsKeyGenerator.java
View File

@@ -13,7 +13,9 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.item;
package org.springframework.batch.retry.interceptor;
import org.springframework.batch.item.ItemReader;
/**
* Extension of the {@link ItemReader} interface that allows items to be
@@ -22,7 +24,7 @@ package org.springframework.batch.item;
* @author Dave Syer
*
*/
public interface ItemKeyGenerator {
public interface MethodArgumentsKeyGenerator {
/**
* Get a unique identifier for the item that can be used to cache it between
@@ -31,6 +33,6 @@ public interface ItemKeyGenerator {
* @param item the current item.
* @return a unique identifier.
*/
Object getKey(Object item);
Object getKey(Object[] item);
}

10
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemRecoverer.java → spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/MethodInvocationRecoverer.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.batch.item;
package org.springframework.batch.retry.interceptor;
/**
@@ -22,7 +22,7 @@ package org.springframework.batch.item;
*
* @author Dave Syer
*/
public interface ItemRecoverer<T,S> {
public interface MethodInvocationRecoverer<T> {
/**
* Recover gracefully from an error. Clients can call this if processing of
@@ -30,11 +30,11 @@ public interface ItemRecoverer<T,S> {
* to decide whether to try more corrective action or perhaps throw an
* exception.
*
* @param data
* the item that failed.
* @param args
* the arguments for the method invocation that failed.
* @param cause
* the cause of the failure that led to this recovery.
* @return the value to be returned to the caller
*/
T recover(S data, Throwable cause);
T recover(Object[] args, Throwable cause);
}

16
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/NewItemIdentifier.java → spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/NewMethodArgumentsIdentifier.java
View File

@@ -14,24 +14,24 @@
* limitations under the License.
*/
package org.springframework.batch.item;
package org.springframework.batch.retry.interceptor;
/**
* Strategy interface to distinguish a new item from one that has been processed
* before and one that has not, e.g. by examining a message flag.
* Strategy interface to distinguish new arguments from ones that have been
* processed before, e.g. by examining a message flag.
*
* @author Dave Syer
*
*/
public interface NewItemIdentifier {
public interface NewMethodArgumentsIdentifier {
/**
* Inspect the item and determine if it has never been processed before.
* The safest choice when the answer is indeterminate is 'false'.
* Inspect the arguments and determine if they have never been processed
* before. The safest choice when the answer is indeterminate is 'false'.
*
* @param item the current item.
* @param args the current method arguments.
* @return true if the item is known to have never been processed before.
*/
boolean isNew(Object item);
boolean isNew(Object[] args);
}

41
spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptor.java
View File

@@ -20,9 +20,6 @@ import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.item.NewItemIdentifier;
import org.springframework.batch.retry.ExhaustedRetryException;
import org.springframework.batch.retry.RecoveryCallback;
import org.springframework.batch.retry.RetryCallback;
@@ -40,8 +37,8 @@ import org.springframework.util.ObjectUtils;
* a method on a service if it fails. The argument to the service method is
* treated as an item to be remembered in case the call fails. So the retry
* operation is stateful, and the item that failed is tracked by its unique key
* (via {@link ItemKeyGenerator}) until the retry is exhausted, at which point
* the {@link ItemRecoverer} is called.<br/>
* (via {@link MethodArgumentsKeyGenerator}) until the retry is exhausted, at which point
* the {@link MethodInvocationRecoverer} is called.<br/>
*
* The main use case for this is where the service is transactional, via a
* transaction interceptor on the interceptor chain. In this case the retry (and
@@ -57,11 +54,11 @@ public class StatefulRetryOperationsInterceptor implements MethodInterceptor {
private transient Log logger = LogFactory.getLog(getClass());
private ItemKeyGenerator keyGenerator;
private MethodArgumentsKeyGenerator keyGenerator;
private ItemRecoverer<? extends Object,Object[]> recoverer;
private MethodInvocationRecoverer<? extends Object> recoverer;
private NewItemIdentifier newItemIdentifier;
private NewMethodArgumentsIdentifier newMethodArgumentsIdentifier;
private final RetryTemplate retryTemplate = new RetryTemplate();
@@ -74,7 +71,7 @@ public class StatefulRetryOperationsInterceptor implements MethodInterceptor {
}
/**
* Public setter for the {@link ItemRecoverer} to use if the retry is
* Public setter for the {@link MethodInvocationRecoverer} to use if the retry is
* exhausted. The recoverer should be able to return an object of the same
* type as the target object because its return value will be used to return
* to the caller in the case of a recovery.<br/>
@@ -82,13 +79,13 @@ public class StatefulRetryOperationsInterceptor implements MethodInterceptor {
* If no recoverer is set then an exhausted retry will result in an
* {@link ExhaustedRetryException}.
*
* @param recoverer the {@link ItemRecoverer} to set
* @param recoverer the {@link MethodInvocationRecoverer} to set
*/
public void setRecoverer(ItemRecoverer<? extends Object,Object[]> recoverer) {
public void setRecoverer(MethodInvocationRecoverer<? extends Object> recoverer) {
this.recoverer = recoverer;
}
public void setKeyGenerator(ItemKeyGenerator keyGenerator) {
public void setKeyGenerator(MethodArgumentsKeyGenerator keyGenerator) {
this.keyGenerator = keyGenerator;
}
@@ -103,29 +100,29 @@ public class StatefulRetryOperationsInterceptor implements MethodInterceptor {
}
/**
* Public setter for the {@link NewItemIdentifier}. Only set this if the
* Public setter for the {@link NewMethodArgumentsIdentifier}. Only set this if the
* arguments to the intercepted method can be inspected to find out if they
* have never been processed before.
* @param newItemIdentifier the {@link NewItemIdentifier} to set
* @param newMethodArgumentsIdentifier the {@link NewMethodArgumentsIdentifier} to set
*/
public void setNewItemIdentifier(NewItemIdentifier newItemIdentifier) {
this.newItemIdentifier = newItemIdentifier;
public void setNewItemIdentifier(NewMethodArgumentsIdentifier newMethodArgumentsIdentifier) {
this.newMethodArgumentsIdentifier = newMethodArgumentsIdentifier;
}
/**
* Wrap the method invocation in a stateful retry with the policy and other
* helpers provided. If there is a failure the exception will generally be
* re-thrown. The only time it is not re-thrown is when retry is exhausted
* and the recovery path is taken (though the {@link ItemRecoverer} provided
* and the recovery path is taken (though the {@link MethodInvocationRecoverer} provided
* if there is one). In that case the value returned from the method
* invocation will be the value returned by the recoverer (so the return
* type for that should be the same as the intercepted method).
*
* @see org.aopalliance.intercept.MethodInterceptor#invoke(org.aopalliance.intercept.MethodInvocation)
* @see ItemRecoverer#recover(Object, Throwable)
* @see MethodInvocationRecoverer#recover(Object[], Throwable)
*
* @throws ExhaustedRetryException if the retry is exhausted and no
* {@link ItemRecoverer} is provided.
* {@link MethodInvocationRecoverer} is provided.
*/
public Object invoke(final MethodInvocation invocation) throws Throwable {
@@ -141,7 +138,7 @@ public class StatefulRetryOperationsInterceptor implements MethodInterceptor {
}
final Object item = arg;
RetryState retryState = new RetryState(keyGenerator != null ? keyGenerator.getKey(item) : item, newItemIdentifier != null ? newItemIdentifier.isNew(item) : false );
RetryState retryState = new RetryState(keyGenerator != null ? keyGenerator.getKey(args) : item, newMethodArgumentsIdentifier != null ? newMethodArgumentsIdentifier.isNew(args) : false );
Object result = retryTemplate.execute(new MethodInvocationRetryCallback(invocation), new ItemRecovererCallback(args, recoverer), retryState);
@@ -192,12 +189,12 @@ public class StatefulRetryOperationsInterceptor implements MethodInterceptor {
private final Object[] args;
private final ItemRecoverer<? extends Object,Object[]> recoverer;
private final MethodInvocationRecoverer<? extends Object> recoverer;
/**
* @param args the item that failed.
*/
private ItemRecovererCallback(Object[] args, ItemRecoverer<? extends Object,Object[]> recoverer) {
private ItemRecovererCallback(Object[] args, MethodInvocationRecoverer<? extends Object> recoverer) {
this.args = args;
this.recoverer = recoverer;
}