BATCH-1799: Intermediate commit with POC code

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

@@ -581,15 +581,24 @@ public class FlatFileItemWriter<T> 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();
 						}
 					});

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

@@ -418,7 +418,8 @@ public class StaxEventItemWriter<T> 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();
 					}

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);
-
 	}
-
 }

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<ByteBuffer> bb = new Capture<ByteBuffer>();
+		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<ByteBuffer> bb = new Capture<ByteBuffer>();
+		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<ByteBuffer> bb = new Capture<ByteBuffer>();
+		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");

spring-batch-parent/pom.xml

@@ -370,13 +370,13 @@
 			<dependency>
 				<groupId>org.easymock</groupId>
 				<artifactId>easymock</artifactId>
-				<version>2.4</version>
+				<version>3.1</version>
 				<scope>test</scope>
 			</dependency>
 			<dependency>
 				<groupId>org.easymock</groupId>
 				<artifactId>easymockclassextension</artifactId>
-				<version>2.4</version>
+				<version>3.1</version>
 				<scope>test</scope>
 			</dependency>
 			<dependency>