Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #131 from olegz/INT-2146

Browse Source
This commit is contained in:
Mark Fisher
2011-10-19 17:08:32 -04:00
parent 761a797ee1 7581c1cea4
commit 821a8c0061
10 changed files with 305 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
docs/src/reference/docbook/ftp.xml
View File

@@ -346,17 +346,34 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/ftp
<section id="ftp-session-caching">
<title>FTP Session Caching</title>
<para>
One of the optimizations implemented by the FTP adapters is session caching. Similar to JDBC pooling of Connections, the FTP Adapters maintain a
pool of Sessions by default. However there are times when this behavior is not desired (e.g., security etc.).
To disable session caching you can set the <code>cache-sessions</code> attribute to <code>false</code> (the default value is <code>true</code>).
<programlisting language="xml"><![CDATA[<int-ftp:inbound-channel-adapter id="ftpInbound"
channel="ftpChannel"
. . .
cache-sessions="false"
. . .
</int-ftp:inbound-channel-adapter>]]></programlisting>
As of version 2.1 we've exposed more flexibility with regard to session management for remote file adapters (e.g., FTP, SFTP etc).
In previous versions the sessions were cached automatically by default. We did expose a <code>cache-sessions</code> attribute for
disabling the auto caching, but that solution did not provide a way to configure other session caching attributes. For example, one
of the requested features was to support a limit on the number of sessions created since a remote server may impose a limit on the
number of client connections. To support that requirement and other configuration options, we decided to promote explicit definition
of the <classname>CachingSessionFactory</classname> instance. That provides the <code>sessionCacheSize</code> and <code>sessionWaitTimeout</code>
properties. As its name suggests, the <code>sessionCacheSize</code> property controls how many active sessions this adapter will
maintain in its cache (the DEFAULT is unbounded). If the <code>sessionCacheSize</code> threshold has been reached, any attempt to
acquire another session will block until either one of the cached sessions becomes available or until the wait time for a Session
expires (the DEFAULT wait time is Integer.MAX_VALUE). The <code>sessionWaitTimeout</code> property enables configuration of that value.
</para>
<para>
If you want your Sessions to be cached, simply configure your default Session Factory as described above and then
wrap it in an instance of <classname>CachingSessionFactory</classname> where you may provide those additional properties.
<programlisting language="xml"><![CDATA[<bean id="ftpSessionFactory" class="org.springframework.integration.ftp.session.DefaultFtpSessionFactory">
<property name="host" value="localhost"/>
</bean>
<bean id="cachingSessionFactory" class="org.springframework.integration.file.remote.session.CachingSessionFactory">
<constructor-arg ref="ftpSessionFactory"/>
<constructor-arg value="10"/>
<property name="sessionWaitTimeout" value="1000"/>
</bean>]]></programlisting>
In the above example you see a <classname>CachingSessionFactory</classname> created with the
<code>sessionCacheSize</code> set to 10 and the <code>sessionWaitTimeout</code> set to 1 second (its value is in millliseconds).
The same attribute can also be used with Outbound Channel Adapters.
</para>
</section>
</chapter>

39
docs/src/reference/docbook/sftp.xml
View File

@@ -289,18 +289,35 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp
</section>
<section id="sftp-session-caching">
<title>SFTP Session Caching</title>
<para>
One of the optimizations implemented by the SFTP adapters is session caching. Similar to JDBC pooling of Connections, the SFTP Adapters maintain a
pool of Sessions by default. However there are times when this behavior is not desired (e.g., security etc.).
To disable session caching you can set the <code>cache-sessions</code> attribute to <code>false</code> (the default value is <code>true</code>).
<programlisting language="xml"><![CDATA[<int-sftp:inbound-channel-adapter id="ftpInbound"
channel="sftpChannel"
. . .
cache-sessions="false"
. . .
</int-sftp:inbound-channel-adapter>]]></programlisting>
<para>
As of version 2.1 we've exposed more flexibility with regard to session management for remote file adapters (e.g., FTP, SFTP etc).
In previous versions the sessions were cached automatically by default. We did expose a <code>cache-sessions</code> attribute for
disabling the auto caching, but that solution did not provide a way to configure other session caching attributes. For example, one
of the requested features was to support a limit on the number of sessions created since a remote server may impose a limit on the
number of client connections. To support that requirement and other configuration options, we decided to promote explicit definition
of the <classname>CachingSessionFactory</classname> instance. That provides the <code>sessionCacheSize</code> and <code>sessionWaitTimeout</code>
properties. As its name suggests, the <code>sessionCacheSize</code> property controls how many active sessions this adapter will
maintain in its cache (the DEFAULT is unbounded). If the <code>sessionCacheSize</code> threshold has been reached, any attempt to
acquire another session will block until either one of the cached sessions becomes available or until the wait time for a Session
expires (the DEFAULT wait time is Integer.MAX_VALUE). The <code>sessionWaitTimeout</code> property enables configuration of that value.
</para>
<para>
If you want your Sessions to be cached, simply configure your default Session Factory as described above and then
wrap it in an instance of <classname>CachingSessionFactory</classname> where you may provide those additional properties.
<programlisting language="xml"><![CDATA[<bean id="sftpSessionFactory" class="org.springframework.integration.sftp.session.DefaultSftpSessionFactory">
<property name="host" value="localhost"/>
</bean>
<bean id="cachingSessionFactory" class="org.springframework.integration.file.remote.session.CachingSessionFactory">
<constructor-arg ref="sftpSessionFactory"/>
<property name="sessionCacheSize" value="10"/>
<property name="sessionWaitTimeout" value="1000"/>
</bean>]]></programlisting>
In the above example you see a <classname>CachingSessionFactory</classname> created with the
<code>sessionCacheSize</code> set to 10 and the <code>sessionWaitTimeout</code> set to 1 second (its value is in millliseconds).
The same attribute can also be used with Outbound Channel Adapters.
</para>
</section>
</chapter>

52
spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java
View File

@@ -18,12 +18,17 @@ package org.springframework.integration.file.config;
import org.w3c.dom.Element;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeanMetadataElement;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.ExpressionFactoryBean;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.file.remote.session.CachingSessionFactory;
import org.springframework.util.StringUtils;
/**
@@ -35,34 +40,41 @@ import org.springframework.util.StringUtils;
*/
public abstract class AbstractRemoteFileInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
private final Log logger = LogFactory.getLog(this.getClass());
@Override
protected final BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
BeanDefinitionBuilder synchronizerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
this.getInboundFileSynchronizerClassname());
// build the SessionFactory and provide as a constructor argument
String cacheSessions = element.getAttribute("cache-sessions");
if ("false".equalsIgnoreCase(cacheSessions)) {
synchronizerBuilder.addConstructorArgReference(element.getAttribute("session-factory"));
// This whole block must be refactored once the cache-session attribute is removed
String sessionFactoryName = element.getAttribute("session-factory");
BeanDefinition sessionFactoryDefinition = parserContext.getReaderContext().getRegistry().getBeanDefinition(sessionFactoryName);
String sessionFactoryClassName = sessionFactoryDefinition.getBeanClassName();
if (StringUtils.hasText(sessionFactoryClassName) && sessionFactoryClassName.endsWith(CachingSessionFactory.class.getName())) {
synchronizerBuilder.addConstructorArgValue(sessionFactoryDefinition);
}
else {
BeanDefinitionBuilder sessionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.file.remote.session.CachingSessionFactory");
sessionFactoryBuilder.addConstructorArgReference(element.getAttribute("session-factory"));
synchronizerBuilder.addConstructorArgValue(sessionFactoryBuilder.getBeanDefinition());
String cacheSessions = element.getAttribute("cache-sessions");
if (StringUtils.hasText(cacheSessions) && logger.isWarnEnabled()) {
logger.warn("The 'cache-sessions' attribute is deprecated as of version 2.1. " +
"Please configure a CachingSessionFactory explicitly instead.");
}
if ("false".equalsIgnoreCase(cacheSessions)) {
synchronizerBuilder.addConstructorArgReference(element.getAttribute("session-factory"));
}
else {
BeanDefinitionBuilder sessionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(CachingSessionFactory.class);
sessionFactoryBuilder.addConstructorArgReference(sessionFactoryName);
synchronizerBuilder.addConstructorArgValue(sessionFactoryBuilder.getBeanDefinition());
}
}
// end of what needs to be refactored once cache-session is removed
// configure the InboundFileSynchronizer properties
IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "remote-directory");
IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "delete-remote-files");
String localFileGeneratorExpression = element.getAttribute("local-filename-generator-expression");
if (StringUtils.hasText(localFileGeneratorExpression)){
BeanDefinitionBuilder localFileGeneratorExpressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class);
localFileGeneratorExpressionBuilder.addConstructorArgValue(localFileGeneratorExpression);
synchronizerBuilder.addPropertyValue("localFilenameGeneratorExpression", localFileGeneratorExpressionBuilder.getBeanDefinition());
}
String remoteFileSeparator = element.getAttribute("remote-file-separator");
synchronizerBuilder.addPropertyValue("remoteFileSeparator", remoteFileSeparator);
IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "temporary-file-suffix");
@@ -72,11 +84,17 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
BeanDefinitionBuilder messageSourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(this.getMessageSourceClassname());
messageSourceBuilder.addConstructorArgValue(synchronizerBuilder.getBeanDefinition());
String comparator = element.getAttribute("comparator");
if (StringUtils.hasText(comparator)){
if (StringUtils.hasText(comparator)) {
messageSourceBuilder.addConstructorArgReference(comparator);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "local-directory");
IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "auto-create-local-directory");
String localFileGeneratorExpression = element.getAttribute("local-filename-generator-expression");
if (StringUtils.hasText(localFileGeneratorExpression)) {
BeanDefinitionBuilder localFileGeneratorExpressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class);
localFileGeneratorExpressionBuilder.addConstructorArgValue(localFileGeneratorExpression);
synchronizerBuilder.addPropertyValue("localFilenameGeneratorExpression", localFileGeneratorExpressionBuilder.getBeanDefinition());
}
return messageSourceBuilder.getBeanDefinition();
}

52
spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileOutboundGatewayParser.java
View File

@@ -15,20 +15,25 @@
*/
package org.springframework.integration.file.config;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.file.remote.session.CachingSessionFactory;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* @author Gary Russell
* @author Oleg Zhurakousky
* @since 2.1
*
*/
public abstract class AbstractRemoteFileOutboundGatewayParser extends
AbstractConsumerEndpointParser {
public abstract class AbstractRemoteFileOutboundGatewayParser extends AbstractConsumerEndpointParser {
private final Log logger = LogFactory.getLog(this.getClass());
@Override
protected String getInputChannelAttributeName() {
@@ -38,17 +43,33 @@ public abstract class AbstractRemoteFileOutboundGatewayParser extends
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(getGatewayClassName());
// build the SessionFactory and provide as a constructor argument
String cacheSessions = element.getAttribute("cache-sessions");
if ("false".equalsIgnoreCase(cacheSessions)) {
builder.addConstructorArgReference(element.getAttribute("session-factory"));
// build the SessionFactory and provide it as a constructor argument
// This whole block must be refactored once cache-session attribute is removed
String sessionFactoryName = element.getAttribute("session-factory");
BeanDefinition sessionFactoryDefinition = parserContext.getReaderContext().getRegistry().getBeanDefinition(sessionFactoryName);
String sessionFactoryClassName = sessionFactoryDefinition.getBeanClassName();
if (StringUtils.hasText(sessionFactoryClassName) && sessionFactoryClassName.endsWith(CachingSessionFactory.class.getName())) {
builder.addConstructorArgValue(sessionFactoryDefinition);
}
else {
BeanDefinitionBuilder sessionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.file.remote.session.CachingSessionFactory");
sessionFactoryBuilder.addConstructorArgReference(element.getAttribute("session-factory"));
builder.addConstructorArgValue(sessionFactoryBuilder.getBeanDefinition());
String cacheSessions = element.getAttribute("cache-sessions");
if (StringUtils.hasText(cacheSessions) && logger.isWarnEnabled()) {
logger.warn("The 'cache-sessions' attribute is deprecated as of version 2.1." +
"Please configure a CachingSessionFactory explicitly instead.");
}
if ("false".equalsIgnoreCase(cacheSessions)) {
builder.addConstructorArgReference(element.getAttribute("session-factory"));
}
else {
BeanDefinitionBuilder sessionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(CachingSessionFactory.class);
sessionFactoryBuilder.addConstructorArgReference(sessionFactoryName);
builder.addConstructorArgValue(sessionFactoryBuilder.getBeanDefinition());
}
}
// end of what needs to be refactored once cache-session is removed
builder.addConstructorArgValue(element.getAttribute("command"));
builder.addConstructorArgValue(element.getAttribute(EXPRESSION_ATTRIBUTE));
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "command-options", "options");
@@ -75,14 +96,17 @@ public abstract class AbstractRemoteFileOutboundGatewayParser extends
if (count > 1) {
parserContext.getReaderContext().error("at most one of 'filename-pattern', " +
"'filename-regex', or 'filter' is allowed on remote file inbound adapter", element);
} else if (hasFilter) {
}
else if (hasFilter) {
builder.addPropertyReference("filter", filter);
} else if (hasFileNamePattern) {
}
else if (hasFileNamePattern) {
BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(
this.getSimplePatternFileListFilterClassname());
filterBuilder.addConstructorArgValue(fileNamePattern);
builder.addPropertyValue("filter", filterBuilder.getBeanDefinition());
} else if (hasFileNameRegex) {
}
else if (hasFileNameRegex) {
BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(
this.getRegexPatternFileListFilterClassname());
filterBuilder.addConstructorArgValue(fileNameRegex);

34
spring-integration-file/src/main/java/org/springframework/integration/file/config/RemoteFileOutboundChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -18,6 +18,8 @@ package org.springframework.integration.file.config;
import org.w3c.dom.Element;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
@@ -26,6 +28,7 @@ import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.file.remote.session.CachingSessionFactory;
import org.springframework.util.StringUtils;
/**
@@ -34,23 +37,34 @@ import org.springframework.util.StringUtils;
* @since 2.0
*/
public class RemoteFileOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
private final Log logger = LogFactory.getLog(this.getClass());
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder handlerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.file.remote.handler.FileTransferringMessageHandler");
// build the SessionFactory and provide as a constructor argument
String cacheSessions = element.getAttribute("cache-sessions");
if ("false".equalsIgnoreCase(cacheSessions)) {
handlerBuilder.addConstructorArgReference(element.getAttribute("session-factory"));
// This whole block must be refactored once cache-session attribute is removed
String sessionFactoryName = element.getAttribute("session-factory");
BeanDefinition sessionFactoryDefinition = parserContext.getReaderContext().getRegistry().getBeanDefinition(sessionFactoryName);
String sessionFactoryClassName = sessionFactoryDefinition.getBeanClassName();
if (StringUtils.hasText(sessionFactoryClassName) && sessionFactoryClassName.endsWith(CachingSessionFactory.class.getName())){
handlerBuilder.addConstructorArgValue(sessionFactoryDefinition);
}
else {
BeanDefinitionBuilder sessionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.file.remote.session.CachingSessionFactory");
sessionFactoryBuilder.addConstructorArgReference(element.getAttribute("session-factory"));
handlerBuilder.addConstructorArgValue(sessionFactoryBuilder.getBeanDefinition());
String cacheSessions = element.getAttribute("cache-sessions");
if (StringUtils.hasText(cacheSessions)){
logger.warn("The 'cache-sessions' attribute is deprecated since v2.1. Consider configuring CachingSessionFactory explicitly");
}
if ("false".equalsIgnoreCase(cacheSessions)) {
handlerBuilder.addConstructorArgReference(element.getAttribute("session-factory"));
}
else {
BeanDefinitionBuilder sessionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(CachingSessionFactory.class);
sessionFactoryBuilder.addConstructorArgReference(sessionFactoryName);
handlerBuilder.addConstructorArgValue(sessionFactoryBuilder.getBeanDefinition());
}
}
// end of what needs to be refactored once cache-session is removed
// configure MessageHandler properties
IntegrationNamespaceUtils.setValueIfAttributeDefined(handlerBuilder, element, "temporary-file-suffix");

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -19,13 +19,12 @@ package org.springframework.integration.file.remote.session;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.integration.util.UpperBound;
/**
* A {@link SessionFactory} implementation that caches Sessions for reuse without
@@ -41,41 +40,41 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean {
private static final Log logger = LogFactory.getLog(CachingSessionFactory.class);
public static final int DEFAULT_POOL_SIZE = 10;
private volatile long sessionWaitTimeout = Integer.MAX_VALUE;
private final Queue<Session> queue;
private final LinkedBlockingQueue<Session> queue = new LinkedBlockingQueue<Session>();
private final SessionFactory sessionFactory;
private final int maxPoolSize;
private final UpperBound sessionPermits;
public CachingSessionFactory(SessionFactory sessionFactory) {
this(sessionFactory, DEFAULT_POOL_SIZE);
this(sessionFactory, 0);
}
public CachingSessionFactory(SessionFactory sessionFactory, int maxPoolSize) {
public CachingSessionFactory(SessionFactory sessionFactory, int sessionCacheSize) {
this.sessionFactory = sessionFactory;
this.maxPoolSize = maxPoolSize;
this.queue = new ArrayBlockingQueue<Session>(this.maxPoolSize, true);
this.sessionPermits = new UpperBound(sessionCacheSize);
}
public Session getSession() {
Session session = this.queue.poll();
if (session == null || !session.isOpen()) {
if (session != null && logger.isTraceEnabled()) {
logger.trace("Located session in the pool but it is stale, will create new one.");
}
session = this.sessionFactory.getSession();
if (logger.isTraceEnabled()) {
logger.trace("Created new session");
}
}
else if (logger.isTraceEnabled()) {
logger.trace("Using session from the pool");
/**
* Sets the limit of how long to wait for a session to become available.
*
* @throws {@link IllegalStateException} if the wait expires prior to a Session becoming available.
*/
public void setSessionWaitTimeout(long sessionWaitTimeout) {
this.sessionWaitTimeout = sessionWaitTimeout;
}
public Session getSession() {
boolean permitted = this.sessionPermits.tryAcquire(this.sessionWaitTimeout);
if (!permitted) {
throw new IllegalStateException("Timed out while waiting to aquire a Session.");
}
Session session = this.doGetSession();
return new CachedSession(session);
}
@@ -87,11 +86,25 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean {
}
}
private Session doGetSession() {
Session session = this.queue.poll();
if (session != null && !session.isOpen()) {
if (logger.isDebugEnabled()) {
logger.debug("Received a stale Session, will attempt to get a new one.");
}
return this.doGetSession();
}
else if (session == null){
session = this.sessionFactory.getSession();
}
return session;
}
private void closeSession(Session session) {
try {
if (session != null) {
session.close();
}
}
}
catch (Throwable e) {
if (logger.isWarnEnabled()) {
@@ -111,18 +124,11 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean {
}
public void close() {
if (queue.size() < maxPoolSize) {
if (logger.isTraceEnabled()) {
logger.trace("Releasing target session back to the pool");
}
queue.add(targetSession);
}
else {
if (logger.isTraceEnabled()) {
logger.trace("Disconnecting target session");
}
targetSession.close();
if (logger.isDebugEnabled()){
logger.debug("Releasing Session back to the pool.");
}
queue.add(targetSession);
sessionPermits.release();
}
public boolean remove(String path) throws IOException{

4
spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.1.xsd
View File

@@ -233,7 +233,7 @@ endpoint itself is a Polling Consumer for a channel with a queue.
<xsd:attribute name="cache-sessions" type="xsd:boolean" default="true">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specify whether the Sessions should be cached. Default is true.
[DEPRECATED] Consider wrapping your SessionFactory in an instance of org.springframework.integration.file.remote.session.CachingSessionFactory.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -384,7 +384,7 @@ endpoint itself is a Polling Consumer for a channel with a queue.
<xsd:attribute name="cache-sessions" type="xsd:string" default="true">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specify whether the Sessions should be cached. Default is true.
[DEPRECATED] Consider wrapping your SessionFactory in an instance of org.springframework.integration.file.remote.session.CachingSessionFactory.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>

10
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundChannelAdapterParserTests-context.xml
View File

@@ -15,6 +15,12 @@
<property name="clientMode" value="0"/>
<property name="fileType" value="2"/>
</bean>
<bean id="cachingSessionFactory" class="org.springframework.integration.file.remote.session.CachingSessionFactory">
<constructor-arg ref="ftpSessionFactory"/>
<constructor-arg value="10"/>
<property name="sessionWaitTimeout" value="1000"/>
</bean>
<int-ftp:outbound-channel-adapter id="ftpOutbound"
channel="ftpChannel"
@@ -29,7 +35,7 @@
<int-ftp:outbound-channel-adapter id="ftpOutbound2"
channel="ftpChannel"
session-factory="ftpSessionFactory"
session-factory="cachingSessionFactory"
remote-directory="foo/bar"
charset="UTF-8"
remote-file-separator="."
@@ -39,7 +45,7 @@
<int-ftp:outbound-channel-adapter id="simpleAdapter"
channel="ftpChannel"
session-factory="ftpSessionFactory"
session-factory="cachingSessionFactory"
remote-directory="foo/bar"/>
<int:publish-subscribe-channel id="ftpChannel"/>

98
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/SessionFactoryTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -16,12 +16,25 @@
package org.springframework.integration.ftp.session;
import java.lang.reflect.Field;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.net.ftp.FTPClient;
import org.junit.Ignore;
import org.junit.Test;
import org.mockito.Mockito;
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.test.util.TestUtils;
import static junit.framework.Assert.fail;
import static org.junit.Assert.assertEquals;
/**
* @author Oleg Zhurakousky
*
@@ -49,4 +62,87 @@ public class SessionFactoryTests {
}
}
}
@Test
public void testStaleConnection() throws Exception{
SessionFactory sessionFactory = Mockito.mock(SessionFactory.class);
Session sessionA = Mockito.mock(Session.class);
Session sessionB = Mockito.mock(Session.class);
Mockito.when(sessionA.isOpen()).thenReturn(true);
Mockito.when(sessionB.isOpen()).thenReturn(false);
Mockito.when(sessionFactory.getSession()).thenReturn(sessionA);
Mockito.when(sessionFactory.getSession()).thenReturn(sessionB);
CachingSessionFactory cachingFactory = new CachingSessionFactory(sessionFactory, 2);
Session firstSession = cachingFactory.getSession();
Session secondSession = cachingFactory.getSession();
secondSession.close();
Session nonStaleSession = cachingFactory.getSession();
assertEquals(TestUtils.getPropertyValue(firstSession, "targetSession"), TestUtils.getPropertyValue(nonStaleSession, "targetSession"));
}
@Test
public void testSameSessionFromThePool() throws Exception{
SessionFactory sessionFactory = Mockito.mock(SessionFactory.class);
Session session = Mockito.mock(Session.class);
Mockito.when(sessionFactory.getSession()).thenReturn(session);
CachingSessionFactory cachingFactory = new CachingSessionFactory(sessionFactory, 2);
Session s1 = cachingFactory.getSession();
s1.close();
Session s2 = cachingFactory.getSession();
s2.close();
assertEquals(TestUtils.getPropertyValue(s1, "targetSession"), TestUtils.getPropertyValue(s2, "targetSession"));
Mockito.verify(sessionFactory, Mockito.times(2)).getSession();
}
@Test (expected=IllegalStateException.class) // timeout expire
public void testSessionWaitExpire() throws Exception{
SessionFactory sessionFactory = Mockito.mock(SessionFactory.class);
Session session = Mockito.mock(Session.class);
Mockito.when(sessionFactory.getSession()).thenReturn(session);
CachingSessionFactory cachingFactory = new CachingSessionFactory(sessionFactory, 2);
cachingFactory.setSessionWaitTimeout(3000);
cachingFactory.getSession();
cachingFactory.getSession();
cachingFactory.getSession();
}
@Test
@Ignore
public void testConnectionLimit() throws Exception{
ExecutorService executor = Executors.newCachedThreadPool();
DefaultFtpSessionFactory sessionFactory = new DefaultFtpSessionFactory();
sessionFactory.setHost("192.168.28.143");
sessionFactory.setPassword("password");
sessionFactory.setUsername("user");
final CachingSessionFactory factory = new CachingSessionFactory(sessionFactory, 2);
final Random random = new Random();
final AtomicInteger failures = new AtomicInteger();
for (int i = 0; i < 30; i++) {
executor.execute(new Runnable() {
public void run() {
try {
Session session = factory.getSession();
Thread.sleep(random.nextInt(5000));
session.close();
} catch (Exception e) {
e.printStackTrace();
failures.incrementAndGet();
}
}
});
}
executor.shutdown();
executor.awaitTermination(10000, TimeUnit.SECONDS);
assertEquals(0, failures.get());
}
}

6
spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-2.1.xsd
View File

@@ -37,7 +37,7 @@
<xsd:attribute name="cache-sessions" type="xsd:string" default="true">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specify whether the Sessions should be cached. Default is true.
[DEPRECATED] Consider wrapping your SessionFactory in an instance of org.springframework.integration.file.remote.session.CachingSessionFactory.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -168,7 +168,7 @@ endpoint itself is a Polling Consumer for a channel with a queue.
<xsd:attribute name="cache-sessions" type="xsd:string" default="true">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specify whether the Sessions should be cached. Default is true.
[DEPRECATED] Consider wrapping your SessionFactory in an instance of org.springframework.integration.file.remote.session.CachingSessionFactory.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -302,7 +302,7 @@ endpoint itself is a Polling Consumer for a channel with a queue.
<xsd:attribute name="cache-sessions" type="xsd:boolean" default="true">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specify whether the Sessions should be cached. Default is true.
[DEPRECATED] Consider wrapping your SessionFactory in an instance of org.springframework.integration.file.remote.session.CachingSessionFactory.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>