INT-3266 (S)FTP Close Dirty Cached Sessions

If an exception occurs on a session it should be physically closed
and not reused because its state is indeterminate and the next
operation might fail.

Note: The booleans within CachedSession do not need to be volatile
because it is a short-lived object only used by the current thread.

Also fixes the assertion message in RFT.get().

JIRA: https://jira.springsource.org/browse/INT-3266

INT-3266 Polishing - PR Comments

Remove need for SuppressWarnings.
Fix 2 test cases where the exception has an additional cause.

Fix invalid `GenericMessage` import
Fix `e.getCause()` in `FtpServerOutboundTests` & `SftpServerOutboundTests` tests

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<F> implements RemoteFileOperations<F>, 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<F, Boolean>() {
 
 			@Override
@@ -294,7 +295,13 @@ public class RemoteFileTemplate<F> implements RemoteFileOperations<F>, 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 {

spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java

@@ -147,11 +147,13 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 		this.pool.removeAllIdleItems();
 	}
 
-	private class CachedSession implements Session<F> {
+	public class CachedSession implements Session<F> {
 
 		private final Session<F> 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<F> implements SessionFactory<F>, DisposableBe
 					}
 					this.targetSession.close();
 				}
+				else if (this.dirty) {
+					this.targetSession.close();
+				}
 				pool.releaseItem(targetSession);
 				released = true;
 			}
@@ -240,6 +245,10 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 			return this.targetSession.finalizeRaw();
 		}
 
+		public void dirty() {
+			this.dirty = true;
+		}
+
 	}
 
 }

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<Object> factory = mock(SessionFactory.class);
+		@SuppressWarnings("unchecked")
+		Session<Object> session = mock(Session.class);
+		when(factory.getSession()).thenReturn(session);
+		when(session.readRaw("foo")).thenReturn(new ByteArrayInputStream("".getBytes()));
+		when(session.finalizeRaw()).thenReturn(true);
+		CachingSessionFactory<Object> ccf = new CachingSessionFactory<Object>(factory);
+		RemoteFileTemplate<Object> template = new RemoteFileTemplate<Object>(ccf);
+		template.setFileNameExpression(new LiteralExpression("foo"));
+		template.setBeanFactory(mock(BeanFactory.class));
+		template.afterPropertiesSet();
+		try {
+			template.get(new GenericMessage<String>("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<String> {
 
 		private int n;