From 1d002bfc4784117f55b14ec5095f2a5c9426e910 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 21 Jan 2014 11:18:57 +0200 Subject: [PATCH] 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. --- .../file/remote/RemoteFileTemplate.java | 9 +++- .../remote/session/CachingSessionFactory.java | 13 +++++- .../session/CachingSessionFactoryTests.java | 44 +++++++++++++++++++ .../ftp/outbound/FtpServerOutboundTests.java | 2 + .../outbound/SftpServerOutboundTests.java | 2 + 5 files changed, 67 insertions(+), 3 deletions(-) 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 323450fa17..f9619b2394 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 @@ -33,6 +33,7 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.expression.Expression; 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; @@ -299,7 +300,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 8d41cd11fd..495acc443a 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 @@ -152,11 +152,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. @@ -185,6 +187,9 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe } this.targetSession.close(); } + else if (this.dirty) { + this.targetSession.close(); + } pool.releaseItem(targetSession); released = true; } @@ -245,6 +250,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..5697e66e09 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,17 +15,30 @@ */ 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.test.util.TestUtils; +import org.springframework.messaging.support.GenericMessage; /** * @author Gary Russell @@ -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; diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java index d1d7a5d90f..bee1dcfde4 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java @@ -130,6 +130,8 @@ public class FtpServerOutboundTests { } catch (Exception e) { Throwable cause = e.getCause(); + assertNotNull(cause); + cause = cause.getCause(); assertThat(cause, Matchers.instanceOf(IllegalArgumentException.class)); assertThat(cause.getMessage(), Matchers.startsWith("Failed to make local directory")); } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java index 5633f90318..3f56861989 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java @@ -218,6 +218,8 @@ public class SftpServerOutboundTests { } catch (Exception e) { Throwable cause = e.getCause(); + assertNotNull(cause); + cause = cause.getCause(); assertThat(cause, Matchers.instanceOf(IllegalArgumentException.class)); assertThat(cause.getMessage(), Matchers.startsWith("Failed to make local directory")); }