diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/AbstractSqlPagingQueryProvider.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/AbstractSqlPagingQueryProvider.java index 444cb8556..fa54047f4 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/AbstractSqlPagingQueryProvider.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/AbstractSqlPagingQueryProvider.java @@ -68,7 +68,7 @@ public abstract class AbstractSqlPagingQueryProvider implements PagingQueryProvi /** * The setter for the group by clause * - * @param SQL GROUP BY clause part of the SQL query string + * @param groupClause SQL GROUP BY clause part of the SQL query string */ public void setGroupClause(String groupClause) { if (StringUtils.hasText(groupClause)) { diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SqlPagingQueryProviderFactoryBean.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SqlPagingQueryProviderFactoryBean.java index d4779ef32..a2e383cc5 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SqlPagingQueryProviderFactoryBean.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SqlPagingQueryProviderFactoryBean.java @@ -81,7 +81,7 @@ public class SqlPagingQueryProviderFactoryBean implements FactoryBean { } /** - * @param SQL GROUP BY clause part of the SQL query string + * @param groupClause SQL GROUP BY clause part of the SQL query string */ public void setGroupClause(String groupClause) { this.groupClause = groupClause; @@ -123,7 +123,7 @@ public class SqlPagingQueryProviderFactoryBean implements FactoryBean { } /** - * @param sortKey the sortKey to set + * @param sortKeys the sortKey to set */ public void setSortKeys(Map sortKeys) { this.sortKeys = sortKeys; 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..e09ff0f22 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 @@ -1,5 +1,5 @@ /* - * Copyright 2006-2007 the original author or authors. + * Copyright 2006-2012 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. @@ -55,6 +55,7 @@ import org.springframework.util.ClassUtils; * @author Tomas Slanina * @author Robert Kasanicky * @author Dave Syer + * @author Michael Minella */ public class FlatFileItemWriter extends ExecutionContextUserSupport implements ResourceAwareItemWriterItemStream, InitializingBean { @@ -578,23 +579,27 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement private Writer getBufferedWriter(FileChannel fileChannel, String encoding) { try { final FileChannel channel = fileChannel; - Writer writer = new BufferedWriter(Channels.newWriter(fileChannel, encoding)) { - @Override - public void flush() throws IOException { - super.flush(); - if (forceSync) { - channel.force(false); - } - } - }; if (transactional) { - return new TransactionAwareBufferedWriter(writer, new Runnable() { + TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(channel, new Runnable() { public void run() { closeStream(); } }); + + writer.setEncoding(encoding); + return writer; } else { + Writer writer = new BufferedWriter(Channels.newWriter(fileChannel, encoding)) { + @Override + public void flush() throws IOException { + super.flush(); + if (forceSync) { + channel.force(false); + } + } + }; + return new BufferedWriter(writer); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/transform/DelimitedLineTokenizer.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/transform/DelimitedLineTokenizer.java index fd53c4eb2..077911bab 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/transform/DelimitedLineTokenizer.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/transform/DelimitedLineTokenizer.java @@ -64,7 +64,7 @@ public class DelimitedLineTokenizer extends AbstractLineTokenizer { * Create a new instance of the {@link DelimitedLineTokenizer} class for the * common case where the delimiter is a {@link #DELIMITER_COMMA comma}. * - * @see #DelimitedLineTokenizer(char) + * @see #DelimitedLineTokenizer(String) * @see #DELIMITER_COMMA */ public DelimitedLineTokenizer() { 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..e427c94c0 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 @@ -1,5 +1,5 @@ /* - * Copyright 2006-2007 the original author or authors. + * Copyright 2006-2012 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. @@ -66,6 +66,7 @@ import org.springframework.util.StringUtils; * * @author Peter Zozom * @author Robert Kasanicky + * @author Michael Minella * */ public class StaxEventItemWriter extends ExecutionContextUserSupport implements @@ -408,23 +409,26 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen try { final FileChannel channel = fileChannel; - Writer writer = new BufferedWriter(new OutputStreamWriter(os, encoding)) { - @Override - public void flush() throws IOException { - super.flush(); - if (forceSync) { - channel.force(false); - } - } - }; if (transactional) { - bufferedWriter = new TransactionAwareBufferedWriter(writer, new Runnable() { + TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(channel, new Runnable() { public void run() { closeStream(); } }); + + writer.setEncoding(encoding); + 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; } delegateEventWriter = createXmlEventWriter(outputFactory, bufferedWriter); 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 d5de7014f..73fa5fd6d 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 @@ -1,5 +1,5 @@ /* - * Copyright 2006-2007 the original author or authors. + * Copyright 2006-2012 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. @@ -17,17 +17,21 @@ 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; /** - * Wrapper for a {@link Writer} that delays actually writing to or closing the + * Wrapper for a {@link FileChannel} 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 + * @author Michael Minella * */ public class TransactionAwareBufferedWriter extends Writer { @@ -40,26 +44,35 @@ public class TransactionAwareBufferedWriter extends Writer { private final String closeKey; - private Writer writer; + private FileChannel channel; private final Runnable closeCallback; - + + // default encoding for writing to output files - set to UTF-8. + private static final String DEFAULT_CHARSET = "UTF-8"; + + private String encoding = DEFAULT_CHARSET; + /** - * Create a new instance with the underlying writer provided, and a callback + * 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 writer actually writes to output + * @param channel channel used to do the actuall file IO * @param closeCallback callback to execute on close */ - public TransactionAwareBufferedWriter(Writer writer, Runnable closeCallback) { + public TransactionAwareBufferedWriter(FileChannel channel, Runnable closeCallback) { super(); - this.writer = writer; + this.channel = channel; this.closeCallback = closeCallback; this.bufferKey = BUFFER_KEY_PREFIX + "." + hashCode(); this.closeKey = CLOSE_KEY_PREFIX + "." + hashCode(); } + public void setEncoding(String encoding) { + this.encoding = encoding; + } + /** * @return */ @@ -72,26 +85,33 @@ public class TransactionAwareBufferedWriter extends Writer { TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() { @Override public void afterCompletion(int status) { + clear(); + } + + @Override + public void beforeCommit(boolean readOnly) { try { - if (status == STATUS_COMMITTED) { + if(!readOnly) { complete(); } } catch (IOException e) { throw new FlushFailedException("Could not write to output buffer", e); } - finally { - clear(); - } } private void complete() throws IOException { StringBuffer buffer = (StringBuffer) TransactionSynchronizationManager.getResource(bufferKey); if (buffer != null) { - writer.write(buffer.toString()); - writer.flush(); + String string = buffer.toString(); + byte[] bytes = string.getBytes(encoding); + int bufferLength = bytes.length; + ByteBuffer bb = ByteBuffer.wrap(bytes); + int bytesWritten = channel.write(bb); + if(bytesWritten != bufferLength) { + throw new IOException("All bytes to be written were not successfully written"); + } if (TransactionSynchronizationManager.hasResource(closeKey)) { - writer.close(); closeCallback.run(); } } @@ -147,7 +167,6 @@ public class TransactionAwareBufferedWriter extends Writer { } return; } - writer.close(); closeCallback.run(); } @@ -159,7 +178,7 @@ public class TransactionAwareBufferedWriter extends Writer { @Override public void flush() throws IOException { if (!transactionActive()) { - writer.flush(); + channel.force(false); } } @@ -172,13 +191,17 @@ public class TransactionAwareBufferedWriter extends Writer { public void write(char[] cbuf, int off, int len) throws IOException { if (!transactionActive()) { - writer.write(cbuf, off, len); + byte[] bytes = new String(Arrays.copyOfRange(cbuf, off, off + len)).getBytes(encoding); + int length = bytes.length; + ByteBuffer bb = ByteBuffer.wrap(bytes); + int bytesWritten = channel.write(bb); + if(bytesWritten != length) { + throw new IOException("Unable to write all data. Bytes to write: " + len + ". Bytes written: " + bytesWritten); + } 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 de834facc..e9db8e161 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 @@ -1,5 +1,5 @@ /* - * Copyright 2006-2007 the original author or authors. + * Copyright 2006-2012 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. @@ -15,14 +15,21 @@ */ package org.springframework.batch.support.transaction; +import static org.easymock.EasyMock.anyObject; +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; +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; import org.springframework.transaction.TransactionStatus; @@ -31,27 +38,34 @@ import org.springframework.transaction.support.TransactionTemplate; /** * @author Dave Syer + * @author Michael Minella * */ public class TransactionAwareBufferedWriterTests { - private Writer stringWriter = new StringWriter(); + private FileChannel fileChannel; - private TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(stringWriter, new Runnable() { - public void run() { - try { - stringWriter.append("c"); + private TransactionAwareBufferedWriter writer; + + @Before + public void init() { + fileChannel = createMock(FileChannel.class); + + writer = new TransactionAwareBufferedWriter(fileChannel, new Runnable() { + public void run() { + try { + ByteBuffer bb = ByteBuffer.wrap("c".getBytes()); + fileChannel.write(bb); + } + catch (IOException e) { + throw new IllegalStateException(e); + } } - catch (IOException e) { - throw new IllegalStateException(e); - } - } - }); + }); + } private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager(); - private boolean flushed = false; - /** * Test method for * {@link org.springframework.batch.support.transaction.TransactionAwareBufferedWriter#write(java.lang.String)} @@ -60,46 +74,56 @@ 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); } @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 writeBuffer = new Capture(); + Capture commitBuffer = new Capture(); + expect(fileChannel.write(capture(writeBuffer))).andReturn(3); + expect(fileChannel.write(capture(commitBuffer))).andReturn(1); + replay(fileChannel); + writer.write("foo"); writer.close(); - assertEquals("fooc", stringWriter.toString()); + + verify(fileChannel); + + assertEquals("foo", getStringFromByteBuffer(writeBuffer.getValue())); + assertEquals("c", getStringFromByteBuffer(commitBuffer.getValue())); } @Test + @SuppressWarnings({"unchecked", "rawtypes"}) public void testFlushInTransaction() throws Exception { - Writer mock = new Writer() { - @Override - public void close() throws IOException { - throw new UnsupportedOperationException(); - } + expect(fileChannel.write((ByteBuffer)anyObject())).andReturn(3); + replay(fileChannel); - @Override - public void flush() throws IOException { - flushed = true; - } - - @Override - public void write(char[] cbuf, int off, int len) throws IOException { - } - }; - writer = new TransactionAwareBufferedWriter(mock, new Runnable() { - public void run() { - } - }); new TransactionTemplate(transactionManager).execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { try { @@ -109,33 +133,21 @@ public class TransactionAwareBufferedWriterTests { catch (IOException e) { throw new IllegalStateException("Unexpected IOException", e); } - assertFalse(flushed); + assertEquals(3, writer.getBufferSize()); return null; } }); - assertTrue(flushed); + + verify(fileChannel); } @Test + @SuppressWarnings({"unchecked", "rawtypes"}) public void testWriteWithCommit() throws Exception { - new TransactionTemplate(transactionManager).execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus status) { - try { - writer.write("foo"); - } - catch (IOException e) { - throw new IllegalStateException("Unexpected IOException", e); - } - assertEquals("", stringWriter.toString()); - return null; - } - }); - // Not closed in transaction - assertEquals("foo", stringWriter.toString()); - } - - @Test - public void tesBufferSizeInTransaction() throws Exception { + Capture bb = new Capture(); + expect(fileChannel.write(capture(bb))).andReturn(3); + replay(fileChannel); + new TransactionTemplate(transactionManager).execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { try { @@ -148,9 +160,37 @@ public class TransactionAwareBufferedWriterTests { return null; } }); + + verify(fileChannel); + assertEquals(0, writer.getBufferSize()); } @Test + @SuppressWarnings({"unchecked", "rawtypes"}) + public void testBufferSizeInTransaction() throws Exception { + Capture bb = new Capture(); + expect(fileChannel.write(capture(bb))).andReturn(3); + replay(fileChannel); + + new TransactionTemplate(transactionManager).execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + try { + writer.write("foo"); + } + catch (IOException e) { + throw new IllegalStateException("Unexpected IOException", e); + } + assertEquals(3, writer.getBufferSize()); + return null; + } + }); + + verify(fileChannel); + assertEquals(0, writer.getBufferSize()); + } + + @Test + @SuppressWarnings({"unchecked", "rawtypes"}) public void testWriteWithRollback() throws Exception { try { new TransactionTemplate(transactionManager).execute(new TransactionCallback() { @@ -161,17 +201,17 @@ public class TransactionAwareBufferedWriterTests { catch (IOException e) { throw new IllegalStateException("Unexpected IOException", e); } - assertEquals("", stringWriter.toString()); throw new RuntimeException("Planned failure"); } }); + fail("Exception was not thrown"); } catch (RuntimeException e) { // expected String message = e.getMessage(); assertEquals("Wrong message: " + message, "Planned failure", message); } - assertEquals("", stringWriter.toString()); + assertEquals(0, writer.getBufferSize()); } @Test @@ -179,5 +219,38 @@ public class TransactionAwareBufferedWriterTests { testWriteWithRollback(); testWriteWithCommit(); } + + @Test + @SuppressWarnings({"unchecked", "rawtypes"}) + public void testExceptionOnFlush() throws Exception { + writer = new TransactionAwareBufferedWriter(fileChannel, new Runnable() { + public void run() { + } + }); + try { + new TransactionTemplate(transactionManager).execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + try { + writer.write("foo"); + } + catch (IOException e) { + throw new IllegalStateException("Unexpected IOException", e); + } + return null; + } + }); + + fail("Exception was not thrown"); + } catch (FlushFailedException ffe) { + assertEquals("Could not write to output buffer", ffe.getMessage()); + } + } + + private String getStringFromByteBuffer(ByteBuffer bb) { + byte[] bytearr = new byte[bb.remaining()]; + bb.get(bytearr); + String s = new String(bytearr); + return s; + } } 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