diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java index 55d1263e5..fdcc2084f 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java @@ -17,8 +17,6 @@ package org.springframework.batch.item.database; import java.util.Map; -import org.hibernate.Query; -import org.hibernate.ScrollMode; import org.hibernate.ScrollableResults; import org.hibernate.Session; import org.hibernate.SessionFactory; @@ -27,12 +25,10 @@ import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemStreamException; -import org.springframework.batch.item.database.support.AbstractHibernateQueryProvider; import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; -import org.springframework.util.StringUtils; /** * {@link ItemReader} for reading database records built on top of Hibernate. It @@ -59,124 +55,70 @@ import org.springframework.util.StringUtils; public class HibernateCursorItemReader extends AbstractItemCountingItemStreamItemReader implements ItemStream, InitializingBean { - private SessionFactory sessionFactory; - - private StatelessSession statelessSession; - - private Session statefulSession; - - private ScrollableResults cursor; - - private String queryString = ""; - - private String queryName = ""; - - private Map parameterValues; - - private HibernateQueryProvider queryProvider; - - private boolean useStatelessSession = true; - - private boolean initialized = false; - - private int fetchSize = 0; + private HibernateItemReaderHelper helper = new HibernateItemReaderHelper(); public HibernateCursorItemReader() { setName(ClassUtils.getShortName(HibernateCursorItemReader.class)); } - + + private ScrollableResults cursor; + + private boolean initialized = false; + + public void afterPropertiesSet() throws Exception { + helper.afterPropertiesSet(); + } + /** * The parameter values to apply to a query (map of name:value). * * @param parameterValues the parameter values to set */ public void setParameterValues(Map parameterValues) { - this.parameterValues = parameterValues; - } - - /** - * Open appropriate type of hibernate session and create the query. - */ - private Query createQuery() { - - if (useStatelessSession) { - statelessSession = sessionFactory.openStatelessSession(); - if (queryProvider != null) { - queryProvider.setStatelessSession(statelessSession); - } - else { - if (StringUtils.hasText(queryName)) { - return statelessSession.getNamedQuery(queryName); - } - else { - return statelessSession.createQuery(queryString); - } - } - } - else { - statefulSession = sessionFactory.openSession(); - if (queryProvider != null) { - queryProvider.setSession(statefulSession); - } - else { - if (StringUtils.hasText(queryName)) { - return statefulSession.getNamedQuery(queryName); - } - else { - return statefulSession.createQuery(queryString); - } - } - } - - // if queryProvider is set use it to create a query - return queryProvider.createQuery(); - - } - - /** - * @param sessionFactory hibernate session factory - */ - public void setSessionFactory(SessionFactory sessionFactory) { - this.sessionFactory = sessionFactory; - } - - public void afterPropertiesSet() throws Exception { - - Assert.state(sessionFactory!=null, "A SessionFactory must be provided"); - Assert.state(fetchSize >= 0, "fetchSize must not be negative"); - - if (queryProvider == null) { - Assert.notNull(sessionFactory, "session factory must be set"); - Assert.state(StringUtils.hasText(queryString) ^ StringUtils.hasText(queryName), - "queryString or queryName must be set"); - } - // making sure that the appropriate (Hibernate) query provider is set - else { - Assert.state(queryProvider instanceof AbstractHibernateQueryProvider, - "Hibernate query provider must be set"); - } - + helper.setParameterValues(parameterValues); } /** + * A query name for an externalized query. Either this or the { + * {@link #setQueryString(String) query string} or the { + * {@link #setQueryProvider(HibernateQueryProvider) query provider} should + * be set. + * * @param queryName name of a hibernate named query */ public void setQueryName(String queryName) { - this.queryName = queryName; - } - - /** - * @param queryString HQL query string - */ - public void setQueryString(String queryString) { - this.queryString = queryString; + helper.setQueryName(queryName); } /** + * A query provider. Either this or the {{@link #setQueryString(String) + * query string} or the {{@link #setQueryName(String) query name} should be + * set. + * * @param queryProvider Hibernate query provider */ public void setQueryProvider(HibernateQueryProvider queryProvider) { - this.queryProvider = queryProvider; + helper.setQueryProvider(queryProvider); + } + + /** + * A query string in HQL. Either this or the { + * {@link #setQueryProvider(HibernateQueryProvider) query provider} or the { + * {@link #setQueryName(String) query name} should be set. + * + * @param queryString HQL query string + */ + public void setQueryString(String queryString) { + helper.setQueryString(queryString); + } + + /** + * The Hibernate SessionFactory to use the create a session. + * + * @param sessionFactory the {@link SessionFactory} to set + */ + public void setSessionFactory(SessionFactory sessionFactory) { + helper.setSessionFactory(sessionFactory); } /** @@ -187,32 +129,7 @@ public class HibernateCursorItemReader extends AbstractItemCountingItemStream * {@link Session} */ public void setUseStatelessSession(boolean useStatelessSession) { - Assert.state(!initialized); - this.useStatelessSession = useStatelessSession; - } - - /** - * Clears the session if not stateful and delegates to super class. - */ - @Override - public void update(ExecutionContext executionContext) throws ItemStreamException { - super.update(executionContext); - if (!useStatelessSession) { - statefulSession.clear(); - } - } - - /** - * Gives the JDBC driver a hint as to the number of rows that should be - * fetched from the database when more rows are needed for this - * ResultSet object. If the fetch size specified is zero, the - * JDBC driver ignores the value. - * - * @param fetchSize the number of rows to fetch, 0 by default - * @see Query#setFetchSize(int) - */ - public void setFetchSize(int fetchSize) { - this.fetchSize = fetchSize; + helper.setUseStatelessSession(useStatelessSession); } protected T doRead() throws Exception { @@ -241,48 +158,53 @@ public class HibernateCursorItemReader extends AbstractItemCountingItemStream } /** - * Open hibernate session and create a forward-only cursor for the - * query. + * Open hibernate session and create a forward-only cursor for the query. */ protected void doOpen() throws Exception { Assert.state(!initialized, "Cannot open an already opened ItemReader, call close first"); - - Query query = createQuery(); - if (parameterValues!=null) { - query.setProperties(parameterValues); - } - cursor = query.setFetchSize(fetchSize).scroll(ScrollMode.FORWARD_ONLY); - + cursor = helper.getForwardOnlyCursor(); initialized = true; } + + /** + * Update the context and clear the session if stateful. + * + * @param executionContext the current {@link ExecutionContext} + * @throws ItemStreamException if there is a problem + */ + @Override + public void update(ExecutionContext executionContext) throws ItemStreamException { + super.update(executionContext); + helper.clear(); + } + /** + * Wind forward through the result set to the item requested. Also clears + * the session every now and then (if stateful) to avoid memory problems. + * The frequency of session clearing is the larger of the fetch size (if + * set) and 100. + * + * @param itemIndex the first item to read + * @throws Exception if there is a problem + * @see AbstractItemCountingItemStreamItemReader#jumpToItem(int) + */ @Override protected void jumpToItem(int itemIndex) throws Exception { - for (int i = 0; i < itemIndex; i++) { - cursor.next(); - } + helper.jumpToItem(cursor, itemIndex); } /** * Close the cursor and hibernate session. */ protected void doClose() throws Exception { + initialized = false; if (cursor != null) { cursor.close(); } - if (useStatelessSession) { - if (statelessSession != null) { - statelessSession.close(); - } - } - else { - if (statefulSession != null) { - statefulSession.close(); - } - } + helper.close(); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateItemReaderHelper.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateItemReaderHelper.java new file mode 100644 index 000000000..d7a3a38d9 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateItemReaderHelper.java @@ -0,0 +1,233 @@ +/* + * Copyright 2006-2010 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.database; + +import java.util.Collection; +import java.util.List; +import java.util.Map; + +import org.hibernate.Query; +import org.hibernate.ScrollMode; +import org.hibernate.ScrollableResults; +import org.hibernate.Session; +import org.hibernate.SessionFactory; +import org.hibernate.StatelessSession; +import org.springframework.batch.item.database.support.AbstractHibernateQueryProvider; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * Internal shared state helper for hibernate readers. + * + * @author Dave Syer + * + */ +class HibernateItemReaderHelper { + + private SessionFactory sessionFactory; + + private String queryString = ""; + + private String queryName = ""; + + private Map parameterValues; + + private HibernateQueryProvider queryProvider; + + private boolean useStatelessSession = true; + + private int fetchSize = 0; + + private StatelessSession statelessSession; + + private Session statefulSession; + + /** + * @param queryName name of a hibernate named query + */ + public void setQueryName(String queryName) { + this.queryName = queryName; + } + + /** + * @param queryString HQL query string + */ + public void setQueryString(String queryString) { + this.queryString = queryString; + } + + /** + * @param queryProvider Hibernate query provider + */ + public void setQueryProvider(HibernateQueryProvider queryProvider) { + this.queryProvider = queryProvider; + } + + /** + * 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(statefulSession == null && statelessSession == null, + "The useStatelessSession flag can only be set before a session is initialized."); + this.useStatelessSession = useStatelessSession; + } + + /** + * The parameter values to apply to a query (map of name:value). + * + * @param parameterValues the parameter values to set + */ + public void setParameterValues(Map parameterValues) { + this.parameterValues = parameterValues; + } + + /** + * @param sessionFactory hibernate session factory + */ + public void setSessionFactory(SessionFactory sessionFactory) { + this.sessionFactory = sessionFactory; + } + + public void afterPropertiesSet() throws Exception { + + Assert.state(sessionFactory != null, "A SessionFactory must be provided"); + Assert.state(fetchSize >= 0, "fetchSize must not be negative"); + + if (queryProvider == null) { + Assert.notNull(sessionFactory, "session factory must be set"); + Assert.state(StringUtils.hasText(queryString) ^ StringUtils.hasText(queryName), + "queryString or queryName must be set"); + } + // making sure that the appropriate (Hibernate) query provider is set + else { + Assert.state(queryProvider instanceof AbstractHibernateQueryProvider, + "Hibernate query provider must be set"); + } + + } + + /** + * Get a cursor over all of the results, with the forward-only flag set. + * + * @return a forward-only {@link ScrollableResults} + */ + public ScrollableResults getForwardOnlyCursor() { + Query query = createQuery(); + if (parameterValues != null) { + query.setProperties(parameterValues); + } + return query.setFetchSize(fetchSize).scroll(ScrollMode.FORWARD_ONLY); + } + + /** + * Open appropriate type of hibernate session and create the query. + */ + private Query createQuery() { + + if (useStatelessSession) { + statelessSession = sessionFactory.openStatelessSession(); + if (queryProvider != null) { + queryProvider.setStatelessSession(statelessSession); + } + else { + if (StringUtils.hasText(queryName)) { + return statelessSession.getNamedQuery(queryName); + } + else { + return statelessSession.createQuery(queryString); + } + } + } + else { + statefulSession = sessionFactory.openSession(); + if (queryProvider != null) { + queryProvider.setSession(statefulSession); + } + else { + if (StringUtils.hasText(queryName)) { + return statefulSession.getNamedQuery(queryName); + } + else { + return statefulSession.createQuery(queryString); + } + } + } + + // if queryProvider is set use it to create a query + return queryProvider.createQuery(); + + } + + /** + * Scroll through the results up to the item specified. + * + * @param cursor the results to scroll over + */ + public void jumpToItem(ScrollableResults cursor, int itemIndex) { + int flushSize = Math.max(fetchSize, 100); + for (int i = 0; i < itemIndex; i++) { + cursor.next(); + if (i % flushSize == 0 && !useStatelessSession) { + statefulSession.clear(); // Clears in-memory cache + } + } + } + + /** + * Close the open session (stateful or otherwise). + */ + public void close() { + if (statelessSession != null) { + statelessSession.close(); + } + if (statefulSession != null) { + statefulSession.close(); + } + } + + /** + * @param page the page to read + * @param pageSize the size of the page + * @return a collection of items + */ + public Collection readPage(int page, int pageSize) { + + clear(); + + Query query = createQuery(); + if (parameterValues != null) { + query.setProperties(parameterValues); + } + @SuppressWarnings("unchecked") + List result = query.setFetchSize(fetchSize).setFirstResult(page * pageSize).setMaxResults(pageSize).list(); + return result; + + } + + /** + * Clear the session if stateful. + */ + public void clear() { + if (statefulSession!=null) { + statefulSession.clear(); + } + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernatePagingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernatePagingItemReader.java new file mode 100644 index 000000000..9af26044e --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernatePagingItemReader.java @@ -0,0 +1,163 @@ +/* + * 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.database; + +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.hibernate.Session; +import org.hibernate.SessionFactory; +import org.hibernate.StatelessSession; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStream; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.util.ClassUtils; + +/** + * {@link ItemReader} for reading database records built on top of Hibernate. It + * executes the HQL query when initialized iterates over the result set as + * {@link #read()} method is called, returning an object corresponding to + * current row. The query can be set directly using + * {@link #setQueryString(String)} or a named query can be used by + * {@link #setQueryName(String)}. + * + *

+ * The reader 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 in + * the {@link #update(ExecutionContext)} method without being flushed (no data + * modifications are expected). + *

+ * + *

+ * The implementation is thread-safe in between calls to + * {@link #open(ExecutionContext)}, but remember to use + * saveState=false if used in a multi-threaded client (no restart + * available). + *

+ * + * @author Dave Syer + * + * @since 2.1 + */ +public class HibernatePagingItemReader extends AbstractPagingItemReader implements ItemStream, InitializingBean { + + private HibernateItemReaderHelper helper = new HibernateItemReaderHelper(); + + public HibernatePagingItemReader() { + setName(ClassUtils.getShortName(HibernatePagingItemReader.class)); + } + + public void afterPropertiesSet() throws Exception { + super.afterPropertiesSet(); + helper.afterPropertiesSet(); + } + + /** + * The parameter values to apply to a query (map of name:value). + * + * @param parameterValues the parameter values to set + */ + public void setParameterValues(Map parameterValues) { + helper.setParameterValues(parameterValues); + } + + /** + * A query name for an externalized query. Either this or the { + * {@link #setQueryString(String) query string} or the { + * {@link #setQueryProvider(HibernateQueryProvider) query provider} should + * be set. + * + * @param queryName name of a hibernate named query + */ + public void setQueryName(String queryName) { + helper.setQueryName(queryName); + } + + /** + * A query provider. Either this or the {{@link #setQueryString(String) + * query string} or the {{@link #setQueryName(String) query name} should be + * set. + * + * @param queryProvider Hibernate query provider + */ + public void setQueryProvider(HibernateQueryProvider queryProvider) { + helper.setQueryProvider(queryProvider); + } + + /** + * A query string in HQL. Either this or the { + * {@link #setQueryProvider(HibernateQueryProvider) query provider} or the { + * {@link #setQueryName(String) query name} should be set. + * + * @param queryString HQL query string + */ + public void setQueryString(String queryString) { + helper.setQueryString(queryString); + } + + /** + * The Hibernate SessionFactory to use the create a session. + * + * @param sessionFactory the {@link SessionFactory} to set + */ + public void setSessionFactory(SessionFactory sessionFactory) { + helper.setSessionFactory(sessionFactory); + } + + /** + * 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) { + helper.setUseStatelessSession(useStatelessSession); + } + + @Override + protected void doOpen() throws Exception { + super.doOpen(); + } + + @Override + protected void doReadPage() { + + if (results == null) { + results = new CopyOnWriteArrayList(); + } + else { + results.clear(); + } + + results.addAll(helper.readPage(getPage(), getPageSize())); + + } + + @Override + protected void doJumpToPage(int itemIndex) { + } + + @Override + protected void doClose() throws Exception { + helper.close(); + super.doClose(); + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/HibernatePagingItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/HibernatePagingItemReaderIntegrationTests.java new file mode 100644 index 000000000..19f000985 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/HibernatePagingItemReaderIntegrationTests.java @@ -0,0 +1,53 @@ +package org.springframework.batch.item.database; + +import org.hibernate.SessionFactory; +import org.hibernate.StatelessSession; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.sample.Foo; +import org.springframework.core.io.ClassPathResource; +import org.springframework.core.io.Resource; +import org.springframework.orm.hibernate3.LocalSessionFactoryBean; + +/** + * Tests for {@link HibernateCursorItemReader} using {@link StatelessSession}. + * + * @author Robert Kasanicky + * @author Dave Syer + */ +public class HibernatePagingItemReaderIntegrationTests extends + AbstractDataSourceItemReaderIntegrationTests { + + protected ItemReader createItemReader() throws Exception { + + LocalSessionFactoryBean factoryBean = new LocalSessionFactoryBean(); + factoryBean.setDataSource(dataSource); + factoryBean.setMappingLocations(new Resource[] { new ClassPathResource("Foo.hbm.xml", getClass()) }); + customizeSessionFactory(factoryBean); + factoryBean.afterPropertiesSet(); + + SessionFactory sessionFactory = (SessionFactory) factoryBean.getObject(); + + HibernatePagingItemReader hibernateReader = new HibernatePagingItemReader(); + setQuery(hibernateReader); + hibernateReader.setPageSize(2); + hibernateReader.setSessionFactory(sessionFactory); + hibernateReader.setUseStatelessSession(isUseStatelessSession()); + hibernateReader.afterPropertiesSet(); + hibernateReader.setSaveState(true); + + return hibernateReader; + + } + + protected void customizeSessionFactory(LocalSessionFactoryBean factoryBean) { + } + + protected void setQuery(HibernatePagingItemReader reader) throws Exception { + reader.setQueryString("from Foo"); + } + + protected boolean isUseStatelessSession() { + return true; + } + +}