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

polishing

Browse Source
This commit is contained in:
Mark Fisher
2011-10-19 17:03:23 -04:00
parent 7f92089584
commit 7581c1cea4
7 changed files with 92 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -346,22 +346,21 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/ftp
<section id="ftp-session-caching">
<title>FTP Session Caching</title>
<para>
Since 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. And although we did expose <code>cache-session</code> attribute which would
allow you to turn auto caching off it was still not sufficient when it came to other session caching attributes.
For example; One of the requirement we received was to control session limit if remote server imposes client connection limit.
Now such behavior is controlled by the <classname>CachingSessionFactory</classname> and its <code>sessionCacheSize</code> and
<code>sessionWaitTimeout</code> properties. As its name suggest <code>sessionCacheSize</code> property controls how many active
sessions this adapter will maintain in its cache (DEFAULT unbounded). If <code>sessionCacheSize</code> threshold has been reached any
attempt to get more session will block until session becomes available or until wait time for a session to become available
expires (DEFAULT Integer.MAX_VALUE). The wait time for a session to become available can also be controlled via <code>sessionWaitTimeout</code>
property.
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>
Since version 2.1 if you need you session to be cahced you can simply configure your default Session Factory as
described above and than wrap it in the <classname>CachingSessionFactory</classname> while providing additional properties.
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>
@@ -372,8 +371,8 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/ftp
<property name="sessionWaitTimeout" value="1000"/>
</bean>]]></programlisting>
In the above example you see <classname>CachingSessionFactory</classname> created with
<code>sessionCacheSize</code> set to 10 with <code>sessionWait</code> timeout set to 1 second.
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).
</para>
</section>

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

@@ -289,22 +289,22 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp
</section>
<section id="sftp-session-caching">
<title>SFTP Session Caching</title>
<para>
Since 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. And although we did expose <code>cache-session</code> attribute which would
allow you to turn auto caching off it was still not sufficient when it came to other session caching attributes.
For example; One of the requirement we received was to control session limit if remote server imposes client connection limit.
Now such behavior is controlled by the <classname>CachingSessionFactory</classname> and its <code>sessionCacheSize</code> and
<code>sessionWaitTimeout</code> properties. As its name suggest <code>sessionCacheSize</code> property controls how many active
sessions this adapter will maintain in its cache (DEFAULT unbounded). If <code>sessionCacheSize</code> threshold has been reached any
attempt to get more session will block until session becomes available or until wait time for a session to become available
expires (DEFAULT Integer.MAX_VALUE). The wait time for a session to become available can also be controlled via <code>sessionWaitTimeout</code>
property.
<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>
Since version 2.1 if you need you session to be cahced you can simply configure your default Session Factory as
described above and than wrap it in the <classname>CachingSessionFactory</classname> while providing additional properties.
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>
@@ -315,8 +315,8 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp
<property name="sessionWaitTimeout" value="1000"/>
</bean>]]></programlisting>
In the above example you see <classname>CachingSessionFactory</classname> created with
<code>sessionCacheSize</code> set to 10 with <code>sessionWait</code> timeout set to 1 second.
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).
</para>
</section>

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

@@ -20,6 +20,7 @@ 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;
@@ -38,6 +39,7 @@ import org.springframework.util.StringUtils;
* @since 2.0
*/
public abstract class AbstractRemoteFileInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
private final Log logger = LogFactory.getLog(this.getClass());
@Override
@@ -45,22 +47,23 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
BeanDefinitionBuilder synchronizerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
this.getInboundFileSynchronizerClassname());
// This whole block must be refactored once cache-session attribute is removed
// 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())){
if (StringUtils.hasText(sessionFactoryClassName) && sessionFactoryClassName.endsWith(CachingSessionFactory.class.getName())) {
synchronizerBuilder.addConstructorArgValue(sessionFactoryDefinition);
}
else {
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 (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 {
else {
BeanDefinitionBuilder sessionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(CachingSessionFactory.class);
sessionFactoryBuilder.addConstructorArgReference(sessionFactoryName);
synchronizerBuilder.addConstructorArgValue(sessionFactoryBuilder.getBeanDefinition());
@@ -71,14 +74,7 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
// 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");
@@ -88,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();
}

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

@@ -30,10 +30,8 @@ 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());
@@ -45,19 +43,21 @@ 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
// 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())){
if (StringUtils.hasText(sessionFactoryClassName) && sessionFactoryClassName.endsWith(CachingSessionFactory.class.getName())) {
builder.addConstructorArgValue(sessionFactoryDefinition);
}
else {
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 (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"));
@@ -69,7 +69,7 @@ public abstract class AbstractRemoteFileOutboundGatewayParser extends
}
}
// 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");
@@ -96,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);

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

@@ -36,47 +36,46 @@ import org.springframework.integration.util.UpperBound;
* @author Mark Fisher
* @since 2.0
*/
public class CachingSessionFactory implements SessionFactory, DisposableBean{
public class CachingSessionFactory implements SessionFactory, DisposableBean {
private static final Log logger = LogFactory.getLog(CachingSessionFactory.class);
private volatile long sessionWaitTimeout = Integer.MAX_VALUE;
private volatile LinkedBlockingQueue<Session> queue = new LinkedBlockingQueue<Session>();
private final LinkedBlockingQueue<Session> queue = new LinkedBlockingQueue<Session>();
private final SessionFactory sessionFactory;
private final UpperBound sessionSizeManager;
private final UpperBound sessionPermits;
public CachingSessionFactory(SessionFactory sessionFactory) {
this(sessionFactory, 0);
}
public CachingSessionFactory(SessionFactory sessionFactory, int sessionCacheSize) {
this.sessionFactory = sessionFactory;
this.sessionSizeManager = new UpperBound(sessionCacheSize);
this.sessionPermits = new UpperBound(sessionCacheSize);
}
/**
* Sets the limit of how long it will wait for a session to become available after which
* it will throw {@link IllegalStateException}.
* 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() {
try {
boolean permitted = this.sessionSizeManager.tryAcquire(this.sessionWaitTimeout);
if (!permitted){
throw new IllegalStateException("Timed out while waiting to aquire Session");
}
Session session = this.doGetSession();
return new CachedSession(session);
} catch (Exception e) {
Thread.currentThread().interrupt();
throw new IllegalStateException("Exception was received during attempt to obtain Session", e);
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);
}
public void destroy() {
@@ -86,13 +85,12 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean{
}
}
}
private Session doGetSession() throws InterruptedException {
Session session = this.queue.poll();
if (session != null && !session.isOpen()){
private Session doGetSession() {
Session session = this.queue.poll();
if (session != null && !session.isOpen()) {
if (logger.isDebugEnabled()) {
logger.debug("Received stale Session, will attempt to get a new one");
logger.debug("Received a stale Session, will attempt to get a new one.");
}
return this.doGetSession();
}
@@ -106,7 +104,7 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean{
try {
if (session != null) {
session.close();
}
}
}
catch (Throwable e) {
if (logger.isWarnEnabled()) {
@@ -127,10 +125,10 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean{
public void close() {
if (logger.isDebugEnabled()){
logger.debug("Releasing Session back into the pool");
logger.debug("Releasing Session back to the pool.");
}
queue.add(targetSession);
sessionSizeManager.release();
sessionPermits.release();
}
public boolean remove(String path) throws IOException{
@@ -161,4 +159,5 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean{
this.targetSession.mkdir(directory);
}
}
}

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[
[DEPRECATED] Consider wrapping you SessionFactory in the org.springframework.integration.file.remote.session.CachingSessionFactory.
[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[
[DEPRECATED] Consider wrapping you SessionFactory in the org.springframework.integration.file.remote.session.CachingSessionFactory.
[DEPRECATED] Consider wrapping your SessionFactory in an instance of org.springframework.integration.file.remote.session.CachingSessionFactory.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>

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[
[DEPRECATED] Consider wrapping you SessionFactory in the org.springframework.integration.file.remote.session.CachingSessionFactory.
[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[
[DEPRECATED] Consider wrapping you SessionFactory in the org.springframework.integration.file.remote.session.CachingSessionFactory.
[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[
[DEPRECATED] Consider wrapping you SessionFactory in the org.springframework.integration.file.remote.session.CachingSessionFactory.
[DEPRECATED] Consider wrapping your SessionFactory in an instance of org.springframework.integration.file.remote.session.CachingSessionFactory.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>