diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java index c5d67d659..dfdfa5837 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java @@ -581,15 +581,24 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement Writer writer = new BufferedWriter(Channels.newWriter(fileChannel, encoding)) { @Override public void flush() throws IOException { + System.out.println("***************************** Flush Thread:" + Thread.currentThread().getId() + "|" + Thread.currentThread().getName()); super.flush(); + System.out.println("++++++++++++++++++++ super.flush called"); if (forceSync) { channel.force(false); } + System.out.println("~~~~~~~~~~~~~~~~~~~~~~ force complete"); + try { + Thread.sleep(10*1000); + } catch (InterruptedException e) { + } } }; if (transactional) { - return new TransactionAwareBufferedWriter(writer, new Runnable() { + return new TransactionAwareBufferedWriter(channel, new Runnable() { +// return new TransactionAwareBufferedWriter(writer, new Runnable() { public void run() { + System.out.println("============================ closing stream"); closeStream(); } }); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java index bc96c648e..7a0665ed6 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java @@ -418,7 +418,8 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen } }; if (transactional) { - bufferedWriter = new TransactionAwareBufferedWriter(writer, new Runnable() { + bufferedWriter = new TransactionAwareBufferedWriter(channel, new Runnable() { +// bufferedWriter = new TransactionAwareBufferedWriter(writer, new Runnable() { public void run() { closeStream(); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriter.java index 1efc62214..4ec058b89 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriter.java @@ -17,6 +17,9 @@ package org.springframework.batch.support.transaction; import java.io.IOException; import java.io.Writer; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.util.Arrays; import org.springframework.transaction.support.TransactionSynchronizationAdapter; import org.springframework.transaction.support.TransactionSynchronizationManager; @@ -41,7 +44,8 @@ public class TransactionAwareBufferedWriter extends Writer { private final String closeKey; - private Writer writer; +// private Writer writer; + private FileChannel channel; private final Runnable closeCallback; @@ -53,9 +57,10 @@ public class TransactionAwareBufferedWriter extends Writer { * @param writer actually writes to output * @param closeCallback callback to execute on close */ - public TransactionAwareBufferedWriter(Writer writer, Runnable closeCallback) { + public TransactionAwareBufferedWriter(FileChannel channel, Runnable closeCallback) { super(); - this.writer = writer; +// this.writer = writer; + this.channel = channel; this.closeCallback = closeCallback; this.bufferKey = BUFFER_KEY_PREFIX + "." + hashCode(); this.closeKey = CLOSE_KEY_PREFIX + "." + hashCode(); @@ -77,9 +82,12 @@ public class TransactionAwareBufferedWriter extends Writer { } @Override - public void afterCommit() { + public void beforeCommit(boolean readOnly) { try { - complete(); + System.out.println("***************************** TransactionSynchronization Thread:" + Thread.currentThread().getId() + "|" + Thread.currentThread().getName()); + if(!readOnly) { + complete(); + } } catch (IOException e) { throw new FlushFailedException("Could not write to output buffer", e); @@ -89,10 +97,18 @@ public class TransactionAwareBufferedWriter extends Writer { private void complete() throws IOException { StringBuffer buffer = (StringBuffer) TransactionSynchronizationManager.getResource(bufferKey); if (buffer != null) { - writer.write(buffer.toString()); - writer.flush(); + String string = buffer.toString(); + int bufferLength = string.length(); + ByteBuffer bb = ByteBuffer.wrap(string.getBytes()); + int bytesWritten = channel.write(bb); + System.out.println("$$$$$$$$$$$$$$$$$$$$$$$$$ length = " + bufferLength + " written = " + bytesWritten); + if(bytesWritten != bufferLength) { + throw new RuntimeException("Unable to write all of the crap you wanted!!!!"); + } +// writer.write(buffer.toString()); +// writer.flush(); if (TransactionSynchronizationManager.hasResource(closeKey)) { - writer.close(); +// writer.close(); closeCallback.run(); } } @@ -148,7 +164,7 @@ public class TransactionAwareBufferedWriter extends Writer { } return; } - writer.close(); +// writer.close(); closeCallback.run(); } @@ -160,7 +176,8 @@ public class TransactionAwareBufferedWriter extends Writer { @Override public void flush() throws IOException { if (!transactionActive()) { - writer.flush(); + channel.force(false); +// writer.flush(); } } @@ -173,13 +190,16 @@ public class TransactionAwareBufferedWriter extends Writer { public void write(char[] cbuf, int off, int len) throws IOException { if (!transactionActive()) { - writer.write(cbuf, off, len); + ByteBuffer bb = ByteBuffer.wrap(new String(Arrays.copyOfRange(cbuf, off, off + len)).getBytes()); + int bytesWritten = channel.write(bb); + if(bytesWritten != len) { + throw new IOException("Unable to write all data. Bytes to write: " + len + ". Bytes written: " + bytesWritten); + } +// writer.write(cbuf, off, len); return; } StringBuffer buffer = getCurrentBuffer(); buffer.append(cbuf, off, len); - } - } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriterTests.java index 9d3711a38..eefb7c510 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriterTests.java @@ -15,6 +15,11 @@ */ package org.springframework.batch.support.transaction; +import static org.easymock.EasyMock.capture; +import static org.easymock.EasyMock.createMock; +import static org.easymock.EasyMock.expect; +import static org.easymock.EasyMock.replay; +import static org.easymock.EasyMock.verify; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@ -23,7 +28,10 @@ import static org.junit.Assert.fail; import java.io.IOException; import java.io.StringWriter; import java.io.Writer; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import org.easymock.Capture; import org.junit.Before; import org.junit.Test; import org.springframework.transaction.PlatformTransactionManager; @@ -39,15 +47,20 @@ import org.springframework.transaction.support.TransactionTemplate; public class TransactionAwareBufferedWriterTests { private Writer stringWriter = new StringWriter(); + + private FileChannel fileChannel; private TransactionAwareBufferedWriter writer; @Before public void init() { - writer = new TransactionAwareBufferedWriter(stringWriter, new Runnable() { + fileChannel = createMock(FileChannel.class); + + writer = new TransactionAwareBufferedWriter(fileChannel, new Runnable() { public void run() { try { - stringWriter.append("c"); + ByteBuffer bb = ByteBuffer.wrap("c".getBytes()); + fileChannel.write(bb); } catch (IOException e) { throw new IllegalStateException(e); @@ -68,23 +81,61 @@ public class TransactionAwareBufferedWriterTests { */ @Test public void testWriteOutsideTransaction() throws Exception { + Capture bb = new Capture(); + expect(fileChannel.write(capture(bb))).andReturn(3); + fileChannel.force(false); + replay(fileChannel); + writer.write("foo"); writer.flush(); // Not closed yet - assertEquals("foo", stringWriter.toString()); + + String s = getStringFromByteBuffer(bb.getValue()); + + verify(fileChannel); + assertEquals("foo", s); + } + + private String getStringFromByteBuffer(ByteBuffer bb) { + byte[] bytearr = new byte[bb.remaining()]; + bb.get(bytearr); + String s = new String(bytearr); + return s; } @Test public void testBufferSizeOutsideTransaction() throws Exception { + Capture bb = new Capture(); + expect(fileChannel.write(capture(bb))).andReturn(3); + replay(fileChannel); + writer.write("foo"); + + verify(fileChannel); assertEquals(0, writer.getBufferSize()); } @Test public void testCloseOutsideTransaction() throws Exception { + Capture bb = new Capture(); + expect(fileChannel.write(capture(bb))).andReturn(3); + expect(fileChannel.write(capture(bb))).andReturn(1); + replay(fileChannel); + writer.write("foo"); writer.close(); - assertEquals("fooc", stringWriter.toString()); + + verify(fileChannel); + + String output = ""; + + for (ByteBuffer curBuffer : bb.getValues()) { + output = output + getStringFromByteBuffer(curBuffer); + + } +// assertEquals("foo", getStringFromByteBuffer(writeBuffer.getValue())); +// assertEquals("c", getStringFromByteBuffer(commitBuffer.getValue())); + assertEquals("fooc", output); } @Test @@ -105,7 +156,7 @@ public class TransactionAwareBufferedWriterTests { public void write(char[] cbuf, int off, int len) throws IOException { } }; - writer = new TransactionAwareBufferedWriter(mock, new Runnable() { + writer = new TransactionAwareBufferedWriter(fileChannel, new Runnable() { public void run() { } }); @@ -210,7 +261,7 @@ public class TransactionAwareBufferedWriterTests { public void close() throws IOException { } }; - writer = new TransactionAwareBufferedWriter(badWriter, new Runnable() { + writer = new TransactionAwareBufferedWriter(fileChannel, new Runnable() { public void run() { try { badWriter.append("c"); diff --git a/spring-batch-parent/pom.xml b/spring-batch-parent/pom.xml index 62ee4e5c3..34c576b14 100644 --- a/spring-batch-parent/pom.xml +++ b/spring-batch-parent/pom.xml @@ -370,13 +370,13 @@ org.easymock easymock - 2.4 + 3.1 test org.easymock easymockclassextension - 2.4 + 3.1 test