diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java index fdfa0fef4..ec98484fa 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java +++ b/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 implements ItemWriter>, InitializingBean { - +public class AsyncItemWriter implements ItemStreamWriter>, InitializingBean { + private ItemWriter 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 implements ItemWriter>, 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(); + } + } } diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemWriterTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemWriterTests.java index 3fd6b63a3..e1b62392b 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemWriterTests.java +++ b/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(); writer = new AsyncItemWriter(); - writer.setDelegate(new ListItemWriter(writtenItems)); } @Test public void testRoseyScenario() throws Exception { + writer.setDelegate(new ListItemWriter(writtenItems)); List> processedItems = new ArrayList>(); processedItems.add(new FutureTask(new Callable() { @@ -77,6 +81,7 @@ public class AsyncItemWriterTests { @Test public void testFilteredItem() throws Exception { + writer.setDelegate(new ListItemWriter(writtenItems)); List> processedItems = new ArrayList>(); processedItems.add(new FutureTask(new Callable() { @@ -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> processedItems = new ArrayList>(); + + 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> processedItems = new ArrayList>(); + + 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 { protected List items; + public boolean isOpened = false; + public boolean isUpdated = false; + public boolean isClosed = false; public ListItemWriter(List items) { this.items = items; @@ -116,4 +160,35 @@ public class AsyncItemWriterTests { this.items.addAll(items); } } + + private class ListItemStreamWriter implements ItemStreamWriter { + public boolean isOpened = false; + public boolean isUpdated = false; + public boolean isClosed = false; + protected List items; + + public ListItemStreamWriter(List items) { + this.items = items; + } + + @Override + public void write(List 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; + } + } }