BATCH-365: Added flush() and clear() to ItemWriter interface. These should be removed from ItemStream after the same is done for the ItemReader interface.

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

@@ -52,6 +52,10 @@ import org.springframework.util.Assert;
  * 
  * Use {@link #write(String)} method to output a line to an item writer.
  * 
+ * <p>This class will be updated in the future to use a buffering approach
+ * to handling transactions, rather than outputting directly to the file and
+ * truncating on rollback</p>
+ * 
  * @author Waseem Malik
  * @author Tomas Slanina
  * @author Robert Kasanicky
@@ -465,23 +469,31 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
 		return true;
 	}
 
-	/*
+	/* To be deleted once interface changes are complete
 	 * (non-Javadoc)
 	 * @see org.springframework.batch.io.support.AbstractTransactionalIoSource#mark(org.springframework.batch.item.ExecutionContext)
 	 */
 	public void mark() {
-		getOutputState().mark();
+		
 	}
 
-	/*
+	/* To be deleted once interface changes are complete
 	 * (non-Javadoc)
 	 * @see org.springframework.batch.io.support.AbstractTransactionalIoSource#reset(org.springframework.batch.item.ExecutionContext)
 	 */
 	public void reset() throws ResetFailedException {
+
+	}
+
+	public void clear() throws Exception {
 		try {
 			getOutputState().reset();
 		} catch (BatchCriticalException e) {
 			throw new ResetFailedException(e);
 		}
 	}
+
+	public void flush() throws Exception {
+		getOutputState().mark();
+	}
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/io/support/HibernateAwareItemWriter.java

@@ -146,32 +146,7 @@ public class HibernateAwareItemWriter implements ItemWriter, RepeatListener, Ini
 	 * @see org.springframework.batch.repeat.RepeatListener#close(org.springframework.batch.repeat.RepeatContext)
 	 */
 	public void close(RepeatContext context) {
-		try {
-			if (delegate instanceof RepeatListener) {
-				RepeatListener interceptor = (RepeatListener) delegate;
-				interceptor.close(context);
-			}
-			flush();
-		} catch (RuntimeException e) {
-			synchronized (failed) {
-				failed.addAll(getProcessed());
-			}
-			// onError will not be called after close() by the framework so we
-			// have to do it here.
-			onError(context, e);
-			throw e;
-		}
-		unsetContext();
-	}
 
-	/**
-	 * Wrapper for Hibernate flush.
-	 */
-	private void flush() {
-		hibernateTemplate.flush();
-		// This should happen when the transaction commits anyway, but to be
-		// sure...
-		hibernateTemplate.clear();
 	}
 
 	/**
@@ -246,7 +221,7 @@ public class HibernateAwareItemWriter implements ItemWriter, RepeatListener, Ini
 	 * 
 	 * @return the context
 	 */
-	private void flushIfNecessary(Object output) {
+	private void flushIfNecessary(Object output) throws Exception{
 		RepeatContext context = (RepeatContext) TransactionSynchronizationManager.getResource(WRITER_REPEAT_CONTEXT);
 		boolean flush;
 		synchronized (failed) {
@@ -264,4 +239,34 @@ public class HibernateAwareItemWriter implements ItemWriter, RepeatListener, Ini
 
 	}
 
+	public void clear() throws Exception {
+		if(delegate != null){
+			delegate.clear();
+		}
+		hibernateTemplate.clear();
+	}
+
+	/**
+	 * Flush the Hibernate session.  The delegate flush will also be called before finishing.
+	 */
+	public void flush() throws Exception {
+		try {
+			if (delegate != null) {
+				delegate.flush();
+			}
+			hibernateTemplate.flush();
+			// This should happen when the transaction commits anyway, but to be
+			// sure...
+			hibernateTemplate.clear();
+		} catch (RuntimeException e) {
+			synchronized (failed) {
+				failed.addAll(getProcessed());
+			}
+			// This used to contain a call to onError, however, I think this
+			// should be handled within the step.
+			throw e;
+		}
+		unsetContext();
+	}
+
 }

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

@@ -438,20 +438,27 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
 		return true;
 	}
 
-	/*
+	/* TODO remove once ItemStream interface is modified.
 	 * (non-Javadoc)
 	 * @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext)
 	 */
 	public void mark() {
+	}
+	
+	public void flush() throws Exception {
 		lastCommitPointPosition = getPosition();
 		lastCommitPointRecordCount = currentRecordCount;
 	}
 
-	/*
+	/* TODO remove once ItemStream interface is modified.
 	 * (non-Javadoc)
 	 * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionContext)
 	 */
 	public void reset() {
+
+	}
+	
+	public void clear() throws Exception {
 		currentRecordCount = lastCommitPointRecordCount;
 		// close output
 		close();

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

@@ -17,10 +17,18 @@
 package org.springframework.batch.item;
 
 /**
- * Basic interface for generic output operations. Class implementing this
+ * <p>Basic interface for generic output operations. Class implementing this
  * interface will be responsible for serializing objects as necessary.
  * Generally, it is responsibility of implementing class to decide which
- * technology to use for mapping and how it should be configured.
+ * technology to use for mapping and how it should be configured.</p>
+ * 
+ * <p>
+ * Due to the nature of batch processing, it is expected that most writers
+ * will buffer output.  A flush method is provided to the interface in order
+ * to ensure that any buffers can be flushed before a transaction is
+ * committed.  Along the same lines, if a transaction has been rolled back,
+ * then the contents of any buffers should be thrown away.
+ * </p>
  * 
  * @author Dave Syer
  * @author Lucas Ward
@@ -38,4 +46,19 @@ public interface ItemWriter {
 	 */
 	public void write(Object item) throws Exception;
 	
+	/**
+	 * Flush any buffers that are being held.  This will usually be performed
+	 * prior to committing any transactions.
+	 * 
+	 * @throws Exception
+	 */
+	public void flush() throws Exception;
+	
+	/**
+	 * Clear any buffers that are being held.  This will usually be performed
+	 * prior to rolling back any transactions.
+	 * 
+	 * @throws Exception
+	 */
+	public void clear() throws Exception;
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/AbstractItemWriter.java

@@ -19,11 +19,17 @@ import org.springframework.batch.item.ItemWriter;
 
 /**
  * Abstract {@link ItemWriter} that allows for base classes to only
- * implement the close method if they need it.
+ * implement the close method if they need it.  Because it is likely
+ * that the flush and clear methods may not need to be implemented,
+ * they are provided in this class.
  * 
  * @author Lucas Ward
- *
  */
 public abstract class AbstractItemWriter implements ItemWriter{
-
+	
+	public void flush() throws Exception {
+	}
+	
+	public void clear() throws Exception {
+	}
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java

@@ -1,6 +1,5 @@
 package org.springframework.batch.item.writer;
 
-import org.springframework.batch.io.Skippable;
 import org.springframework.batch.item.ItemWriter;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.util.Assert;
@@ -11,7 +10,7 @@ import org.springframework.util.Assert;
  * @author Dave Syer
  * @author Robert Kasanicky
  */
-public class DelegatingItemWriter implements ItemWriter, Skippable, InitializingBean {
+public class DelegatingItemWriter implements ItemWriter, InitializingBean {
 
 	private ItemWriter writer;
 
@@ -43,14 +42,22 @@ public class DelegatingItemWriter implements ItemWriter, Skippable, Initializing
 		this.writer = writer;
 	}
 
-	public void skip() {
-		if (writer instanceof Skippable) {
-			((Skippable) writer).skip();
-		}
-	}
-
 	public void afterPropertiesSet() throws Exception {
 		Assert.notNull(writer);
 	}
 
+	/**
+	 * Delegates to {@link ItemWriter#clear()}
+	 */
+	public void clear() throws Exception {
+		writer.clear();
+	}
+
+	/**
+	 * Delegates to {@link ItemWriter#flush()}
+	 */
+	public void flush() throws Exception {
+		writer.flush();
+	}
+
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/ItemWriterAdapter.java

@@ -37,11 +37,15 @@ public class ItemWriterAdapter extends AbstractMethodInvokingDelegator implement
     /*
      * No-op, can't call more than one method.
      * 
-     * (non-Javadoc)
-     * @see org.springframework.batch.item.ItemWriter#close()
      */
-	public void close() throws Exception {
-		
+	public void clear() throws Exception {
+	}
+
+    /*
+     * No-op, can't call more than one method.
+     * 
+     */
+	public void flush() throws Exception {
 	}
 	
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/PropertyExtractingDelegatingItemWriter.java

@@ -67,6 +67,10 @@ public class PropertyExtractingDelegatingItemWriter extends AbstractMethodInvoki
 	}
 
 
-	public void close() throws Exception {
+	public void clear() throws Exception {
+	}
+
+
+	public void flush() throws Exception {
 	}
 }

spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java

@@ -24,7 +24,6 @@ import java.util.Collections;
 
 import junit.framework.TestCase;
 
-import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.writer.ItemTransformer;
 import org.springframework.core.io.FileSystemResource;
@@ -302,12 +301,12 @@ public class FlatFileItemWriterTests extends TestCase {
 		assertEquals(0, streamContext.getLong(FlatFileItemWriter.RESTART_DATA_NAME));
 	}
 
-	private void commit() {
-		((ItemStream) inputSource).mark();
+	private void commit() throws Exception{
+		inputSource.flush();
 	}
 
-	private void rollback() {
-		((ItemStream) inputSource).reset();
+	private void rollback() throws Exception{
+		inputSource.clear();
 	}
 
 }

spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/HibernateAwareItemWriterTests.java

@@ -73,6 +73,14 @@ public class HibernateAwareItemWriterTests extends TestCase {
 
 		public void close() throws Exception {
 		}
+
+		public void clear() throws Exception {
+			list.add("clear");
+		}
+
+		public void flush() throws Exception {
+			list.add("flush");
+		}
 	}
 
 	HibernateAwareItemWriter writer = new HibernateAwareItemWriter();
@@ -148,7 +156,7 @@ public class HibernateAwareItemWriterTests extends TestCase {
 	 * Test method for
 	 * {@link org.springframework.batch.io.support.HibernateAwareItemWriter#write(java.lang.Object)}.
 	 */
-	public void testCloseWithFailure() {
+	public void testCloseWithFailure() throws Exception{
 		final RuntimeException ex = new RuntimeException("bar");
 		writer.setHibernateTemplate(new HibernateTemplate() {
 			public void flush() throws DataAccessException {
@@ -156,14 +164,13 @@ public class HibernateAwareItemWriterTests extends TestCase {
 			}
 		});
 		try {
-			writer.close(context);
+			writer.flush();
 			fail("Expected RuntimeException");
 		} catch (RuntimeException e) {
 			assertEquals("bar", e.getMessage());
 		}
-		assertEquals(2, list.size());
-		assertTrue(list.contains(ex));
-		assertTrue(list.contains(context));
+		assertEquals(1, list.size());
+		assertTrue(list.contains("flush"));
 	}
 
 	/**
@@ -180,14 +187,13 @@ public class HibernateAwareItemWriterTests extends TestCase {
 		});
 		writer.write("foo");
 		try {
-			writer.close(context);
+			writer.flush();
 			fail("Expected RuntimeException");
 		} catch (RuntimeException e) {
 			assertEquals("bar", e.getMessage());
 		}
-		assertEquals(3, list.size());
-		assertTrue(list.contains(ex));
-		assertTrue(list.contains(context));
+		assertEquals(2, list.size());
+		assertTrue(list.contains("flush"));
 		writer.setHibernateTemplate(new HibernateTemplateWrapper() {
 			public void flush() throws DataAccessException {
 				list.add("flush");
@@ -221,8 +227,8 @@ public class HibernateAwareItemWriterTests extends TestCase {
 	 * Test method for
 	 * {@link org.springframework.batch.io.support.HibernateAwareItemWriter#close(org.springframework.batch.repeat.RepeatContext)}.
 	 */
-	public void testClose() {
-		writer.close(context);
+	public void testFlush() throws Exception{
+		writer.flush();
 		assertEquals(3, list.size());
 		assertTrue(list.contains("flush"));
 	}
@@ -231,11 +237,11 @@ public class HibernateAwareItemWriterTests extends TestCase {
 	 * Test method for
 	 * {@link org.springframework.batch.io.support.HibernateAwareItemWriter#close(org.springframework.batch.repeat.RepeatContext)}.
 	 */
-	public void testCloseAfterClear() {
+	public void testCloseAfterClear() throws Exception{
 		Map map = TransactionSynchronizationManager.getResourceMap();
 		String key = (String) map.keySet().iterator().next();
 		TransactionSynchronizationManager.unbindResource(key);
-		writer.close(context);
+		writer.flush();
 		assertEquals(3, list.size());
 		assertTrue(list.contains("flush"));
 		assertTrue(list.contains("clear"));

spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemWriterTests.java

@@ -23,7 +23,7 @@ import org.springframework.xml.transform.StaxResult;
 /**
  * Tests for {@link StaxStreamWriterOutputSource}.
  */
-public class StaxEventWriterItemWriterTests extends TestCase {
+public class StaxEventItemWriterTests extends TestCase {
 
 	// object under test
 	private StaxEventItemWriter writer;
@@ -67,7 +67,7 @@ public class StaxEventWriterItemWriterTests extends TestCase {
 	public void testRollback() throws Exception {
 		writer.write(record);
 		// rollback
-		writer.reset();
+		writer.clear();
 		assertEquals("", outputFileContent());
 	}
 
@@ -77,7 +77,7 @@ public class StaxEventWriterItemWriterTests extends TestCase {
 	public void testCommit() throws Exception {
 		writer.write(record);
 		// commit
-		writer.mark();
+		writer.flush();
 		assertTrue(outputFileContent().contains(TEST_STRING));
 	}
 
@@ -126,7 +126,7 @@ public class StaxEventWriterItemWriterTests extends TestCase {
 	/**
 	 * Open method writes the root tag, close method adds corresponding end tag.
 	 */
-	public void testOpenAndClose() throws IOException {
+	public void testOpenAndClose() throws Exception {
 		writer.setRootTagName("testroot");
 		writer.setRootElementAttributes(new HashMap() {
 			{
@@ -134,7 +134,7 @@ public class StaxEventWriterItemWriterTests extends TestCase {
 			}
 		});
 		writer.open();
-		writer.mark();
+		writer.flush();
 
 		assertTrue(outputFileContent().indexOf("<testroot attribute=\"value\"") != NOT_FOUND);
 

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/DelegatingItemWriterTests.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2006-2008 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.item.writer;
+
+import org.easymock.MockControl;
+import org.springframework.batch.item.ItemWriter;
+
+import junit.framework.TestCase;
+
+/**
+ * @author Lucas Ward
+ *
+ */
+public class DelegatingItemWriterTests extends TestCase {
+
+	MockControl writerControl = MockControl.createControl(ItemWriter.class);
+	ItemWriter itemWriter;
+	DelegatingItemWriter delegatingWriter;
+	
+	/* (non-Javadoc)
+	 * @see junit.framework.TestCase#setUp()
+	 */
+	protected void setUp() throws Exception {
+		super.setUp();
+		
+		itemWriter = (ItemWriter)writerControl.getMock();
+		delegatingWriter = new DelegatingItemWriter();
+		delegatingWriter.setDelegate(itemWriter);
+	}
+
+	public void testFlush() throws Exception{
+		itemWriter.flush();
+		writerControl.replay();
+		delegatingWriter.flush();
+		writerControl.verify();
+	}
+	
+	public void testClear() throws Exception{
+		itemWriter.clear();
+		writerControl.replay();
+		delegatingWriter.clear();
+		writerControl.verify();
+	}
+	
+	public void testCreation() throws Exception{
+		try{
+			delegatingWriter.setDelegate(null);
+			delegatingWriter.afterPropertiesSet();
+			fail();
+		}
+		catch(IllegalArgumentException ex){
+			//expected
+		}
+	}
+	
+	public void testWrite() throws Exception{
+		
+		ProcessingWriter writer = new ProcessingWriter();
+		writer.setDelegate(itemWriter);
+		Object item = new Object();
+		itemWriter.write(item);
+		writerControl.replay();
+		writer.write(item);
+		writerControl.verify();
+		assertTrue(writer.isDoProcessCalled());
+	}
+	
+	private class ProcessingWriter extends DelegatingItemWriter{
+		
+		boolean doProcessCalled = false;
+		
+		protected Object doProcess(Object item) throws Exception {
+			doProcessCalled = true;
+			return super.doProcess(item);
+		}
+		
+		public boolean isDoProcessCalled() {
+			return doProcessCalled;
+		}
+	}
+
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/ItemWriterItemProcessorTests.java

@@ -1,112 +0,0 @@
-/*
- * Copyright 2006-2007 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.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.springframework.batch.item.writer;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import junit.framework.TestCase;
-
-import org.springframework.batch.io.Skippable;
-import org.springframework.batch.item.ExecutionContext;
-import org.springframework.batch.item.ItemWriter;
-import org.springframework.batch.support.PropertiesConverter;
-
-/**
- * @author Dave Syer
- * 
- */
-public class ItemWriterItemProcessorTests extends TestCase {
-
-	private DelegatingItemWriter processor = new DelegatingItemWriter();
-
-	private ItemWriter writer;
-
-	/*
-	 * (non-Javadoc)
-	 * @see junit.framework.TestCase#setUp()
-	 */
-	protected void setUp() throws Exception {
-		writer = new MockOutputSource("test");
-		processor.setDelegate(writer);
-		processor.afterPropertiesSet();
-	}
-
-	public void testProcess() throws Exception {
-		processor.write("foo");
-		assertEquals(1, list.size());
-		assertEquals("test:foo", list.get(0));
-	}
-
-	public void testSkip() {
-		processor.skip();
-		assertEquals(1, list.size());
-		assertEquals("after skip", list.get(0));
-	}
-
-	/**
-	 * ItemWriter property must be set.
-	 */
-	public void testAfterPropertiesSet() throws Exception {
-		processor.setDelegate(null);
-		try {
-			processor.afterPropertiesSet();
-			fail();
-		}
-		catch (IllegalArgumentException e) {
-			// expected
-		}
-	}
-
-	private List list = new ArrayList();
-
-	/**
-	 * @author Dave Syer
-	 * 
-	 */
-	public class MockOutputSource extends AbstractItemStreamItemWriter implements Skippable {
-
-		private String value;
-
-		public MockOutputSource(String string) {
-			this.value = string;
-		}
-
-		public void write(Object output) {
-			list.add(value + ":" + output);
-		}
-
-		public void close() {
-		}
-
-		public void open() {
-		}
-
-		public ExecutionContext getExecutionContext() {
-			return new ExecutionContext(PropertiesConverter.stringToProperties("value=foo"));
-		}
-
-		public void restoreFrom(ExecutionContext data) {
-			value = data.getProperties().getProperty("value");
-		}
-
-		public void skip() {
-			list.add("after skip");
-		}
-
-	}
-
-}