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

INT-1614 renamed QueuedSftpSessionPool to CachingSftpSessionFactory and it no longer implements Lifecycle

Browse Source
This commit is contained in:
Mark Fisher
2010-11-19 10:43:55 -05:00
parent 66fea3257e
commit f08299c8b9
5 changed files with 46 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java
View File

@@ -38,7 +38,6 @@ public class SftpInboundChannelAdapterParser extends AbstractPollingInboundChann
@Override
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
String sessionFactoryName = element.getAttribute("session-factory");
String autoStartup = element.getAttribute("auto-startup");
String fileNamePattern = element.getAttribute("filename-pattern");
String filter = element.getAttribute("filter");
boolean hasFileNamePattern = StringUtils.hasText(fileNamePattern);
@@ -49,12 +48,11 @@ public class SftpInboundChannelAdapterParser extends AbstractPollingInboundChann
"is allowed on SFTP inbound adapter");
}
}
BeanDefinitionBuilder sessionPollBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.sftp.session.QueuedSftpSessionPool");
sessionPollBuilder.addConstructorArgReference(sessionFactoryName);
sessionPollBuilder.addPropertyValue("autoStartup", autoStartup);
BeanDefinitionBuilder sessionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.sftp.session.CachingSftpSessionFactory");
sessionFactoryBuilder.addConstructorArgReference(sessionFactoryName);
String sessionPollName = BeanDefinitionReaderUtils.registerWithGeneratedName(
sessionPollBuilder.getBeanDefinition(), parserContext.getRegistry());
sessionFactoryBuilder.getBeanDefinition(), parserContext.getRegistry());
BeanDefinitionBuilder synchronizerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.sftp.inbound.SftpInboundSynchronizer");
synchronizerBuilder.addConstructorArgReference(sessionPollName);

2
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpOutboundChannelAdapterParser.java
View File

@@ -40,7 +40,7 @@ public class SftpOutboundChannelAdapterParser extends AbstractOutboundChannelAda
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder sessionPoolBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.sftp.session.QueuedSftpSessionPool");
"org.springframework.integration.sftp.session.CachingSftpSessionFactory");
sessionPoolBuilder.addConstructorArgReference(element.getAttribute("session-factory"));
String sessionPoolName = BeanDefinitionReaderUtils.registerWithGeneratedName(
sessionPoolBuilder.getBeanDefinition(), parserContext.getRegistry());

118
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/QueuedSftpSessionPool.java → spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/CachingSftpSessionFactory.java
View File

@@ -21,69 +21,56 @@ import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;
import org.springframework.context.SmartLifecycle;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.util.Assert;
import com.jcraft.jsch.Channel;
import com.jcraft.jsch.ChannelSftp;
/**
* This approach - of having a SessionPool ({@link SftpSessionPool}) that has an
* implementation of a queued SessionPool ({@link QueuedSftpSessionPool}) - was
* implementation of a queued SessionPool ({@link CachingSftpSessionFactory}) - was
* taken almost directly from the Spring Integration FTP adapter.
*
* @author Josh Long
* @author Oleg Zhurakousky
* @since 2.0
*/
public class QueuedSftpSessionPool implements SftpSessionFactory, SmartLifecycle {
public class CachingSftpSessionFactory implements SftpSessionFactory, DisposableBean {
private static Logger logger = Logger.getLogger(QueuedSftpSessionPool.class.getName());
private static Logger logger = Logger.getLogger(CachingSftpSessionFactory.class.getName());
public static final int DEFAULT_POOL_SIZE = 10;
private volatile Queue<SftpSession> queue;
private final Queue<SftpSession> queue;
private final SimpleSftpSessionFactory sftpSessionFactory;
private final int maxPoolSize;
private volatile boolean running;
private volatile boolean autoStartup;
private volatile int phase = 0;
private final ReentrantLock lock = new ReentrantLock();
public QueuedSftpSessionPool(SimpleSftpSessionFactory factory) {
this(DEFAULT_POOL_SIZE, factory);
public CachingSftpSessionFactory(SimpleSftpSessionFactory sessionFactory) {
this(sessionFactory, DEFAULT_POOL_SIZE);
}
public QueuedSftpSessionPool(int maxPoolSize, SimpleSftpSessionFactory sessionFactory) {
public CachingSftpSessionFactory(SimpleSftpSessionFactory sessionFactory, int maxPoolSize) {
this.sftpSessionFactory = sessionFactory;
this.maxPoolSize = maxPoolSize;
this.queue = new ArrayBlockingQueue<SftpSession>(this.maxPoolSize, true);
}
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
public void setPhase(int phase) {
this.phase = phase;
}
public SftpSession getSession() {
Assert.notNull(this.queue, "SftpSession is unavailable since the pool component is not started");
this.lock.lock();
try {
SftpSession session = this.queue.poll();
if (null == session) {
session = this.sftpSessionFactory.getSession();
session = sftpSessionFactory.getSession();
}
return session;
return (session != null) ? new PooledSftpSession(session) : null;
}
finally {
this.lock.unlock();
@@ -91,34 +78,17 @@ public class QueuedSftpSessionPool implements SftpSessionFactory, SmartLifecycle
}
public void release(SftpSession sftpSession) {
if (this.running) {
this.lock.lock();
try {
if (queue.size() < maxPoolSize && sftpSession != null) {
queue.add(sftpSession);
}
else {
this.destroySftpSession(sftpSession);
}
public void destroy() {
if (this.queue != null) {
for (SftpSession sftpSession : this.queue) {
this.destroySftpSession(sftpSession);
}
finally {
this.lock.unlock();
}
}
else {
this.destroySftpSession(sftpSession);
}
}
private void destroySftpSession(SftpSession sftpSession) {
try {
if (sftpSession != null) {
Channel channel = sftpSession.getChannel();
if (channel.isConnected()) {
channel.disconnect();
}
sftpSession.disconnect();
}
}
@@ -129,50 +99,30 @@ public class QueuedSftpSessionPool implements SftpSessionFactory, SmartLifecycle
}
// SmartLifeycle implementation
private class PooledSftpSession implements SftpSession {
public boolean isAutoStartup() {
return this.autoStartup;
}
private final SftpSession targetSession;
public int getPhase() {
return this.phase;
}
public boolean isRunning() {
return this.running;
}
public void start() {
Assert.isTrue(this.maxPoolSize > 0, "poolSize must be greater than 0");
this.lock.lock();
try {
this.queue = new ArrayBlockingQueue<SftpSession>(this.maxPoolSize, true);
this.running = true;
private PooledSftpSession(SftpSession targetSession) {
this.targetSession = targetSession;
}
finally {
this.lock.unlock();
}
}
public void stop() {
if (this.queue != null) {
for (SftpSession sftpSession : this.queue) {
this.destroySftpSession(sftpSession);
public ChannelSftp getChannel() {
return targetSession.getChannel();
}
public void connect() {
targetSession.connect();
}
public void disconnect() {
if (queue.size() < maxPoolSize) {
queue.add(targetSession);
}
else {
targetSession.disconnect();
}
}
}
public void stop(Runnable callback) {
this.lock.lock();
try {
this.stop();
callback.run();
}
finally {
this.running = false;
this.lock.unlock();
}
}
}

4
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSession.java
View File

@@ -20,6 +20,7 @@ import java.io.InputStream;
import org.apache.commons.lang.StringUtils;
import com.jcraft.jsch.Channel;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.JSchException;
@@ -122,6 +123,9 @@ public class DefaultSftpSession implements SftpSession {
public void disconnect() {
if (targetSession.isConnected()) {
targetSession.disconnect();
if (channel.isConnected()) {
channel.disconnect();
}
}
}

5
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/OutboundChannelAdapaterParserTests.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.sftp.config;
import static junit.framework.Assert.assertEquals;
@@ -30,7 +31,7 @@ import org.springframework.expression.spel.standard.SpelExpression;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.file.FileNameGenerator;
import org.springframework.integration.sftp.outbound.SftpSendingMessageHandler;
import org.springframework.integration.sftp.session.QueuedSftpSessionPool;
import org.springframework.integration.sftp.session.CachingSftpSessionFactory;
import org.springframework.integration.sftp.session.SimpleSftpSessionFactory;
import org.springframework.integration.test.util.TestUtils;
@@ -56,7 +57,7 @@ public class OutboundChannelAdapaterParserTests {
assertEquals("UTF-8", TestUtils.getPropertyValue(handler, "charset"));
assertNotNull(TestUtils.getPropertyValue(handler, "temporaryBufferFolder"));
assertNotNull(TestUtils.getPropertyValue(handler, "temporaryBufferFolderFile"));
QueuedSftpSessionPool sessionFactory = (QueuedSftpSessionPool) TestUtils.getPropertyValue(handler, "sessionFactory");
CachingSftpSessionFactory sessionFactory = (CachingSftpSessionFactory) TestUtils.getPropertyValue(handler, "sessionFactory");
SimpleSftpSessionFactory clientFactory = (SimpleSftpSessionFactory) TestUtils.getPropertyValue(sessionFactory, "sftpSessionFactory");
assertEquals("localhost", TestUtils.getPropertyValue(clientFactory, "host"));
assertEquals(2222, TestUtils.getPropertyValue(clientFactory, "port"));