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

RESOLVED - issue BATCH-1452: Stream closed exception when combining MultiResourceItemWriter and FlatFileItemWriter

Browse Source
This commit is contained in:
dsyer
2009-11-27 09:10:25 +00:00
parent 6d59d6fc2a
commit 683b4e7caa
9 changed files with 360 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -434,25 +434,31 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
catch (IOException ioe) {
throw new ItemStreamException("Unable to close the the ItemWriter", ioe);
}
finally {
if (!transactional) {
closeStream();
}
}
}
private void closeStream() {
try {
if (fileChannel != null) {
fileChannel.close();
}
}
catch (IOException ioe) {
throw new ItemStreamException("Unable to close the the ItemWriter", ioe);
}
finally {
try {
if (fileChannel != null) {
fileChannel.close();
if (os != null) {
os.close();
}
}
catch (IOException ioe) {
throw new ItemStreamException("Unable to close the the ItemWriter", ioe);
}
finally {
try {
if (os != null) {
os.close();
}
}
catch (IOException ioe) {
throw new ItemStreamException("Unable to close the the ItemWriter", ioe);
}
}
}
}
@@ -518,7 +524,11 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
try {
Writer writer = Channels.newWriter(fileChannel, encoding);
if (transactional) {
return new TransactionAwareBufferedWriter(writer, getName());
return new TransactionAwareBufferedWriter(writer, new Runnable() {
public void run() {
closeStream();
}
});
} else {
return new BufferedWriter(writer);
}

31
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemWriter.java
View File

@@ -64,20 +64,26 @@ public class MultiResourceItemWriter<T> extends ExecutionContextUserSupport impl
private boolean saveState = true;
private boolean opened = false;
public MultiResourceItemWriter() {
setName(ClassUtils.getShortName(MultiResourceItemWriter.class));
}
public void write(List<? extends T> items) throws Exception {
if (!opened) {
delegate.open(new ExecutionContext());
opened = true;
}
delegate.write(items);
currentResourceItemCount += items.size();
if (currentResourceItemCount >= itemCountLimitPerResource) {
delegate.close();
resourceIndex++;
currentResourceItemCount = 0;
setResourceToDelegate();
delegate.open(new ExecutionContext());
opened = false;
}
delegate.write(items);
currentResourceItemCount += items.size();
}
/**
@@ -120,7 +126,9 @@ public class MultiResourceItemWriter<T> extends ExecutionContextUserSupport impl
public void close() throws ItemStreamException {
resourceIndex = 1;
currentResourceItemCount = 0;
delegate.close();
if (opened) {
delegate.close();
}
}
public void open(ExecutionContext executionContext) throws ItemStreamException {
@@ -131,14 +139,23 @@ public class MultiResourceItemWriter<T> extends ExecutionContextUserSupport impl
setResourceToDelegate();
}
catch (IOException e) {
throw new ItemStreamException("Couldn't open resource", e);
throw new ItemStreamException("Couldn't assign resource", e);
}
if (executionContext.containsKey(getKey(CURRENT_RESOURCE_ITEM_COUNT))) {
// It's a restart
delegate.open(executionContext);
}
else {
opened = false;
}
delegate.open(executionContext);
}
public void update(ExecutionContext executionContext) throws ItemStreamException {
if (saveState) {
delegate.update(executionContext);
if (opened) {
delegate.update(executionContext);
}
executionContext.putInt(getKey(CURRENT_RESOURCE_ITEM_COUNT), currentResourceItemCount);
executionContext.putInt(getKey(RESOURCE_INDEX_KEY), resourceIndex);
}

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

@@ -16,12 +16,13 @@
package org.springframework.batch.item.xml;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.io.Writer;
import java.nio.channels.FileChannel;
import java.util.List;
import java.util.Map;
@@ -65,8 +66,8 @@ import org.springframework.xml.transform.StaxResult;
* @author Robert Kasanicky
*
*/
public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implements ResourceAwareItemWriterItemStream<T>,
InitializingBean {
public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implements
ResourceAwareItemWriterItemStream<T>, InitializingBean {
private static final Log log = LogFactory.getLog(StaxEventItemWriter.class);
@@ -126,7 +127,9 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
private StaxWriterCallback footerCallback;
private TransactionAwareBufferedWriter transactionAwareBufferedWriter;
private Writer bufferedWriter;
private boolean transactional = true;
public StaxEventItemWriter() {
setName(ClassUtils.getShortName(StaxEventItemWriter.class));
@@ -165,6 +168,16 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
this.footerCallback = footerCallback;
}
/**
* Flag to indicate that writes should be deferred to the end of a
* transaction if present. Defaults to true.
*
* @param transactional the flag to set
*/
public void setTransactional(boolean transactional) {
this.transactional = transactional;
}
/**
* Get used encoding.
*
@@ -324,11 +337,20 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
// http://jira.springframework.org/browse/BATCH-761.
outputFactory.setProperty("com.ctc.wstx.automaticEndElements", Boolean.FALSE);
}
try {
transactionAwareBufferedWriter = new TransactionAwareBufferedWriter(
new OutputStreamWriter(os, encoding), getName());
delegateEventWriter = outputFactory.createXMLEventWriter(transactionAwareBufferedWriter);
if (transactional) {
bufferedWriter = new TransactionAwareBufferedWriter(new OutputStreamWriter(os, encoding),
new Runnable() {
public void run() {
closeStream();
}
});
}
else {
bufferedWriter = new BufferedWriter(new OutputStreamWriter(os, encoding));
}
delegateEventWriter = outputFactory.createXMLEventWriter(bufferedWriter);
eventWriter = new NoStartEndDocumentStreamWriter(delegateEventWriter);
if (!restarted) {
startDocument(delegateEventWriter);
@@ -390,9 +412,8 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
// writer.writeEndDocument(); <- this doesn't work after restart
// we need to write end tag of the root element manually
ByteBuffer bbuf = ByteBuffer.wrap(("</" + getRootTagName() + ">").getBytes());
try {
channel.write(bbuf);
bufferedWriter.write("</" + getRootTagName() + ">");
}
catch (IOException ioe) {
throw new DataAccessResourceFailureException("Unable to close file resource: [" + resource + "]", ioe);
@@ -433,21 +454,34 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
try {
eventWriter.close();
}
catch (XMLStreamException xse) {
log.error("Unable to close file resource: [" + resource + "] " + xse);
catch (XMLStreamException e) {
log.error("Unable to close file resource: [" + resource + "] " + e);
}
finally {
try {
channel.close();
bufferedWriter.close();
}
catch (IOException ioe) {
log.error("Unable to close file resource: [" + resource + "] " + ioe);
catch (IOException e) {
log.error("Unable to close file resource: [" + resource + "] " + e);
}
finally {
if (!transactional) {
closeStream();
}
}
}
}
}
private void closeStream() {
try {
channel.close();
}
catch (IOException ioe) {
log.error("Unable to close file resource: [" + resource + "] " + ioe);
}
}
/**
* Write the value objects and flush them to the file.
*
@@ -499,7 +533,10 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
try {
eventWriter.flush();
position = channel.position() + transactionAwareBufferedWriter.getBufferSize();
position = channel.position();
if (bufferedWriter instanceof TransactionAwareBufferedWriter) {
position += ((TransactionAwareBufferedWriter) bufferedWriter).getBufferSize();
}
}
catch (Exception e) {
throw new DataAccessResourceFailureException("Unable to write to file resource: [" + resource + "]", e);

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

@@ -22,9 +22,9 @@ import org.springframework.transaction.support.TransactionSynchronizationAdapter
import org.springframework.transaction.support.TransactionSynchronizationManager;
/**
* Wrapper for a {@link Writer} that delays actually writing to the 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
* Wrapper for a {@link Writer} that delays actually writing to or closing the
* 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
@@ -34,19 +34,30 @@ public class TransactionAwareBufferedWriter extends Writer {
private static final String BUFFER_KEY_PREFIX = TransactionAwareBufferedWriter.class.getName() + ".BUFFER_KEY";
private static final String CLOSE_KEY_PREFIX = TransactionAwareBufferedWriter.class.getName() + ".CLOSE_KEY";
private final String bufferKey;
private final String closeKey;
private Writer writer;
private final Runnable closeCallback;
/**
* Create a new instance with the underlying writer provided, and a callback
* to execute on close. The callback should clean up related resources like
* output streams or channels.
*
* @param writer actually writes to output
* @param name used to identify the transactional buffer used by this
* instance
* @param closeCallback callback to execute on close
*/
public TransactionAwareBufferedWriter(Writer writer, String name) {
public TransactionAwareBufferedWriter(Writer writer, Runnable closeCallback) {
super();
this.writer = writer;
this.bufferKey = BUFFER_KEY_PREFIX + "." + name;
this.closeCallback = closeCallback;
this.bufferKey = BUFFER_KEY_PREFIX + "." + hashCode();
this.closeKey = CLOSE_KEY_PREFIX + "." + hashCode();
}
/**
@@ -61,18 +72,34 @@ public class TransactionAwareBufferedWriter extends Writer {
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
@Override
public void afterCompletion(int status) {
StringBuffer buffer = (StringBuffer) TransactionSynchronizationManager.getResource(bufferKey);
if (status == STATUS_COMMITTED) {
complete();
}
}
private void complete() {
StringBuffer buffer = (StringBuffer) TransactionSynchronizationManager.getResource(bufferKey);
if (buffer != null) {
try {
writer.write(buffer.toString());
writer.flush();
if (TransactionSynchronizationManager.hasResource(closeKey)) {
writer.close();
closeCallback.run();
}
}
catch (IOException e) {
throw new FlushFailedException("Could not write to output buffer", e);
}
finally {
TransactionSynchronizationManager.unbindResource(bufferKey);
if (TransactionSynchronizationManager.hasResource(closeKey)) {
TransactionSynchronizationManager.unbindResource(closeKey);
}
}
}
TransactionSynchronizationManager.unbindResource(bufferKey);
}
});
}
@@ -108,7 +135,14 @@ public class TransactionAwareBufferedWriter extends Writer {
*/
@Override
public void close() throws IOException {
if (transactionActive()) {
if (getCurrentBuffer().length() > 0) {
TransactionSynchronizationManager.bindResource(closeKey, Boolean.TRUE);
}
return;
}
writer.close();
closeCallback.run();
}
/*