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

BATCH-354: added an ItemWriter implementation for JPA, general cleanup, added javax.persistence to OSGi manifest template

Browse Source
This commit is contained in:
trisberg
2008-08-03 06:20:41 +00:00
parent b93903602b
commit 70ef31d9b9
4 changed files with 311 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-batch-infrastructure/pom.xml
View File

@@ -80,6 +80,7 @@
org.springframework.xml.*;version="[1.0.0, 2.0.0)";resolution:=optional,
org.springframework.batch.*,
org.springframework.*;version="[2.0.0, 3.0.0)",
javax.persistence.*,
javax.sql.*,
javax.xml.namespace.*,
javax.xml.transform.*

109
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaAwareItemWriter.java Normal file
View File

@@ -0,0 +1,109 @@
package org.springframework.batch.item.database;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.ClearFailedException;
import org.springframework.util.Assert;
import org.springframework.orm.jpa.EntityManagerFactoryUtils;
import org.springframework.dao.DataAccessResourceFailureException;
import javax.persistence.EntityManager;
import javax.persistence.EntityManagerFactory;
/**
* {@link org.springframework.batch.item.ItemWriter} that is aware of the JPA EntityManagerFactory and can
* take some responsibilities to do with chunk boundaries away from a less smart
* {@link org.springframework.batch.item.ItemWriter} (the delegate). A delegate is required, and will be used
* to do the actual writing of the item.<br/>
*
* It is required that {@link #write(Object)} is called inside a transaction,
* and that {@link #flush()} is then subsequently called before the transaction
* commits, or {@link #clear()} before it rolls back.<br/>
*
* The reader must be configured with an {@link javax.persistence.EntityManagerFactory} that is capable
* of participating in Spring managed transactions.
*
* The writer is thread safe after its properties are set (normal singleton
* behaviour), so it can be used to write in multiple concurrent transactions.
* Note, however, that the set of failed items is stored in a collection
* internally, and this collection is never cleared, so it is not a great idea
* to go on using the writer indefinitely. Normally it would be used for the
* duration of a batch job and then discarded.
*
* @author Dave Syer
* @author Thomas Risberg
*
*/
public class JpaAwareItemWriter<T> extends AbstractTransactionalResourceItemWriter<T> implements InitializingBean {
/**
* Key for items processed in the current transaction {@link org.springframework.batch.repeat.RepeatContext}.
*/
private static final String ITEMS_PROCESSED = JpaAwareItemWriter.class.getName() + ".ITEMS_PROCESSED";
private ItemWriter<? super T> delegate;
private EntityManagerFactory entityManagerFactory;
/**
* Public setter for the {@link org.springframework.batch.item.ItemWriter} property.
*
* @param delegate the delegate to set
*/
public void setDelegate(ItemWriter<? super T> delegate) {
this.delegate = delegate;
}
/**
* Set the EntityManager to be used internally.
*
* @param entityManagerFactory the entityManagerFactory to set
*/
public void setEntityManagerFactory(EntityManagerFactory entityManagerFactory) {
this.entityManagerFactory = entityManagerFactory;
}
/**
* Check mandatory properties - there must be a delegate and entityManagerFactory.
*/
public void afterPropertiesSet() throws Exception {
Assert.notNull(delegate, "An ItemWriter to be used as a delegate is required.");
Assert.notNull(entityManagerFactory, "An EntityManagerFactory is required");
}
/**
* Delegate to subclass and flush the EntityManager.
*/
protected void doFlush() {
delegate.flush();
EntityManager entityManager =
EntityManagerFactoryUtils.getTransactionalEntityManager(entityManagerFactory);
if (entityManager == null) {
throw new DataAccessResourceFailureException("Unable to obtain a transactional EntityManager");
}
entityManager.flush();
entityManager.clear();
}
/**
* Call the delegate clear() method, and then clear the EntityManager.
*/
protected void doClear() throws ClearFailedException {
delegate.clear();
EntityManager entityManager =
EntityManagerFactoryUtils.getTransactionalEntityManager(entityManagerFactory);
if (entityManager == null) {
throw new DataAccessResourceFailureException("Unable to obtain a transactional EntityManager");
}
entityManager.clear();
}
protected String getResourceKey() {
return ITEMS_PROCESSED;
}
protected void doWrite(T item) throws Exception {
delegate.write(item);
}
}

24
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java
View File

@@ -1,3 +1,19 @@
/*
* Copyright 2006-2008 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 org.springframework.batch.item.support.AbstractBufferedItemReaderItemStream;
@@ -24,8 +40,14 @@ import java.util.List;
* are requested when needed as {@link #read()} method is called, returning
* an object corresponding to current position.
*
* The performance of the paging depends on the JPA implementation and its use of database specific
* features to limit the number of returned rows.
*
* Setting a fairly large page size and using a commit interval that matches the page size should provide
* better performance.
*
* The reader must be configured with an {@link javax.persistence.EntityManagerFactory} that is capable
* of participating in SPring managed transactions.
* of participating in Spring managed transactions.
*
* The implementation is *not* thread-safe.
*

178
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JpaAwareItemWriterTests.java Normal file
View File

@@ -0,0 +1,178 @@
/*
* Copyright 2006-2008 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 static org.easymock.EasyMock.*;
import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.Test;
import org.springframework.orm.jpa.EntityManagerHolder;
import org.springframework.batch.item.ItemWriter;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import javax.persistence.EntityManagerFactory;
import javax.persistence.EntityManager;
import java.util.List;
import java.util.ArrayList;
/**
* @author Thomas Risberg
*
*/
public class JpaAwareItemWriterTests {
JpaAwareItemWriter<Object> writer = new JpaAwareItemWriter<Object>();
ItemWriter<Object> delegate;
EntityManagerFactory emf;
final List<Object> list = new ArrayList<Object>();
@Before
@SuppressWarnings({"unchecked"})
public void setUp() throws Exception {
if (TransactionSynchronizationManager.isSynchronizationActive()) {
TransactionSynchronizationManager.clearSynchronization();
}
delegate = createMock("delegate", ItemWriter.class);
writer.setDelegate(delegate);
emf = createMock("emf", EntityManagerFactory.class);
writer.setEntityManagerFactory(emf);
}
@Test
public void testAfterPropertiesSet() throws Exception {
writer = new JpaAwareItemWriter<Object>();
try {
writer.afterPropertiesSet();
fail("Expected IllegalArgumentException");
} catch (IllegalArgumentException e) {
// expected
assertTrue("Wrong message for exception: " + e.getMessage(), e
.getMessage().indexOf("delegate") >= 0);
}
writer.setDelegate(delegate);
try {
writer.afterPropertiesSet();
fail("Expected IllegalArgumentException");
} catch (IllegalArgumentException e) {
// expected
assertTrue("Wrong message for exception: " + e.getMessage(), e
.getMessage().indexOf("EntityManagerFactory") >= 0);
}
}
@Test
public void testWrite() throws Exception {
delegate.write("foo");
replay(delegate);
writer.write("foo");
verify(delegate);
}
@Test
public void testFlushWithFailure() throws Exception{
final RuntimeException ex = new RuntimeException("bar");
EntityManager em = createMock("em", EntityManager.class);
em.joinTransaction();
em.flush();
expectLastCall().andThrow(ex);
replay(em);
expect(emf.createEntityManager()).andReturn(em);
replay(emf);
TransactionSynchronizationManager.bindResource(emf, new EntityManagerHolder(em));
delegate.flush();
replay(delegate);
try {
writer.flush();
fail("Expected RuntimeException");
} catch (RuntimeException e) {
assertEquals("bar", e.getMessage());
}
TransactionSynchronizationManager.unbindResource(emf);
}
@Test
public void testWriteAndFlushWithFailure() throws Exception {
final RuntimeException ex = new RuntimeException("bar");
EntityManager em = createMock("em", EntityManager.class);
em.flush();
expectLastCall().andThrow(ex);
em.flush();
em.clear();
replay(em);
replay(emf);
TransactionSynchronizationManager.bindResource(emf, new EntityManagerHolder(em));
delegate.write("foo");
delegate.flush();
delegate.write("spam");
delegate.flush();
replay(delegate);
writer.write("foo");
try {
writer.flush();
fail("Expected RuntimeException");
} catch (RuntimeException e) {
assertEquals("bar", e.getMessage());
}
writer.write("spam");
writer.flush();
verify(delegate);
verify(em);
TransactionSynchronizationManager.unbindResource(emf);
}
@Test
public void testFlush() throws Exception{
EntityManager em = createMock("em", EntityManager.class);
em.flush();
em.clear();
replay(em);
replay(emf);
TransactionSynchronizationManager.bindResource(emf, new EntityManagerHolder(em));
delegate.flush();
replay(delegate);
writer.flush();
verify(delegate);
verify(em);
TransactionSynchronizationManager.unbindResource(emf);
}
@Test
public void testClear() throws Exception{
EntityManager em = createMock("em", EntityManager.class);
em.clear();
replay(em);
replay(emf);
TransactionSynchronizationManager.bindResource(emf, new EntityManagerHolder(em));
delegate.clear();
replay(delegate);
writer.clear();
verify(delegate);
verify(em);
TransactionSynchronizationManager.unbindResource(emf);
}
}