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

INT-2146

Browse Source 
Expose session limitation on remote file adapters

  Deprecate 'cache-sessions' attribute

  Add 'sessionCacheSize' and 'sessionWaitTimeout' attributes on CachingSessionFactory

  Update documentation
This commit is contained in:
Oleg Zhurakousky
2011-10-14 14:31:57 -04:00
committed by Mark Fisher
parent 761a797ee1
commit 7f92089584
10 changed files with 292 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -18,12 +18,16 @@ 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;
/**
@@ -34,23 +38,35 @@ import org.springframework.util.StringUtils;
* @since 2.0
*/
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 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.warn("The 'cache-sessions' attribute is deprecated since v2.1. Consider configuring CachingSessionFactory explicitly");
}
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");

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

@@ -15,20 +15,27 @@
*/
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 {
private final Log logger = LogFactory.getLog(this.getClass());
@Override
protected String getInputChannelAttributeName() {
@@ -39,16 +46,30 @@ public abstract class AbstractRemoteFileOutboundGatewayParser extends
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"));
// 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.warn("The 'cache-sessions' attribute is deprecated since v2.1. Consider configuring CachingSessionFactory explicitly");
}
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");

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");

93
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
@@ -37,46 +36,47 @@ import org.springframework.beans.factory.DisposableBean;
* @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);
public static final int DEFAULT_POOL_SIZE = 10;
private volatile long sessionWaitTimeout = Integer.MAX_VALUE;
private final Queue<Session> queue;
private volatile LinkedBlockingQueue<Session> queue = new LinkedBlockingQueue<Session>();
private final SessionFactory sessionFactory;
private final int maxPoolSize;
private final UpperBound sessionSizeManager;
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.sessionSizeManager = 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}.
*/
public void setSessionWaitTimeout(long sessionWaitTimeout) {
this.sessionWaitTimeout = sessionWaitTimeout;
}
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");
}
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);
}
else if (logger.isTraceEnabled()) {
logger.trace("Using session from the pool");
}
return new CachedSession(session);
}
public void destroy() {
@@ -86,6 +86,21 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean {
}
}
}
private Session doGetSession() throws InterruptedException {
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");
}
return this.doGetSession();
}
else if (session == null){
session = this.sessionFactory.getSession();
}
return session;
}
private void closeSession(Session session) {
try {
@@ -111,18 +126,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 into the pool");
}
queue.add(targetSession);
sessionSizeManager.release();
}
public boolean remove(String path) throws IOException{
@@ -153,5 +161,4 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean {
this.targetSession.mkdir(directory);
}
}
}