diff --git a/infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorInputSource.java b/infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorInputSource.java index 6e7e3cb67..5bc4f6878 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorInputSource.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorInputSource.java @@ -18,6 +18,7 @@ package org.springframework.batch.io.cursor; import java.util.Properties; import org.hibernate.ScrollableResults; +import org.hibernate.Session; import org.hibernate.SessionFactory; import org.hibernate.StatelessSession; import org.springframework.batch.io.InputSource; @@ -36,6 +37,17 @@ import org.springframework.util.ClassUtils; /** * {@link InputSource} for reading database records built on top of Hibernate. * + * It executes the HQL {@link #queryString} when initialized and iterates over + * the result set as {@link #read()} method is called, returning an object corresponding + * to current row. + * + * Input source can be configured to use either {@link StatelessSession} sufficient + * for simple mappings without the need to cascade to associated objects or standard hibernate + * {@link Session} for more advanced mappings or when caching is desired. + * + * When stateful session is used it will be cleared after successful commit *without* being flushed + * (no inserts or updates are expected). + * * @author Robert Kasanicky */ public class HibernateCursorInputSource implements InputSource, Restartable, InitializingBean, DisposableBean, @@ -44,13 +56,21 @@ public class HibernateCursorInputSource implements InputSource, Restartable, Ini private static final String RESTART_DATA_ROW_NUMBER_KEY = ClassUtils.getShortName(HibernateCursorInputSource.class)+".rowNumber"; private SessionFactory sessionFactory; - private StatelessSession session; + + private StatelessSession statelessSession; + + private Session statefulSession; + private ScrollableResults cursor; + private String queryString; + private boolean useStatelessSession = true; + private int lastCommitRowNumber = 0; private boolean initialized = false; + private TransactionSynchronization synchronization = new HibernateInputSourceTransactionSynchronization(); @@ -66,25 +86,37 @@ public class HibernateCursorInputSource implements InputSource, Restartable, Ini } /** - * Close the resultset cursor and hibernate session. + * Closes the result set cursor and hibernate session. */ public void close() { initialized = false; cursor.close(); - session.close(); + if (useStatelessSession) { + statelessSession.close(); + } else { + statefulSession.close(); + } } /** - * Create cursor for the query + * Creates cursor for the query. */ public void open() { - session = sessionFactory.openStatelessSession(); - cursor = session.createQuery(queryString).scroll(); + if (useStatelessSession) { + statelessSession = sessionFactory.openStatelessSession(); + cursor = statelessSession.createQuery(queryString).scroll(); + } else { + statefulSession = sessionFactory.openSession(); + cursor = statefulSession.createQuery(queryString).scroll(); + } BatchTransactionSynchronizationManager.registerSynchronization(synchronization ); initialized = true; } + /** + * @param sessionFactory hibernate session factory + */ public void setSessionFactory(SessionFactory sessionFactory) { this.sessionFactory = sessionFactory; } @@ -98,10 +130,24 @@ public class HibernateCursorInputSource implements InputSource, Restartable, Ini close(); } + /** + * @param queryString HQL query string + */ public void setQueryString(String queryString) { this.queryString = queryString; } + /** + * Can be set only in uninitialized state. + * @param useStatelessSession + * true to use {@link StatelessSession} + * false to use standard hibernate {@link Session} + */ + public void setUseStatelessSession(boolean useStatelessSession) { + Assert.state(!initialized); + this.useStatelessSession = useStatelessSession; + } + /** * @return the current row number wrapped as RestartData */ @@ -113,7 +159,7 @@ public class HibernateCursorInputSource implements InputSource, Restartable, Ini } /** - * Set the cursor to the received row number. + * Sets the cursor to the received row number. */ public void restoreFrom(RestartData data) { Assert.state(!initialized, @@ -138,6 +184,9 @@ public class HibernateCursorInputSource implements InputSource, Restartable, Ini cursor.setRowNumber(lastCommitRowNumber); } else if (status == TransactionSynchronization.STATUS_COMMITTED) { lastCommitRowNumber = cursor.getRowNumber(); + if (!useStatelessSession) { + statefulSession.clear(); + } } } } diff --git a/infrastructure/src/test/java/org/springframework/batch/io/cursor/HibernateInputSourceIntegrationTests.java b/infrastructure/src/test/java/org/springframework/batch/io/cursor/HibernateCursorInputSourceIntegrationTests.java similarity index 50% rename from infrastructure/src/test/java/org/springframework/batch/io/cursor/HibernateInputSourceIntegrationTests.java rename to infrastructure/src/test/java/org/springframework/batch/io/cursor/HibernateCursorInputSourceIntegrationTests.java index 1c0937737..7227377fa 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/cursor/HibernateInputSourceIntegrationTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/cursor/HibernateCursorInputSourceIntegrationTests.java @@ -1,36 +1,61 @@ package org.springframework.batch.io.cursor; import org.hibernate.SessionFactory; +import org.hibernate.StatelessSession; import org.springframework.batch.io.InputSource; -import org.springframework.batch.io.cursor.HibernateCursorInputSource; import org.springframework.batch.io.support.AbstractDataSourceInputSourceIntegrationTests; import org.springframework.core.io.ClassPathResource; import org.springframework.core.io.Resource; import org.springframework.orm.hibernate3.LocalSessionFactoryBean; /** - * Tests for {@link HibernateCursorInputSource} + * Tests for {@link HibernateCursorInputSource} using {@link StatelessSession}. * * @author Robert Kasanicky */ -public class HibernateInputSourceIntegrationTests extends AbstractDataSourceInputSourceIntegrationTests { +public class HibernateCursorInputSourceIntegrationTests extends AbstractDataSourceInputSourceIntegrationTests { protected InputSource createInputSource() throws Exception { LocalSessionFactoryBean factoryBean = new LocalSessionFactoryBean(); factoryBean.setDataSource(super.getJdbcTemplate().getDataSource()); - factoryBean.setMappingLocations(new Resource[]{new ClassPathResource("Foo.hbm.xml", getClass())}); + factoryBean.setMappingLocations(new Resource[] { new ClassPathResource("Foo.hbm.xml", getClass()) }); factoryBean.afterPropertiesSet(); - + SessionFactory sessionFactory = (SessionFactory) factoryBean.getObject(); - + String hsqlQuery = "from Foo"; - + HibernateCursorInputSource inputSource = new HibernateCursorInputSource(); inputSource.setQueryString(hsqlQuery); inputSource.setSessionFactory(sessionFactory); + inputSource.setUseStatelessSession(isUseStatelessSession()); inputSource.afterPropertiesSet(); - + return inputSource; } + protected boolean isUseStatelessSession() { + return true; + } + + /** + * Exception scenario. + * + * {@link HibernateCursorInputSource#setUseStatelessSession(boolean)} can be + * called only in uninitialized state. + */ + public void testSetUseStatelessSession() { + HibernateCursorInputSource inputSource = ((HibernateCursorInputSource) source); + + // initialize and call setter => error + inputSource.open(); + try { + inputSource.setUseStatelessSession(false); + fail(); + } + catch (IllegalStateException e) { + // expected + } + } + } diff --git a/infrastructure/src/test/java/org/springframework/batch/io/cursor/HibernateCursorInputSourceStatefulIntegrationTests.java b/infrastructure/src/test/java/org/springframework/batch/io/cursor/HibernateCursorInputSourceStatefulIntegrationTests.java new file mode 100644 index 000000000..946be0881 --- /dev/null +++ b/infrastructure/src/test/java/org/springframework/batch/io/cursor/HibernateCursorInputSourceStatefulIntegrationTests.java @@ -0,0 +1,16 @@ +package org.springframework.batch.io.cursor; + +import org.hibernate.Session; + +/** + * Tests for {@link HibernateCursorInputSource} using standard hibernate {@link Session}. + * + * @author Robert Kasanicky + */ +public class HibernateCursorInputSourceStatefulIntegrationTests extends HibernateCursorInputSourceIntegrationTests{ + + protected boolean isUseStatelessSession() { + return false; + } + +}