RESOLVED - issue BATCH-194: Incorrect exception handling when using Hibernate
http://opensource.atlassian.com/projects/spring/browse/BATCH-194 Created HibernateAwareItemWriter to abstract the flushing concerns into a framework class - to use it you have to register it with the chunkOperations as an interceptor *and* inject it with an ItemWriter for the step concerned. There might be an AOP approach to this that would reduce the burden of having to confugure it - we can look into that as part of the namespace / DSL work. See hibernateJob.xml for sample.
This commit is contained in:
dsyer
@@ -0,0 +1,272 @@
|
||||
/*
|
||||
* 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.io.support;
|
||||
|
||||
import java.util.HashSet;
|
||||
import java.util.Set;
|
||||
|
||||
import org.hibernate.SessionFactory;
|
||||
import org.springframework.batch.io.ItemWriter;
|
||||
import org.springframework.batch.repeat.ExitStatus;
|
||||
import org.springframework.batch.repeat.RepeatContext;
|
||||
import org.springframework.batch.repeat.RepeatInterceptor;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.core.AttributeAccessor;
|
||||
import org.springframework.orm.hibernate3.HibernateOperations;
|
||||
import org.springframework.orm.hibernate3.HibernateTemplate;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* {@link ItemWriter} that is aware of the Hibernate session and can take some
|
||||
* responsibilities to do with chunk boundaries away from a less smart
|
||||
* {@link ItemWriter} (the delegate). A delegate is required, and will be used
|
||||
* to do the actual writing of the item.<br/>
|
||||
*
|
||||
* This class implements {@link RepeatInterceptor} and it will only work if
|
||||
* properly registered. If the delegate is also a {@link RepeatInterceptor} then
|
||||
* it does not need to be separately registered as we make the callbacks here in
|
||||
* the right places.
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public class HibernateAwareItemWriter implements
|
||||
ItemWriter, RepeatInterceptor, InitializingBean {
|
||||
|
||||
/**
|
||||
* Key for items processed in the current transaction {@link RepeatContext}.
|
||||
*/
|
||||
private static final String ITEMS_PROCESSED = HibernateAwareItemWriterTests.class.getName()+".ITEMS_PROCESSED";
|
||||
|
||||
/**
|
||||
* Key for {@link RepeatContext} in transaction resource context.
|
||||
*/
|
||||
private static final String WRITER_REPEAT_CONTEXT = HibernateAwareItemWriterTests.class.getName()+".WRITER_REPEAT_CONTEXT";
|
||||
|
||||
private Set failed = new HashSet();
|
||||
|
||||
private ItemWriter delegate;
|
||||
|
||||
private HibernateOperations hibernateTemplate;
|
||||
|
||||
/**
|
||||
* Public setter for the {@link ItemWriter} property.
|
||||
*
|
||||
* @param delegate
|
||||
* the delegate to set
|
||||
*/
|
||||
public void setDelegate(ItemWriter delegate) {
|
||||
this.delegate = delegate;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the {@link HibernateOperations} property.
|
||||
*
|
||||
* @param hibernateTemplate the hibernateTemplate to set
|
||||
*/
|
||||
public void setHibernateTemplate(HibernateOperations hibernateTemplate) {
|
||||
this.hibernateTemplate = hibernateTemplate;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the Hibernate SessionFactory to be used internally.
|
||||
* Will automatically create a HibernateTemplate for the given SessionFactory.
|
||||
* @see #setHibernateTemplate
|
||||
*/
|
||||
public final void setSessionFactory(SessionFactory sessionFactory) {
|
||||
this.hibernateTemplate = new HibernateTemplate(sessionFactory);;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check mandatory properties - there must be a delegate.
|
||||
*
|
||||
* @see org.springframework.dao.support.DaoSupport#initDao()
|
||||
*/
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
Assert
|
||||
.notNull(delegate,
|
||||
"HibernateAwareItemWriter requires an ItemWriter as a delegate.");
|
||||
Assert
|
||||
.notNull(hibernateTemplate, "HibernateAwareItemWriter requires a HibernateOperations");
|
||||
}
|
||||
|
||||
/**
|
||||
* Use the delegate to actually do the writing, but flush aggressively if
|
||||
* the item was previously part of a failed chunk.
|
||||
*
|
||||
* @see org.springframework.batch.io.OutputSource#write(java.lang.Object)
|
||||
*/
|
||||
public void write(Object output) {
|
||||
getProcessed().add(output);
|
||||
delegate.write(output);
|
||||
flushIfNecessary(output);
|
||||
}
|
||||
|
||||
/**
|
||||
* Does nothing unless the delegate is also a {@link RepeatInterceptor} in
|
||||
* which case pass on the call to him.
|
||||
*
|
||||
* @see org.springframework.batch.repeat.RepeatInterceptor#before(org.springframework.batch.repeat.RepeatContext)
|
||||
*/
|
||||
public void before(RepeatContext context) {
|
||||
if (delegate instanceof RepeatInterceptor) {
|
||||
RepeatInterceptor interceptor = (RepeatInterceptor) delegate;
|
||||
interceptor.before(context);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Does nothing unless the delegate is also a {@link RepeatInterceptor} in
|
||||
* which case pass on the call to him.
|
||||
*
|
||||
* @see org.springframework.batch.repeat.RepeatInterceptor#after(org.springframework.batch.repeat.RepeatContext,
|
||||
* org.springframework.batch.repeat.ExitStatus)
|
||||
*/
|
||||
public void after(RepeatContext context, ExitStatus result) {
|
||||
if (delegate instanceof RepeatInterceptor) {
|
||||
RepeatInterceptor interceptor = (RepeatInterceptor) delegate;
|
||||
interceptor.after(context, result);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Flush the Hibernate session so that any batch exceptions are within the
|
||||
* RepeatContext. If the delegate is also a {@link RepeatInterceptor} then
|
||||
* it will be given the call before flushing.
|
||||
*
|
||||
*
|
||||
* @see org.springframework.batch.repeat.RepeatInterceptor#close(org.springframework.batch.repeat.RepeatContext)
|
||||
*/
|
||||
public void close(RepeatContext context) {
|
||||
try {
|
||||
if (delegate instanceof RepeatInterceptor) {
|
||||
RepeatInterceptor interceptor = (RepeatInterceptor) delegate;
|
||||
interceptor.close(context);
|
||||
}
|
||||
hibernateTemplate.flush();
|
||||
} catch (RuntimeException e) {
|
||||
synchronized (failed) {
|
||||
failed.addAll(getProcessed());
|
||||
}
|
||||
// onError will not be called after close() by the framework so we
|
||||
// have to do it here.
|
||||
onError(context, e);
|
||||
throw e;
|
||||
}
|
||||
unsetContext();
|
||||
}
|
||||
|
||||
/**
|
||||
* Does nothing unless the delegate is also a {@link RepeatInterceptor} in
|
||||
* which case pass on the call to him.
|
||||
*
|
||||
* @see org.springframework.batch.repeat.RepeatInterceptor#onError(org.springframework.batch.repeat.RepeatContext,
|
||||
* java.lang.Throwable)
|
||||
*/
|
||||
public void onError(RepeatContext context, Throwable e) {
|
||||
if (delegate instanceof RepeatInterceptor) {
|
||||
RepeatInterceptor interceptor = (RepeatInterceptor) delegate;
|
||||
interceptor.onError(context, e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets up the context as a transaction resource so that we can store state
|
||||
* and refer back to it in the {@link #write(Object)} method. If the
|
||||
* delegate is also a {@link RepeatInterceptor} then it will be given the
|
||||
* call afterwards.
|
||||
*
|
||||
* @see org.springframework.batch.repeat.RepeatInterceptor#open(org.springframework.batch.repeat.RepeatContext)
|
||||
*/
|
||||
public void open(RepeatContext context) {
|
||||
this.setContext(context);
|
||||
getProcessed().clear();
|
||||
if (delegate instanceof RepeatInterceptor) {
|
||||
RepeatInterceptor interceptor = (RepeatInterceptor) delegate;
|
||||
interceptor.open(context);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Accessor for the list of processed items in this transaction.
|
||||
*
|
||||
* @return the processed
|
||||
*/
|
||||
private Set getProcessed() {
|
||||
Assert.state(TransactionSynchronizationManager
|
||||
.hasResource(WRITER_REPEAT_CONTEXT),
|
||||
"RepeatContext not bound to transaction.");
|
||||
Set processed = (Set) ((AttributeAccessor) TransactionSynchronizationManager
|
||||
.getResource(WRITER_REPEAT_CONTEXT))
|
||||
.getAttribute(ITEMS_PROCESSED);
|
||||
return processed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set up the {@link RepeatContext} as a transaction resource.
|
||||
*
|
||||
* @param context
|
||||
* the context to set
|
||||
*/
|
||||
private void setContext(RepeatContext context) {
|
||||
if (TransactionSynchronizationManager
|
||||
.hasResource(WRITER_REPEAT_CONTEXT)) {
|
||||
return;
|
||||
}
|
||||
TransactionSynchronizationManager.bindResource(WRITER_REPEAT_CONTEXT,
|
||||
context);
|
||||
context.setAttribute(ITEMS_PROCESSED, new HashSet());
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove the transaction resource associated with this context.
|
||||
*/
|
||||
private void unsetContext() {
|
||||
if (!TransactionSynchronizationManager
|
||||
.hasResource(WRITER_REPEAT_CONTEXT)) {
|
||||
return;
|
||||
}
|
||||
TransactionSynchronizationManager.unbindResource(WRITER_REPEAT_CONTEXT);
|
||||
}
|
||||
|
||||
/**
|
||||
* Accessor for the context property.
|
||||
*
|
||||
* @param output
|
||||
*
|
||||
* @return the context
|
||||
*/
|
||||
private void flushIfNecessary(Object output) {
|
||||
RepeatContext context = (RepeatContext) TransactionSynchronizationManager
|
||||
.getResource(WRITER_REPEAT_CONTEXT);
|
||||
boolean flush;
|
||||
synchronized (failed) {
|
||||
flush = failed.contains(output);
|
||||
}
|
||||
if (flush) {
|
||||
// Force early completion to commit aggressively if we encounter a
|
||||
// failed item (from a failed chunk but we don't know which one was
|
||||
// the problem).
|
||||
context.setCompleteOnly();
|
||||
// Flush now, so that if there is a failure this record can be
|
||||
// skipped.
|
||||
hibernateTemplate.flush();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,250 @@
|
||||
/*
|
||||
* 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.io.support;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import junit.framework.TestCase;
|
||||
|
||||
import org.springframework.batch.io.ItemWriter;
|
||||
import org.springframework.batch.repeat.ExitStatus;
|
||||
import org.springframework.batch.repeat.RepeatContext;
|
||||
import org.springframework.batch.repeat.RepeatInterceptor;
|
||||
import org.springframework.batch.repeat.context.RepeatContextSupport;
|
||||
import org.springframework.dao.DataAccessException;
|
||||
import org.springframework.orm.hibernate3.HibernateTemplate;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
|
||||
/**
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public class HibernateAwareItemWriterTests extends TestCase {
|
||||
|
||||
private class StubItemWriter implements ItemWriter, RepeatInterceptor {
|
||||
public void write(Object item) {
|
||||
list.add(item);
|
||||
}
|
||||
|
||||
public void after(RepeatContext context, ExitStatus result) {
|
||||
list.add(result);
|
||||
}
|
||||
|
||||
public void before(RepeatContext context) {
|
||||
list.add(context);
|
||||
}
|
||||
|
||||
public void close(RepeatContext context) {
|
||||
list.add(context);
|
||||
}
|
||||
|
||||
public void onError(RepeatContext context, Throwable e) {
|
||||
list.add(e);
|
||||
}
|
||||
|
||||
public void open(RepeatContext context) {
|
||||
list.add(context);
|
||||
}
|
||||
}
|
||||
|
||||
HibernateAwareItemWriter writer = new HibernateAwareItemWriter();
|
||||
|
||||
final List list = new ArrayList();
|
||||
|
||||
private RepeatContextSupport context;
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
*
|
||||
* @see junit.framework.TestCase#setUp()
|
||||
*/
|
||||
protected void setUp() throws Exception {
|
||||
writer.setDelegate(new StubItemWriter());
|
||||
context = new RepeatContextSupport(null);
|
||||
writer.open(context);
|
||||
writer.setHibernateTemplate(new HibernateTemplate() {
|
||||
public void flush() throws DataAccessException {
|
||||
list.add("flush");
|
||||
}
|
||||
});
|
||||
list.clear();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see junit.framework.TestCase#tearDown()
|
||||
*/
|
||||
protected void tearDown() throws Exception {
|
||||
Map map = TransactionSynchronizationManager.getResourceMap();
|
||||
for (Iterator iterator = map.keySet().iterator(); iterator.hasNext();) {
|
||||
String key = (String) iterator.next();
|
||||
TransactionSynchronizationManager.unbindResource(key);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Test method for
|
||||
* {@link org.springframework.batch.io.support.HibernateAwareItemWriter#initDao()}.
|
||||
*
|
||||
* @throws Exception
|
||||
*/
|
||||
public void testAfterPropertiesSet() throws Exception {
|
||||
writer = new HibernateAwareItemWriter();
|
||||
try {
|
||||
writer.afterPropertiesSet();
|
||||
fail("Expected IllegalArgumentException");
|
||||
} catch (IllegalArgumentException e) {
|
||||
// expected
|
||||
assertTrue("Wrong message for exception: " + e.getMessage(), e
|
||||
.getMessage().indexOf("delegate") >= 0);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Test method for
|
||||
* {@link org.springframework.batch.io.support.HibernateAwareItemWriter#initDao()}.
|
||||
*
|
||||
* @throws Exception
|
||||
*/
|
||||
public void testAfterPropertiesSetWithDelegate() throws Exception {
|
||||
writer.afterPropertiesSet();
|
||||
}
|
||||
|
||||
/**
|
||||
* Test method for
|
||||
* {@link org.springframework.batch.io.support.HibernateAwareItemWriter#write(java.lang.Object)}.
|
||||
*/
|
||||
public void testWrite() {
|
||||
writer.write("foo");
|
||||
assertEquals(1, list.size());
|
||||
assertTrue(list.contains("foo"));
|
||||
}
|
||||
|
||||
/**
|
||||
* Test method for
|
||||
* {@link org.springframework.batch.io.support.HibernateAwareItemWriter#write(java.lang.Object)}.
|
||||
*/
|
||||
public void testCloseWithFailure() {
|
||||
final RuntimeException ex = new RuntimeException("bar");
|
||||
writer.setHibernateTemplate(new HibernateTemplate() {
|
||||
public void flush() throws DataAccessException {
|
||||
throw ex;
|
||||
}
|
||||
});
|
||||
try {
|
||||
writer.close(context);
|
||||
fail("Expected RuntimeException");
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("bar", e.getMessage());
|
||||
}
|
||||
assertEquals(2, list.size());
|
||||
assertTrue(list.contains(ex));
|
||||
assertTrue(list.contains(context));
|
||||
}
|
||||
|
||||
/**
|
||||
* Test method for
|
||||
* {@link org.springframework.batch.io.support.HibernateAwareItemWriter#write(java.lang.Object)}.
|
||||
*/
|
||||
public void testWriteAndCloseWithFailure() {
|
||||
final RuntimeException ex = new RuntimeException("bar");
|
||||
writer.setHibernateTemplate(new HibernateTemplate() {
|
||||
public void flush() throws DataAccessException {
|
||||
throw ex;
|
||||
}
|
||||
});
|
||||
writer.write("foo");
|
||||
try {
|
||||
writer.close(context);
|
||||
fail("Expected RuntimeException");
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("bar", e.getMessage());
|
||||
}
|
||||
assertEquals(3, list.size());
|
||||
assertTrue(list.contains(ex));
|
||||
assertTrue(list.contains(context));
|
||||
writer.setHibernateTemplate(new HibernateTemplate() {
|
||||
public void flush() throws DataAccessException {
|
||||
list.add("flush");
|
||||
}
|
||||
});
|
||||
writer.write("foo");
|
||||
assertEquals(5, list.size());
|
||||
assertTrue(list.contains("flush"));
|
||||
}
|
||||
|
||||
/**
|
||||
* Test method for
|
||||
* {@link org.springframework.batch.io.support.HibernateAwareItemWriter#before(org.springframework.batch.repeat.RepeatContext)}.
|
||||
*/
|
||||
public void testBefore() {
|
||||
writer.before(context);
|
||||
assertEquals(1, list.size());
|
||||
}
|
||||
|
||||
/**
|
||||
* Test method for
|
||||
* {@link org.springframework.batch.io.support.HibernateAwareItemWriter#after(org.springframework.batch.repeat.RepeatContext, org.springframework.batch.repeat.ExitStatus)}.
|
||||
*/
|
||||
public void testAfter() {
|
||||
writer.after(context, ExitStatus.FINISHED);
|
||||
assertEquals(1, list.size());
|
||||
}
|
||||
|
||||
/**
|
||||
* Test method for
|
||||
* {@link org.springframework.batch.io.support.HibernateAwareItemWriter#close(org.springframework.batch.repeat.RepeatContext)}.
|
||||
*/
|
||||
public void testClose() {
|
||||
writer.close(context);
|
||||
assertEquals(2, list.size());
|
||||
assertTrue(list.contains("flush"));
|
||||
}
|
||||
|
||||
/**
|
||||
* Test method for
|
||||
* {@link org.springframework.batch.io.support.HibernateAwareItemWriter#close(org.springframework.batch.repeat.RepeatContext)}.
|
||||
*/
|
||||
public void testCloseAfterClear() {
|
||||
Map map = TransactionSynchronizationManager.getResourceMap();
|
||||
String key = (String) map.keySet().iterator().next();
|
||||
TransactionSynchronizationManager.unbindResource(key);
|
||||
writer.close(context);
|
||||
assertEquals(2, list.size());
|
||||
assertTrue(list.contains("flush"));
|
||||
}
|
||||
|
||||
/**
|
||||
* Test method for
|
||||
* {@link org.springframework.batch.io.support.HibernateAwareItemWriter#onError(org.springframework.batch.repeat.RepeatContext, java.lang.Throwable)}.
|
||||
*/
|
||||
public void testOnError() {
|
||||
writer.onError(context, new Exception());
|
||||
assertEquals(1, list.size());
|
||||
}
|
||||
|
||||
/**
|
||||
* Test method for
|
||||
* {@link org.springframework.batch.io.support.HibernateAwareItemWriter#open(org.springframework.batch.repeat.RepeatContext)}.
|
||||
*/
|
||||
public void testOpen() {
|
||||
writer.open(context);
|
||||
assertEquals(1, list.size());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -16,18 +16,13 @@
|
||||
package org.springframework.batch.sample.dao;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
|
||||
import org.springframework.batch.repeat.ExitStatus;
|
||||
import org.springframework.batch.repeat.RepeatContext;
|
||||
import org.springframework.batch.repeat.RepeatInterceptor;
|
||||
import org.springframework.batch.sample.domain.CustomerCredit;
|
||||
import org.springframework.core.AttributeAccessor;
|
||||
import org.springframework.orm.hibernate3.support.HibernateDaoSupport;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* @author Lucas Ward
|
||||
@@ -37,17 +32,8 @@ import org.springframework.util.Assert;
|
||||
public class HibernateCreditWriter extends HibernateDaoSupport implements
|
||||
CustomerCreditWriter, RepeatInterceptor {
|
||||
|
||||
/**
|
||||
*
|
||||
*/
|
||||
private static final String ITEMS_PROCESSED = "ITEMS_PROCESSED";
|
||||
/**
|
||||
*
|
||||
*/
|
||||
private static final String WRITER_REPEAT_CONTEXT = "WRITER_REPEAT_CONTEXT";
|
||||
private int failOnFlush = -1;
|
||||
private List errors = new ArrayList();
|
||||
private Set failed = new HashSet();
|
||||
|
||||
/**
|
||||
* Public accessor for the errors property.
|
||||
@@ -82,99 +68,45 @@ public class HibernateCreditWriter extends HibernateDaoSupport implements
|
||||
* @see org.springframework.batch.io.OutputSource#write(java.lang.Object)
|
||||
*/
|
||||
public void write(Object output) {
|
||||
getProcessed().add(output);
|
||||
writeCredit((CustomerCredit) output);
|
||||
if (failed.contains(output)) {
|
||||
// Force early completion to commit aggressively if we encounter a
|
||||
// failed item (from a failed chunk but we don't know which one was
|
||||
// the problem).
|
||||
getContext().setCompleteOnly();
|
||||
// Flush now, so that if there is a failure this record will be
|
||||
// skipped.
|
||||
getHibernateTemplate().flush();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the failOnFlush property.
|
||||
*
|
||||
* @param failOnFlush
|
||||
* true if you want to fail on flush (for testing)
|
||||
* the ID of the record you want to fail on flush (for testing)
|
||||
*/
|
||||
public void setFailOnFlush(int failOnFlush) {
|
||||
this.failOnFlush = failOnFlush;
|
||||
}
|
||||
|
||||
public void before(RepeatContext context) {
|
||||
}
|
||||
|
||||
public void after(RepeatContext context, ExitStatus result) {
|
||||
}
|
||||
|
||||
/**
|
||||
* Flush the Hibernate session so that any batch exceptions are within the
|
||||
* RepeatContext.
|
||||
*
|
||||
* @see org.springframework.batch.repeat.RepeatInterceptor#close(org.springframework.batch.repeat.RepeatContext)
|
||||
*/
|
||||
public void close(RepeatContext context) {
|
||||
try {
|
||||
getHibernateTemplate().flush();
|
||||
} catch (RuntimeException e) {
|
||||
failed.addAll(getProcessed());
|
||||
// onError will not be called after close() by the framework so we
|
||||
// have to do it here.
|
||||
onError(context, e);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
public void onError(RepeatContext context, Throwable e) {
|
||||
errors.add(e);
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.batch.repeat.RepeatInterceptor#after(org.springframework.batch.repeat.RepeatContext, org.springframework.batch.repeat.ExitStatus)
|
||||
*/
|
||||
public void after(RepeatContext context, ExitStatus result) {
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.batch.repeat.RepeatInterceptor#before(org.springframework.batch.repeat.RepeatContext)
|
||||
*/
|
||||
public void before(RepeatContext context) {
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.batch.repeat.RepeatInterceptor#close(org.springframework.batch.repeat.RepeatContext)
|
||||
*/
|
||||
public void close(RepeatContext context) {
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.batch.repeat.RepeatInterceptor#open(org.springframework.batch.repeat.RepeatContext)
|
||||
*/
|
||||
public void open(RepeatContext context) {
|
||||
this.setContext(context);
|
||||
errors.clear();
|
||||
getProcessed().clear();
|
||||
}
|
||||
|
||||
/**
|
||||
* Public accessor for the processed property.
|
||||
*
|
||||
* @return the processed
|
||||
*/
|
||||
private Set getProcessed() {
|
||||
Assert.state(TransactionSynchronizationManager
|
||||
.hasResource(WRITER_REPEAT_CONTEXT),
|
||||
"RepeatContext not bound to transaction.");
|
||||
Set processed = (Set) ((AttributeAccessor) TransactionSynchronizationManager
|
||||
.getResource(WRITER_REPEAT_CONTEXT))
|
||||
.getAttribute(ITEMS_PROCESSED);
|
||||
return processed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the {@link RepeatContext} property.
|
||||
*
|
||||
* @param context
|
||||
* the context to set
|
||||
*/
|
||||
private void setContext(RepeatContext context) {
|
||||
if (TransactionSynchronizationManager.hasResource(WRITER_REPEAT_CONTEXT)){
|
||||
return;
|
||||
}
|
||||
TransactionSynchronizationManager.bindResource(WRITER_REPEAT_CONTEXT, context);
|
||||
context.setAttribute(ITEMS_PROCESSED, new HashSet());
|
||||
}
|
||||
|
||||
/**
|
||||
* Public accessor for the context property.
|
||||
*
|
||||
* @return the context
|
||||
*/
|
||||
private RepeatContext getContext() {
|
||||
return (RepeatContext) TransactionSynchronizationManager.getResource(WRITER_REPEAT_CONTEXT);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -40,7 +40,7 @@
|
||||
<property name="chunkOperations">
|
||||
<bean
|
||||
class="org.springframework.batch.repeat.support.RepeatTemplate">
|
||||
<property name="interceptor"
|
||||
<property name="interceptors"
|
||||
ref="hibernateOutputSource" />
|
||||
<property name="completionPolicy">
|
||||
<bean
|
||||
@@ -64,7 +64,14 @@
|
||||
</property>
|
||||
</bean>
|
||||
|
||||
<!-- This is a framework class that needs a delegate and also needs to be registered as a RepeatInterceptor in the chunk -->
|
||||
<bean id="hibernateOutputSource"
|
||||
class="org.springframework.batch.io.support.HibernateAwareItemWriter">
|
||||
<property name="sessionFactory" ref="sessionFactory" />
|
||||
<property name="delegate" ref="hibernateCreditWriter" />
|
||||
</bean>
|
||||
|
||||
<bean id="hibernateCreditWriter"
|
||||
class="org.springframework.batch.sample.dao.HibernateCreditWriter">
|
||||
<property name="sessionFactory" ref="sessionFactory" />
|
||||
</bean>
|
||||
|
||||
Reference in New Issue
Block a user