BATCH-2312 Implement ItemStream in AsyncItemWriter

This commit adds the feature of allowing a delegate ItemWriter to be an
ItemStreamWriter and have Spring Batch respect the ItemStream lifecycle
events without explicitly configuring the delegate as a stream.

spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2007 the original author or authors.
+ * Copyright 2006-2015 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,6 +15,10 @@
  */
 package org.springframework.batch.integration.async;
 
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemStream;
+import org.springframework.batch.item.ItemStreamException;
+import org.springframework.batch.item.ItemStreamWriter;
 import org.springframework.batch.item.ItemWriter;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.util.Assert;
@@ -23,14 +27,14 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.Future;
 
-public class AsyncItemWriter<T> implements ItemWriter<Future<T>>, InitializingBean {
-	
+public class AsyncItemWriter<T> implements ItemStreamWriter<Future<T>>, InitializingBean {
+
 	private ItemWriter<T> delegate;
 
 	public void afterPropertiesSet() throws Exception {
 		Assert.notNull(delegate, "A delegate ItemWriter must be provided.");
 	}
-	
+
 	/**
 	 * @param delegate ItemWriter that does the actual writing of the Future results
 	 */
@@ -57,4 +61,25 @@ public class AsyncItemWriter<T> implements ItemWriter<Future<T>>, InitializingBe
 		}
 		delegate.write(list);
 	}
+
+	@Override
+	public void open(ExecutionContext executionContext) throws ItemStreamException {
+		if (delegate instanceof ItemStream) {
+			((ItemStream) delegate).open(executionContext);
+		}
+	}
+
+	@Override
+	public void update(ExecutionContext executionContext) throws ItemStreamException {
+		if (delegate instanceof ItemStream) {
+			((ItemStream) delegate).update(executionContext);
+		}
+	}
+
+	@Override
+	public void close() throws ItemStreamException {
+		if (delegate instanceof ItemStream) {
+			((ItemStream) delegate).close();
+		}
+	}
 }

spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemWriterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-2015 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,6 +17,9 @@ package org.springframework.batch.integration.async;
 
 import org.junit.Before;
 import org.junit.Test;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemStreamException;
+import org.springframework.batch.item.ItemStreamWriter;
 import org.springframework.batch.item.ItemWriter;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.core.task.TaskExecutor;
@@ -27,6 +30,7 @@ import java.util.concurrent.Callable;
 import java.util.concurrent.FutureTask;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 /**
@@ -43,11 +47,11 @@ public class AsyncItemWriterTests {
 		taskExecutor = new SimpleAsyncTaskExecutor();
 		writtenItems = new ArrayList<String>();
 		writer = new AsyncItemWriter<String>();
-		writer.setDelegate(new ListItemWriter(writtenItems));
 	}
 
 	@Test
 	public void testRoseyScenario() throws Exception {
+		writer.setDelegate(new ListItemWriter(writtenItems));
 		List<FutureTask<String>> processedItems = new ArrayList<FutureTask<String>>();
 
 		processedItems.add(new FutureTask<String>(new Callable<String>() {
@@ -77,6 +81,7 @@ public class AsyncItemWriterTests {
 
 	@Test
 	public void testFilteredItem() throws Exception {
+		writer.setDelegate(new ListItemWriter(writtenItems));
 		List<FutureTask<String>> processedItems = new ArrayList<FutureTask<String>>();
 
 		processedItems.add(new FutureTask<String>(new Callable<String>() {
@@ -103,9 +108,48 @@ public class AsyncItemWriterTests {
 		assertTrue(writtenItems.contains("foo"));
 	}
 
+	@Test
+	public void testStreamDelegate() throws Exception {
+		ListItemStreamWriter itemWriter = new ListItemStreamWriter(writtenItems);
+		writer.setDelegate(itemWriter);
+
+		List<FutureTask<String>> processedItems = new ArrayList<FutureTask<String>>();
+
+		ExecutionContext executionContext = new ExecutionContext();
+		writer.open(executionContext);
+		writer.write(processedItems);
+		writer.update(executionContext);
+		writer.close();
+
+		assertTrue(itemWriter.isOpened);
+		assertTrue(itemWriter.isUpdated);
+		assertTrue(itemWriter.isClosed);
+	}
+
+	@Test
+	public void testNonStreamDelegate() throws Exception {
+		ListItemWriter itemWriter = new ListItemWriter(writtenItems);
+		writer.setDelegate(itemWriter);
+
+		List<FutureTask<String>> processedItems = new ArrayList<FutureTask<String>>();
+
+		ExecutionContext executionContext = new ExecutionContext();
+		writer.open(executionContext);
+		writer.write(processedItems);
+		writer.update(executionContext);
+		writer.close();
+
+		assertFalse(itemWriter.isOpened);
+		assertFalse(itemWriter.isUpdated);
+		assertFalse(itemWriter.isClosed);
+	}
+
 	private class ListItemWriter implements ItemWriter<String> {
 
 		protected List<String> items;
+		public boolean isOpened = false;
+		public boolean isUpdated = false;
+		public boolean isClosed = false;
 
 		public ListItemWriter(List<String> items) {
 			this.items = items;
@@ -116,4 +160,35 @@ public class AsyncItemWriterTests {
 			this.items.addAll(items);
 		}
 	}
+
+	private class ListItemStreamWriter implements ItemStreamWriter<String> {
+		public boolean isOpened = false;
+		public boolean isUpdated = false;
+		public boolean isClosed = false;
+		protected List<String> items;
+
+		public ListItemStreamWriter(List<String> items) {
+			this.items = items;
+		}
+
+		@Override
+		public void write(List<? extends String> items) throws Exception {
+			this.items.addAll(items);
+		}
+
+		@Override
+		public void open(ExecutionContext executionContext) throws ItemStreamException {
+			isOpened = true;
+		}
+
+		@Override
+		public void update(ExecutionContext executionContext) throws ItemStreamException {
+			isUpdated = true;
+		}
+
+		@Override
+		public void close() throws ItemStreamException {
+			isClosed = true;
+		}
+	}
 }