diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java index b11f503e36..37b458bd27 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java @@ -36,6 +36,7 @@ import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessagingException; import org.springframework.integration.file.DefaultFileNameGenerator; import org.springframework.integration.file.FileNameGenerator; +import org.springframework.integration.file.remote.session.CachingSessionFactory; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor; @@ -272,7 +273,7 @@ public class RemoteFileTemplate implements RemoteFileOperations, Initializ @Override public boolean get(final Message message, final InputStreamCallback callback) { - Assert.notNull(this.fileNameProcessor, "'fileNameProcessor' needed to use get"); + Assert.notNull(this.fileNameProcessor, "A 'fileNameExpression' is needed to use get"); return this.execute(new SessionCallback() { @Override @@ -294,7 +295,13 @@ public class RemoteFileTemplate implements RemoteFileOperations, Initializ Assert.notNull(session, "failed to acquire a Session"); return callback.doInSession(session); } - catch (IOException e) { + catch (Exception e) { + if (session instanceof CachingSessionFactory.CachedSession) { + ((CachingSessionFactory.CachedSession) session).dirty(); + } + if (e instanceof MessagingException) { + throw (MessagingException) e; + } throw new MessagingException("Failed to execute on session", e); } finally { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java index d28e8762c8..3121ab80bd 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java @@ -147,11 +147,13 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe this.pool.removeAllIdleItems(); } - private class CachedSession implements Session { + public class CachedSession implements Session { private final Session targetSession; - private volatile boolean released; + private boolean released; + + private boolean dirty; /** * The epoch in which this session was created. @@ -180,6 +182,9 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe } this.targetSession.close(); } + else if (this.dirty) { + this.targetSession.close(); + } pool.releaseItem(targetSession); released = true; } @@ -240,6 +245,10 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe return this.targetSession.finalizeRaw(); } + public void dirty() { + this.dirty = true; + } + } } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java index f0cca26600..48e7dfcd49 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java @@ -15,16 +15,29 @@ */ package org.springframework.integration.file.remote.session; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; import org.junit.Test; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.file.remote.InputStreamCallback; +import org.springframework.integration.file.remote.RemoteFileTemplate; +import org.springframework.integration.message.GenericMessage; import org.springframework.integration.test.util.TestUtils; /** @@ -63,6 +76,37 @@ public class CachingSessionFactoryTests { assertFalse(sess1.isOpen()); } + @Test + public void testDirtySession() throws Exception { + @SuppressWarnings("unchecked") + SessionFactory factory = mock(SessionFactory.class); + @SuppressWarnings("unchecked") + Session session = mock(Session.class); + when(factory.getSession()).thenReturn(session); + when(session.readRaw("foo")).thenReturn(new ByteArrayInputStream("".getBytes())); + when(session.finalizeRaw()).thenReturn(true); + CachingSessionFactory ccf = new CachingSessionFactory(factory); + RemoteFileTemplate template = new RemoteFileTemplate(ccf); + template.setFileNameExpression(new LiteralExpression("foo")); + template.setBeanFactory(mock(BeanFactory.class)); + template.afterPropertiesSet(); + try { + template.get(new GenericMessage("foo"), new InputStreamCallback() { + + @Override + public void doWithInputStream(InputStream stream) throws IOException { + throw new RuntimeException("bar"); + } + }); + fail("Expected exception"); + } + catch (Exception e) { + assertThat(e.getCause(), instanceOf(RuntimeException.class)); + assertThat(e.getCause().getMessage(), equalTo("bar")); + } + verify(session).close(); + } + private class TestSessionFactory implements SessionFactory { private int n;