diff --git a/infrastructure/src/main/java/org/springframework/batch/io/support/HibernateAwareItemWriter.java b/infrastructure/src/main/java/org/springframework/batch/io/support/HibernateAwareItemWriter.java new file mode 100644 index 000000000..0e35d6743 --- /dev/null +++ b/infrastructure/src/main/java/org/springframework/batch/io/support/HibernateAwareItemWriter.java @@ -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.
+ * + * 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(); + } + + } + +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/support/HibernateAwareItemWriterTests.java b/infrastructure/src/test/java/org/springframework/batch/io/support/HibernateAwareItemWriterTests.java new file mode 100644 index 000000000..58fdeff7b --- /dev/null +++ b/infrastructure/src/test/java/org/springframework/batch/io/support/HibernateAwareItemWriterTests.java @@ -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()); + } + +} diff --git a/samples/src/main/java/org/springframework/batch/sample/dao/HibernateCreditWriter.java b/samples/src/main/java/org/springframework/batch/sample/dao/HibernateCreditWriter.java index beb43f3a6..ad0a392d0 100644 --- a/samples/src/main/java/org/springframework/batch/sample/dao/HibernateCreditWriter.java +++ b/samples/src/main/java/org/springframework/batch/sample/dao/HibernateCreditWriter.java @@ -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); } } diff --git a/samples/src/main/resources/jobs/hibernateJob.xml b/samples/src/main/resources/jobs/hibernateJob.xml index dc69b352f..3f5bc90cd 100644 --- a/samples/src/main/resources/jobs/hibernateJob.xml +++ b/samples/src/main/resources/jobs/hibernateJob.xml @@ -40,7 +40,7 @@ - + + + + + +