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 dfdfa5837..c0b47f516 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 { @@ -581,24 +582,15 @@ 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(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/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 7a0665ed6..003ee1c77 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 @@ -419,7 +420,6 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen }; if (transactional) { 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 4ec058b89..24ca2ee4f 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 @@ -25,7 +25,7 @@ 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. @@ -44,22 +44,20 @@ public class TransactionAwareBufferedWriter extends Writer { private final String closeKey; -// private Writer writer; private FileChannel channel; private final Runnable closeCallback; /** - * 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(FileChannel channel, Runnable closeCallback) { super(); -// this.writer = writer; this.channel = channel; this.closeCallback = closeCallback; this.bufferKey = BUFFER_KEY_PREFIX + "." + hashCode(); @@ -84,7 +82,6 @@ public class TransactionAwareBufferedWriter extends Writer { @Override public void beforeCommit(boolean readOnly) { try { - System.out.println("***************************** TransactionSynchronization Thread:" + Thread.currentThread().getId() + "|" + Thread.currentThread().getName()); if(!readOnly) { complete(); } @@ -101,14 +98,10 @@ public class TransactionAwareBufferedWriter extends Writer { 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!!!!"); + throw new IOException("All bytes to be written were not successfully written"); } -// writer.write(buffer.toString()); -// writer.flush(); if (TransactionSynchronizationManager.hasResource(closeKey)) { -// writer.close(); closeCallback.run(); } } @@ -164,7 +157,6 @@ public class TransactionAwareBufferedWriter extends Writer { } return; } -// writer.close(); closeCallback.run(); } @@ -177,7 +169,6 @@ public class TransactionAwareBufferedWriter extends Writer { public void flush() throws IOException { if (!transactionActive()) { channel.force(false); -// writer.flush(); } } @@ -195,7 +186,6 @@ public class TransactionAwareBufferedWriter extends Writer { if(bytesWritten != len) { throw new IOException("Unable to write all data. Bytes to write: " + len + ". Bytes written: " + bytesWritten); } -// writer.write(cbuf, off, len); return; } 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 eefb7c510..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 @@ -15,19 +15,16 @@ */ 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; @@ -46,8 +43,6 @@ import org.springframework.transaction.support.TransactionTemplate; */ public class TransactionAwareBufferedWriterTests { - private Writer stringWriter = new StringWriter(); - private FileChannel fileChannel; private TransactionAwareBufferedWriter writer; @@ -71,8 +66,6 @@ public class TransactionAwareBufferedWriterTests { private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager(); - private boolean flushed = false; - /** * Test method for * {@link org.springframework.batch.support.transaction.TransactionAwareBufferedWriter#write(java.lang.String)} @@ -96,13 +89,6 @@ public class TransactionAwareBufferedWriterTests { 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(); @@ -117,9 +103,10 @@ public class TransactionAwareBufferedWriterTests { @Test public void testCloseOutsideTransaction() throws Exception { - Capture bb = new Capture(); - expect(fileChannel.write(capture(bb))).andReturn(3); - expect(fileChannel.write(capture(bb))).andReturn(1); + 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"); @@ -127,39 +114,16 @@ public class TransactionAwareBufferedWriterTests { 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); + 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(fileChannel, new Runnable() { - public void run() { - } - }); new TransactionTemplate(transactionManager).execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { try { @@ -169,35 +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 - @SuppressWarnings({"unchecked", "rawtypes"}) - 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 { @@ -210,6 +160,33 @@ 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 @@ -224,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 @@ -246,29 +223,8 @@ public class TransactionAwareBufferedWriterTests { @Test @SuppressWarnings({"unchecked", "rawtypes"}) public void testExceptionOnFlush() throws Exception { - final Writer badWriter = new Writer() { - - @Override - public void write(char[] cbuf, int off, int len) throws IOException { - } - - @Override - public void flush() throws IOException { - throw new IOException("This should be bubbled"); - } - - @Override - public void close() throws IOException { - } - }; writer = new TransactionAwareBufferedWriter(fileChannel, new Runnable() { public void run() { - try { - badWriter.append("c"); - } - catch (IOException e) { - throw new IllegalStateException(e); - } } }); @@ -281,7 +237,6 @@ public class TransactionAwareBufferedWriterTests { catch (IOException e) { throw new IllegalStateException("Unexpected IOException", e); } - assertEquals("", stringWriter.toString()); return null; } }); @@ -291,4 +246,11 @@ public class TransactionAwareBufferedWriterTests { 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; + } }