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

Merge conflict

Browse Source
This commit is contained in:
Michael Minella
2013-07-26 17:00:05 -05:00
parent 1061041f29 853c960bdf
commit 7e85b50616
41 changed files with 720 additions and 541 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractCursorItemReader.java
View File

@@ -394,7 +394,7 @@ implements InitializingBean {
* Execute the statement to open the cursor.
*/
@Override
protected final void doOpen() throws Exception {
protected void doOpen() throws Exception {
Assert.state(!initialized, "Stream is already initialized. Close before re-opening.");
Assert.isNull(rs, "ResultSet still open! Close before re-opening.");

4
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateItemWriter.java
View File

@@ -81,7 +81,7 @@ public class HibernateItemWriter<T> implements ItemWriter<T>, InitializingBean {
*
* @param sessionFactory session factory to be used by the writer
*/
public final void setSessionFactory(SessionFactory sessionFactory) {
public void setSessionFactory(SessionFactory sessionFactory) {
this.sessionFactory = sessionFactory;
}
@@ -101,7 +101,7 @@ public class HibernateItemWriter<T> implements ItemWriter<T>, InitializingBean {
* @see org.springframework.batch.item.ItemWriter#write(java.util.List)
*/
@Override
public final void write(List<? extends T> items) {
public void write(List<? extends T> items) {
if(sessionFactory == null) {
doWrite(hibernateTemplate, items);
hibernateTemplate.flush();

19
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java
View File

@@ -50,7 +50,7 @@ import org.springframework.util.ClassUtils;
* specified in {@link #setPageSize(int)}. Additional pages are requested when
* needed as {@link #read()} method is called, returning an object corresponding
* to current position. On restart it uses the last sort key value to locate the
* first page to read (so it doesn't matter if the successfully processed itmes
* first page to read (so it doesn't matter if the successfully processed items
* have been removed or modified).
* </p>
*
@@ -94,6 +94,8 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
private String remainingPagesSql;
private Map<String, Object> startAfterValues;
private Map<String, Object> previousStartAfterValues;
private int fetchSize = VALUE_NOT_SET;
@@ -210,6 +212,7 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
}
else {
previousStartAfterValues = startAfterValues;
if (logger.isDebugEnabled()) {
logger.debug("SQL used for reading remaining pages: [" + remainingPagesSql + "]");
}
@@ -230,10 +233,20 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
super.update(executionContext);
if (isSaveState() && startAfterValues != null) {
executionContext.put(getExecutionContextKey(START_AFTER_VALUE), startAfterValues);
if (isSaveState()) {
if (isAtEndOfPage() && startAfterValues != null) {
// restart on next page
executionContext.put(getExecutionContextKey(START_AFTER_VALUE), startAfterValues);
} else if (previousStartAfterValues != null) {
// restart on current page
executionContext.put(getExecutionContextKey(START_AFTER_VALUE), previousStartAfterValues);
}
}
}
private boolean isAtEndOfPage() {
return getCurrentItemCount() % getPageSize() == 0;
}
@Override
@SuppressWarnings("unchecked")

2
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaItemWriter.java
View File

@@ -76,7 +76,7 @@ public class JpaItemWriter<T> implements ItemWriter<T>, InitializingBean {
* @see org.springframework.batch.item.ItemWriter#write(java.util.List)
*/
@Override
public final void write(List<? extends T> items) {
public void write(List<? extends T> items) {
EntityManager entityManager = EntityManagerFactoryUtils.getTransactionalEntityManager(entityManagerFactory);
if (entityManager == null) {
throw new DataAccessResourceFailureException("Unable to obtain a transactional EntityManager");

26
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/DerbyPagingQueryProvider.java
View File

@@ -35,18 +35,34 @@ import org.springframework.jdbc.support.JdbcUtils;
* @since 2.0
*/
public class DerbyPagingQueryProvider extends SqlWindowingPagingQueryProvider {
private String version;
private static final String MINIMAL_DERBY_VERSION = "10.4.1.3";
@Override
public void init(DataSource dataSource) throws Exception {
super.init(dataSource);
version = JdbcUtils.extractDatabaseMetaData(dataSource, "getDatabaseProductVersion").toString();
if ("10.4.1.3".compareTo(version) > 0) {
throw new InvalidDataAccessResourceUsageException("Apache Derby version " + version + " is not supported by this class, Only version 10.4.1.3 or later is supported");
String version = JdbcUtils.extractDatabaseMetaData(dataSource, "getDatabaseProductVersion").toString();
if (!isDerbyVersionSupported(version)) {
throw new InvalidDataAccessResourceUsageException("Apache Derby version " + version + " is not supported by this class, Only version " + MINIMAL_DERBY_VERSION + " or later is supported");
}
}
// derby version numbering is M.m.f.p [ {alpha|beta} ] see http://db.apache.org/derby/papers/versionupgrade.html#Basic+Numbering+Scheme
private boolean isDerbyVersionSupported(String version) {
String[] minimalVersionParts = MINIMAL_DERBY_VERSION.split("\\.");
String[] versionParts = version.split("[\\. ]");
for (int i = 0; i < minimalVersionParts.length; i++) {
int minimalVersionPart = Integer.valueOf(minimalVersionParts[i]);
int versionPart = Integer.valueOf(versionParts[i]);
if (versionPart < minimalVersionPart) {
return false;
} else if (versionPart > minimalVersionPart) {
return true;
}
}
return true;
}
@Override
protected String getOrderedQueryAlias() {
return "TMP_ORDERED";

3
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java
View File

@@ -614,6 +614,7 @@ InitializingBean {
});
writer.setEncoding(encoding);
writer.setForceSync(forceSync);
return writer;
}
else {
@@ -627,7 +628,7 @@ InitializingBean {
}
};
return new BufferedWriter(writer);
return writer;
}
}
catch (UnsupportedCharsetException ucse) {

2
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/transform/DelimitedLineTokenizer.java
View File

@@ -117,7 +117,7 @@ public class DelimitedLineTokenizer extends AbstractLineTokenizer {
*
* @see #DEFAULT_QUOTE_CHARACTER
*/
public final void setQuoteCharacter(char quoteCharacter) {
public void setQuoteCharacter(char quoteCharacter) {
this.quoteCharacter = quoteCharacter;
this.quoteString = "" + quoteCharacter;
}

2
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemCountingItemStreamItemReader.java
View File

@@ -75,7 +75,7 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> extends Abstra
}
@Override
public final T read() throws Exception, UnexpectedInputException, ParseException {
public T read() throws Exception, UnexpectedInputException, ParseException {
if (currentItemCount >= maxItemCount) {
return null;
}

26
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemProcessor.java
View File

@@ -28,37 +28,47 @@ import org.springframework.util.Assert;
* transformation is the entry value of the next).<br/>
* <br/>
*
* Note the user is responsible for injecting a chain of {@link ItemProcessor} s
* Note the user is responsible for injecting a chain of {@link ItemProcessor}s
* that conforms to declared input and output types.
*
* @author Robert Kasanicky
*/
public class CompositeItemProcessor<I, O> implements ItemProcessor<I, O>, InitializingBean {
private List<ItemProcessor<Object, Object>> delegates;
private List<? extends ItemProcessor<?, ?>> delegates;
@Override
@Override
@SuppressWarnings("unchecked")
public O process(I item) throws Exception {
Object result = item;
for (ItemProcessor<Object, Object> delegate : delegates) {
for (ItemProcessor<?, ?> delegate : delegates) {
if (result == null) {
return null;
}
result = delegate.process(result);
result = processItem(delegate, result);
}
return (O) result;
}
/*
* Helper method to work around wildcard capture compiler error: see http://docs.oracle.com/javase/tutorial/java/generics/capture.html
* The method process(capture#1-of ?) in the type ItemProcessor<capture#1-of ?,capture#2-of ?> is not applicable for the arguments (Object)
*/
@SuppressWarnings("unchecked")
private <T> Object processItem(ItemProcessor<T, ?> processor, Object input) throws Exception {
return processor.process((T) input);
}
@Override
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(delegates, "The 'delegates' may not be null");
Assert.notEmpty(delegates, "The 'delegates' may not be empty");
}
public void setDelegates(List<ItemProcessor<Object, Object>> delegates) {
public void setDelegates(List<? extends ItemProcessor<?, ?>> delegates) {
this.delegates = delegates;
}
}
}

28
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java
View File

@@ -395,7 +395,6 @@ ResourceAwareItemWriterItemStream<T>, InitializingBean {
/**
* Helper method for opening output source at given file position
*/
@SuppressWarnings("resource")
private void open(long position, boolean restarted) {
File file;
@@ -443,25 +442,20 @@ ResourceAwareItemWriterItemStream<T>, InitializingBean {
});
writer.setEncoding(encoding);
writer.setForceSync(forceSync);
bufferedWriter = writer;
}
else {
Writer writer = new BufferedWriter(new OutputStreamWriter(os, encoding)) {
@Override
public void flush() throws IOException {
super.flush();
if (forceSync) {
channel.force(false);
}
}
};
bufferedWriter = writer;
bufferedWriter = new BufferedWriter(new OutputStreamWriter(os, encoding));
}
delegateEventWriter = createXmlEventWriter(outputFactory, bufferedWriter);
eventWriter = new NoStartEndDocumentStreamWriter(delegateEventWriter);
initNamespaceContext(delegateEventWriter);
if (!restarted) {
startDocument(delegateEventWriter);
if (forceSync) {
channel.force(false);
}
}
}
catch (XMLStreamException xse) {
@@ -470,8 +464,10 @@ ResourceAwareItemWriterItemStream<T>, InitializingBean {
catch (UnsupportedEncodingException e) {
throw new DataAccessResourceFailureException("Unable to write to file resource: [" + resource
+ "] with encoding=[" + encoding + "]", e);
}
catch (IOException e) {
throw new DataAccessResourceFailureException("Unable to write to file resource: [" + resource + "]", e);
}
}
/**
@@ -658,7 +654,7 @@ ResourceAwareItemWriterItemStream<T>, InitializingBean {
finally {
try {
eventWriter.close();
delegateEventWriter.close();
}
catch (XMLStreamException e) {
log.error("Unable to close file resource: [" + resource + "] " + e);
@@ -716,9 +712,15 @@ ResourceAwareItemWriterItemStream<T>, InitializingBean {
}
try {
eventWriter.flush();
if (forceSync) {
channel.force(false);
}
}
catch (XMLStreamException e) {
throw new WriteFailedException("Failed to flush the events", e);
}
catch (IOException e) {
throw new WriteFailedException("Failed to flush the events", e);
}
}

6
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/stax/NoStartEndDocumentStreamWriter.java
View File

@@ -39,4 +39,10 @@ public class NoStartEndDocumentStreamWriter extends AbstractEventWriterWrapper {
wrappedEventWriter.add(event);
}
}
// prevents OXM Marshallers from closing the XMLEventWriter
@Override
public void close() throws XMLStreamException {
flush();
}
}

34
spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriter.java
View File

@@ -30,10 +30,10 @@ import org.springframework.transaction.support.TransactionSynchronizationManager
* buffer if a transaction is active. If a transaction is detected on the call
* to {@link #write(String)} the parameter is buffered and passed on to the
* underlying writer only when the transaction is committed.
*
*
* @author Dave Syer
* @author Michael Minella
*
*
*/
public class TransactionAwareBufferedWriter extends Writer {
@@ -50,11 +50,13 @@ public class TransactionAwareBufferedWriter extends Writer {
private String encoding = DEFAULT_CHARSET;
private boolean forceSync = false;
/**
* Create a new instance with the underlying file channel provided, and a callback
* to execute on close. The callback should clean up related resources like
* output streams or channels.
*
*
* @param channel channel used to do the actual file IO
* @param closeCallback callback to execute on close
*/
@@ -70,6 +72,19 @@ public class TransactionAwareBufferedWriter extends Writer {
this.encoding = encoding;
}
/**
* Flag to indicate that changes should be force-synced to disk on flush.
* Defaults to false, which means that even with a local disk changes could
* be lost if the OS crashes in between a write and a cache flush. Setting
* to true may result in slower performance for usage patterns involving
* many frequent writes.
*
* @param forceSync the flag value to set
*/
public void setForceSync(boolean forceSync) {
this.forceSync = forceSync;
}
/**
* @return
*/
@@ -108,6 +123,9 @@ public class TransactionAwareBufferedWriter extends Writer {
if(bytesWritten != bufferLength) {
throw new IOException("All bytes to be written were not successfully written");
}
if (forceSync) {
channel.force(false);
}
if (TransactionSynchronizationManager.hasResource(closeKey)) {
closeCallback.run();
}
@@ -134,7 +152,7 @@ public class TransactionAwareBufferedWriter extends Writer {
/**
* Convenience method for clients to determine if there is any unflushed
* data.
*
*
* @return the current size (in bytes) of unflushed buffered data
*/
public long getBufferSize() {
@@ -157,7 +175,7 @@ public class TransactionAwareBufferedWriter extends Writer {
/*
* (non-Javadoc)
*
*
* @see java.io.Writer#close()
*/
@Override
@@ -173,19 +191,19 @@ public class TransactionAwareBufferedWriter extends Writer {
/*
* (non-Javadoc)
*
*
* @see java.io.Writer#flush()
*/
@Override
public void flush() throws IOException {
if (!transactionActive()) {
if (!transactionActive() && forceSync) {
channel.force(false);
}
}
/*
* (non-Javadoc)
*
*
* @see java.io.Writer#write(char[], int, int)
*/
@Override