diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/interceptor/CompositeJobListener.java b/spring-batch-core/src/main/java/org/springframework/batch/core/interceptor/CompositeJobListener.java index 8da625c23..0ceeb8290 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/interceptor/CompositeJobListener.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/interceptor/CompositeJobListener.java @@ -22,7 +22,6 @@ import java.util.List; import org.springframework.batch.core.domain.JobExecution; import org.springframework.batch.core.domain.JobListener; -import org.springframework.batch.core.domain.StepListener; /** * @author Dave Syer @@ -41,17 +40,6 @@ public class CompositeJobListener implements JobListener { this.listeners = Arrays.asList(listeners); } - /** - * Public setter for the listeners. The result will be as if - * {@link #setListeners(StepListener[])} was called with an array of length - * one. - * - * @param listener - */ - public void setListener(JobListener listener) { - setListeners(new JobListener[] {listener}); - } - /** * Register additional listener. * diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/interceptor/CompositeStepListener.java b/spring-batch-core/src/main/java/org/springframework/batch/core/interceptor/CompositeStepListener.java index 3f5822921..e3ed24e2e 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/interceptor/CompositeStepListener.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/interceptor/CompositeStepListener.java @@ -41,17 +41,6 @@ public class CompositeStepListener implements StepListener { this.listeners = Arrays.asList(listeners); } - /** - * Public setter for the listeners. The result will be as if - * {@link #setListeners(StepListener[])} was called with an array of length - * one. - * - * @param listener - */ - public void setListener(StepListener listener) { - setListeners(new StepListener[] {listener}); - } - /** * Register additional listener. * diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/interceptor/CompositeJobListenerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/interceptor/CompositeJobListenerTests.java index 1b906ef5c..1b44d4173 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/interceptor/CompositeJobListenerTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/interceptor/CompositeJobListenerTests.java @@ -58,7 +58,7 @@ public class CompositeJobListenerTests extends TestCase { * {@link org.springframework.batch.core.interceptor.CompositeJobListener#setListener(org.springframework.batch.core.domain.JobListener)}. */ public void testSetListener() { - listener.setListener(new JobListenerSupport() { + listener.register(new JobListenerSupport() { public void afterJob() { list.add("fail"); } @@ -72,7 +72,7 @@ public class CompositeJobListenerTests extends TestCase { * {@link org.springframework.batch.core.interceptor.CompositeJobListener#beforeJob(JobExecution)}. */ public void testOpen() { - listener.setListener(new JobListenerSupport() { + listener.register(new JobListenerSupport() { public void beforeJob(JobExecution stepExecution) { list.add("foo"); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/interceptor/CompositeStepListenerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/interceptor/CompositeStepListenerTests.java index f78324d6b..2e95e16da 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/interceptor/CompositeStepListenerTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/interceptor/CompositeStepListenerTests.java @@ -60,7 +60,7 @@ public class CompositeStepListenerTests extends TestCase { * {@link org.springframework.batch.core.interceptor.CompositeStepListener#setListener(org.springframework.batch.core.domain.StepListener)}. */ public void testSetListener() { - listener.setListener(new StepListenerSupport() { + listener.register(new StepListenerSupport() { public ExitStatus afterStep() { list.add("fail"); return ExitStatus.FAILED; @@ -75,7 +75,7 @@ public class CompositeStepListenerTests extends TestCase { * {@link org.springframework.batch.core.interceptor.CompositeStepListener#beforeStep(StepExecution)}. */ public void testOpen() { - listener.setListener(new StepListenerSupport() { + listener.register(new StepListenerSupport() { public void beforeStep(StepExecution stepExecution) { list.add("foo"); } @@ -89,7 +89,7 @@ public class CompositeStepListenerTests extends TestCase { * {@link org.springframework.batch.core.interceptor.CompositeStepListener#beforeStep(StepExecution)}. */ public void testOnError() { - listener.setListener(new StepListenerSupport() { + listener.register(new StepListenerSupport() { public ExitStatus onErrorInStep(Throwable e) { list.add("foo"); return null; diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java index 2bdc8d5df..3026c8e29 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java @@ -24,9 +24,11 @@ import org.springframework.batch.core.domain.JobInstance; import org.springframework.batch.core.domain.JobInterruptedException; import org.springframework.batch.core.domain.StepContribution; import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.core.domain.StepListener; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.runtime.ExitStatusExceptionClassifier; import org.springframework.batch.core.tasklet.Tasklet; +import org.springframework.batch.execution.step.support.ListenerMulticaster; import org.springframework.batch.execution.step.support.SimpleExitStatusExceptionClassifier; import org.springframework.batch.execution.step.support.StepInterruptionPolicy; import org.springframework.batch.execution.step.support.ThreadStepInterruptionPolicy; @@ -39,7 +41,6 @@ import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.KeyedItemReader; import org.springframework.batch.item.exception.CommitFailedException; -import org.springframework.batch.item.stream.SimpleStreamManager; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.repeat.RepeatCallback; import org.springframework.batch.repeat.RepeatContext; @@ -96,7 +97,20 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { private int commitInterval = 0; - private SimpleStreamManager streamManager; + private ListenerMulticaster listener = new ListenerMulticaster(); + + /** + * Register each of the objects as listeners. The {@link ItemOrientedStep} + * accepts listeners of type {@link ItemStream}, {@link StepListener}, + * TODO: complete the list. + * + * @param listeners an array of listener objects of known types. + */ + public void setListeners(Object[] listeners) { + for (int i = 0; i < listeners.length; i++) { + listener.register(listeners[i]); + } + } /** * The {@link RepeatOperations} to use for the outer loop of the batch @@ -183,8 +197,6 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { retryCallback = new ItemReaderRetryCallback((KeyedItemReader) itemReader, itemWriter); } - streamManager = new SimpleStreamManager(); - if (this.chunkOperations instanceof RepeatTemplate && commitInterval > 0) { ((RepeatTemplate) chunkOperations).setCompletionPolicy(new SimpleCompletionPolicy(commitInterval)); } @@ -219,6 +231,9 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { ExitStatus status = ExitStatus.FAILED; final ExceptionHolder fatalException = new ExceptionHolder(); + + // This could go in applyConfiguration(), but some unit tests do not call that + possiblyRegisterStreams(); try { @@ -228,8 +243,6 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { // the caller. fatalException.setException(updateStatus(stepExecution, BatchStatus.STARTED)); - possiblyRegisterStreams(); - if (isRestart && lastStepExecution != null) { stepExecution.setExecutionContext(lastStepExecution.getExecutionContext()); } @@ -237,11 +250,11 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { stepExecution.setExecutionContext(new ExecutionContext()); } - // Open the stream manager *after* the execution context is fixed in - // the step, otherwise it will not be the same reference that is - // updated by the streams. TODO: this is a little fragile - maybe - // StreamManager.update() should accept the context as a parameter. - streamManager.open(stepExecution.getExecutionContext()); + // Execute step level listeners *after* the execution context is + // fixed in the step. E.g. ItemStream instances need the the same + // reference to the ExecutionContext as the step execution. + listener.open(stepExecution.getExecutionContext()); + listener.beforeStep(stepExecution); status = stepOperations.iterate(new RepeatCallback() { @@ -253,9 +266,10 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { // interruption. interruptionPolicy.checkInterrupted(context); - ExitStatus result; + ExitStatus result = ExitStatus.CONTINUABLE; - TransactionStatus transaction = transactionManager.getTransaction(new DefaultTransactionDefinition()); + TransactionStatus transaction = transactionManager + .getTransaction(new DefaultTransactionDefinition()); try { itemReader.mark(); @@ -272,7 +286,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { // only if chunk was successful stepExecution.apply(contribution); - streamManager.update(stepExecution.getExecutionContext()); + listener.update(stepExecution.getExecutionContext()); try { stepExecution.setStatus(BatchStatus.COMPLETED); jobRepository.saveOrUpdateExecutionContext(stepExecution); @@ -284,6 +298,13 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { } } + + try { + result = result.and(listener.afterStep()); + } + catch (RuntimeException e) { + logger.error("Unexpected error in listener after step.", e); + } try { itemReader.mark(); @@ -311,6 +332,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { synchronized (stepExecution) { stepExecution.rollback(); } + try { itemReader.reset(); itemWriter.clear(); @@ -333,7 +355,6 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { logger.error("Exception should not cause step to fail", t); } - result = ExitStatus.CONTINUABLE; } // Check for interruption after transaction as well, so that @@ -356,11 +377,18 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { // classify exception so an exit code can be stored. status = exceptionClassifier.classifyForExitCode(e); + if (e.getCause() instanceof JobInterruptedException) { updateStatus(stepExecution, BatchStatus.STOPPED); throw (JobInterruptedException) e.getCause(); } else if (!fatalException.hasException()) { + try { + status = status.and(listener.onErrorInStep(e)); + } + catch (RuntimeException ex) { + logger.error("Unexpected error in listener on error in step.", ex); + } updateStatus(stepExecution, BatchStatus.FAILED); throw e; } @@ -388,7 +416,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { } try { - streamManager.close(stepExecution.getExecutionContext()); + listener.close(stepExecution.getExecutionContext()); } catch (RuntimeException e) { String msg = "Fatal error detected during close of streams. " @@ -410,17 +438,12 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { } /** - * + * Register the item reader and writer as listeners. If they are manually + * registered anyway, it shouldn't matter. */ private void possiblyRegisterStreams() { - if (itemReader instanceof ItemStream) { - ItemStream stream = (ItemStream) itemReader; - streamManager.register(stream); - } - if (itemWriter instanceof ItemStream) { - ItemStream stream = (ItemStream) itemWriter; - streamManager.register(stream); - } + listener.register(itemReader); + listener.register(itemWriter); } /** diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ListenerMulticaster.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ListenerMulticaster.java new file mode 100644 index 000000000..f3205ea78 --- /dev/null +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ListenerMulticaster.java @@ -0,0 +1,114 @@ +/* + * 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.execution.step.support; + +import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.core.domain.StepListener; +import org.springframework.batch.core.interceptor.CompositeStepListener; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.exception.StreamException; +import org.springframework.batch.item.stream.CompositeItemStream; +import org.springframework.batch.repeat.ExitStatus; + +/** + * @author Dave Syer + * + */ +public class ListenerMulticaster implements ItemStream, StepListener { + + private CompositeItemStream stream = new CompositeItemStream(); + + private CompositeStepListener stepListener = new CompositeStepListener(); + + /** + * Register each of the objects as listeners. Once registered, calls to the + * {@link ListenerMulticaster} broadcast to the individual listeners. + * + * @param listeners an array of listener objects of types known to the + * multicaster. + */ + public void setListeners(Object[] listeners) { + for (int i = 0; i < listeners.length; i++) { + register(listeners[i]); + } + } + + /** + * Register the listener for callbacks on the appropriate interfaces + * implemented. + */ + public void register(Object listener) { + if (listener instanceof StepListener) { + this.stepListener.register((StepListener) listener); + } + if (listener instanceof ItemStream) { + this.stream.register((ItemStream) listener); + } + } + + /** + * @return + * @see org.springframework.batch.core.interceptor.CompositeStepListener#afterStep() + */ + public ExitStatus afterStep() { + return stepListener.afterStep(); + } + + /** + * @param stepExecution + * @see org.springframework.batch.core.interceptor.CompositeStepListener#beforeStep(org.springframework.batch.core.domain.StepExecution) + */ + public void beforeStep(StepExecution stepExecution) { + stepListener.beforeStep(stepExecution); + } + + /** + * @param e + * @return + * @see org.springframework.batch.core.interceptor.CompositeStepListener#onErrorInStep(java.lang.Throwable) + */ + public ExitStatus onErrorInStep(Throwable e) { + return stepListener.onErrorInStep(e); + } + + /** + * @param executionContext + * @throws StreamException + * @see org.springframework.batch.item.stream.CompositeItemStream#close(org.springframework.batch.item.ExecutionContext) + */ + public void close(ExecutionContext executionContext) throws StreamException { + stream.close(executionContext); + } + + /** + * @param executionContext + * @throws StreamException + * @see org.springframework.batch.item.stream.CompositeItemStream#open(org.springframework.batch.item.ExecutionContext) + */ + public void open(ExecutionContext executionContext) throws StreamException { + stream.open(executionContext); + } + + /** + * @param executionContext + * @see org.springframework.batch.item.stream.CompositeItemStream#update(org.springframework.batch.item.ExecutionContext) + */ + public void update(ExecutionContext executionContext) { + stream.update(executionContext); + } + +} diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java index 2f9e9e228..ec4d72df6 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java @@ -29,6 +29,7 @@ import org.springframework.batch.core.domain.JobInterruptedException; import org.springframework.batch.core.domain.JobParameters; import org.springframework.batch.core.domain.StepContribution; import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.core.interceptor.StepListenerSupport; import org.springframework.batch.execution.job.JobSupport; import org.springframework.batch.execution.repository.SimpleJobRepository; import org.springframework.batch.execution.repository.dao.MapJobExecutionDao; @@ -353,6 +354,85 @@ public class ItemOrientedStepTests extends TestCase { assertEquals("bar", stepExecution.getExecutionContext().getString("foo")); } + public void testDirectlyInjectedItemStream() throws Exception { + itemOrientedStep.setListeners(new Object[] {new ItemStreamSupport() { + public void update(ExecutionContext executionContext) { + executionContext.putString("foo", "bar"); + } + }}); + JobExecution jobExecution = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecution); + + assertEquals(false, stepExecution.getExecutionContext().containsKey("foo")); + + itemOrientedStep.execute(stepExecution); + + assertEquals("bar", stepExecution.getExecutionContext().getString("foo")); + } + + public void testDirectlyInjectedListener() throws Exception { + itemOrientedStep.setListeners(new Object[] {new StepListenerSupport() { + public void beforeStep(StepExecution stepExecution) { + list.add("foo"); + } + public ExitStatus afterStep() { + list.add("bar"); + return null; + } + }}); + JobExecution jobExecution = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecution); + itemOrientedStep.execute(stepExecution); + assertEquals(2, list.size()); + } + + public void testDirectlyInjectedListenerOnError() throws Exception { + itemOrientedStep.setListeners(new Object[] {new StepListenerSupport() { + public ExitStatus onErrorInStep(Throwable e) { + list.add(e); + return null; + } + }}); + itemOrientedStep.setItemReader(new MockRestartableItemReader() { + public Object read() throws Exception { + throw new RuntimeException("FOO"); + } + }); + JobExecution jobExecution = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecution); + try { + itemOrientedStep.execute(stepExecution); + fail("Expected RuntimeException"); + } catch (RuntimeException e) { + assertEquals("FOO", e.getMessage()); + } + assertEquals(1, list.size()); + } + + public void testDirectlyInjectedStreamWhichIsAlsoReader() throws Exception { + MockRestartableItemReader reader = new MockRestartableItemReader() { + public Object read() throws Exception { + return "foo"; + } + public void update(ExecutionContext executionContext) { + // TODO Auto-generated method stub + executionContext.putString("foo", "bar"); + } + }; + itemOrientedStep.setItemReader(reader); + itemOrientedStep.setListeners(new Object[] {reader}); + JobExecution jobExecution = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecution); + + assertEquals(false, stepExecution.getExecutionContext().containsKey("foo")); + + itemOrientedStep.execute(stepExecution); + + // At least once in that process the statistics service was asked for + // statistics... + assertEquals("bar", stepExecution.getExecutionContext().getString("foo")); + } + public void testStatusForInterruptedException() { StepInterruptionPolicy interruptionPolicy = new StepInterruptionPolicy() { diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/CompositeItemStream.java similarity index 88% rename from spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java rename to spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/CompositeItemStream.java index cebbee542..6baad7422 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/CompositeItemStream.java @@ -16,6 +16,7 @@ package org.springframework.batch.item.stream; import java.util.ArrayList; +import java.util.Arrays; import java.util.Iterator; import java.util.List; @@ -24,20 +25,28 @@ import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.exception.StreamException; /** - * Simple {@link StreamManager} that tries to resolve conflicts between key - * names by using the class name of a stream to prefix property keys. + * Simple {@link ItemStream} that delegates to a list of other streams. * * @author Dave Syer * */ -public class SimpleStreamManager implements ItemStream { +public class CompositeItemStream implements ItemStream { private List streams = new ArrayList(); + /** + * Public setter for the listeners. + * + * @param listeners + */ + public void setStreams(ItemStream[] listeners) { + this.streams = Arrays.asList(listeners); + } + /** * */ - public SimpleStreamManager() { + public CompositeItemStream() { super(); } @@ -82,7 +91,6 @@ public class SimpleStreamManager implements ItemStream { ItemStream itemStream = (ItemStream) it.next(); itemStream.close(executionContext); } - streams.clear(); } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java index 1b1b5995a..cad01395d 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java @@ -21,6 +21,7 @@ import java.util.List; import junit.framework.TestCase; import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.exception.StreamException; /** @@ -29,13 +30,13 @@ import org.springframework.batch.item.exception.StreamException; */ public class SimpleStreamManagerTests extends TestCase { - private SimpleStreamManager manager = new SimpleStreamManager(); + private CompositeItemStream manager = new CompositeItemStream(); private List list = new ArrayList(); /** * Test method for - * {@link org.springframework.batch.item.stream.SimpleStreamManager#commit(org.springframework.transaction.TransactionStatus)}. + * {@link org.springframework.batch.item.stream.CompositeItemStream#commit(org.springframework.transaction.TransactionStatus)}. */ public void testRegisterAndOpen() { ItemStreamSupport stream = new ItemStreamSupport() { @@ -50,7 +51,7 @@ public class SimpleStreamManagerTests extends TestCase { /** * Test method for - * {@link org.springframework.batch.item.stream.SimpleStreamManager#commit(org.springframework.transaction.TransactionStatus)}. + * {@link org.springframework.batch.item.stream.CompositeItemStream#commit(org.springframework.transaction.TransactionStatus)}. */ public void testRegisterTwice() { ItemStreamSupport stream = new ItemStreamSupport() { @@ -66,7 +67,7 @@ public class SimpleStreamManagerTests extends TestCase { /** * Test method for - * {@link org.springframework.batch.item.stream.SimpleStreamManager#commit(org.springframework.transaction.TransactionStatus)}. + * {@link org.springframework.batch.item.stream.CompositeItemStream#commit(org.springframework.transaction.TransactionStatus)}. */ public void testMark() { manager.register(new ItemStreamSupport() { @@ -80,7 +81,7 @@ public class SimpleStreamManagerTests extends TestCase { /** * Test method for - * {@link org.springframework.batch.item.stream.SimpleStreamManager#commit(org.springframework.transaction.TransactionStatus)}. + * {@link org.springframework.batch.item.stream.CompositeItemStream#commit(org.springframework.transaction.TransactionStatus)}. */ public void testClose() { manager.register(new ItemStreamSupport() { @@ -94,18 +95,18 @@ public class SimpleStreamManagerTests extends TestCase { /** * Test method for - * {@link org.springframework.batch.item.stream.SimpleStreamManager#commit(org.springframework.transaction.TransactionStatus)}. + * {@link org.springframework.batch.item.stream.CompositeItemStream#commit(org.springframework.transaction.TransactionStatus)}. */ - public void testCloseUnregisters() { - manager.register(new ItemStreamSupport() { + public void testCloseDoesNotUnregister() { + manager.setStreams(new ItemStream[] { new ItemStreamSupport() { public void open(ExecutionContext executionContext) throws StreamException { list.add("bar"); } - }); + } }); manager.open(null); manager.close(null); manager.open(null); - assertEquals(1, list.size()); + assertEquals(2, list.size()); } }