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

INT-1664, INT-1665 added isOpen() method to the Session strategy, added check for stale connection to the CachingSessionFactory which will ensure that it only returns valid (non-stale) Session

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-12-06 09:06:29 -05:00
parent 0f761f3adb
commit 583b2771cd
5 changed files with 43 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -25,7 +25,6 @@ import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.util.Assert;
/**
* A {@link SessionFactory} implementation that caches Sessions for reuse without
@@ -63,21 +62,23 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean {
this.queue = new ArrayBlockingQueue<Session>(this.maxPoolSize, true);
}
public Session getSession() {
Assert.notNull(this.queue, "SftpSession is unavailable since the pool component is not started");
this.lock.lock();
try {
Session session = this.queue.poll();
if (session != null && !session.isOpen()){
this.queue.remove(session);
session = null;
}
if (null == session) {
session = sessionFactory.getSession();
}
return (session != null) ? new CachedSession(session) : null;
}
finally {
this.lock.unlock();
}
}
public void destroy() {
@@ -96,7 +97,7 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean {
}
catch (Throwable e) {
// log and ignore
logger.warning("Exception was thrown while destroying SftpSession. " + e);
logger.warning("Exception was thrown while destroying Session. " + e);
}
}
@@ -133,6 +134,10 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean {
public void copy(InputStream inputStream, String destination) throws IOException{
this.targetSession.copy(inputStream, destination);
}
public boolean isOpen() {
return this.targetSession.isOpen();
}
}
}

2
spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java
View File

@@ -40,4 +40,6 @@ public interface Session {
void copy(InputStream inputStream, String destination) throws IOException;
void close();
boolean isOpen();
}

57
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/AbstractFtpSessionFactory.java
View File

@@ -18,8 +18,6 @@ package org.springframework.integration.ftp.session;
import java.io.IOException;
import java.net.SocketException;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.net.ftp.FTP;
@@ -27,7 +25,6 @@ import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPClientConfig;
import org.apache.commons.net.ftp.FTPReply;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.integration.MessagingException;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
@@ -139,7 +136,6 @@ public abstract class AbstractFtpSessionFactory<T extends FTPClient> implements
}
}
@SuppressWarnings("unchecked")
private T createClient() throws SocketException, IOException {
final T client = this.createClientInstance();
Assert.notNull(client, "client must not be null");
@@ -148,37 +144,28 @@ public abstract class AbstractFtpSessionFactory<T extends FTPClient> implements
this.postProcessClientBeforeConnect(client);
ProxyFactory factory = new ProxyFactory(client);
factory.setProxyTargetClass(true);
factory.addAdvice(new MethodInterceptor() {
public Object invoke(MethodInvocation invocation) throws Throwable {
String methodName = invocation.getMethod().getName();
if (!methodName.endsWith("connect")){ // will take care of both 'connect' and 'disconnect'
if (!client.isConnected()){
client.connect(host);
postProcessClientAfterConnect (client);
if (!FTPReply.isPositiveCompletion(client.getReplyCode())) {
throw new MessagingException("Connecting to server [" +
host + ":" + port + "] failed. Please check the connection.");
}
logger.debug("Connected to server [" + host + ":" + port + "]");
if (!client.login(username, password)) {
throw new IllegalStateException("Login failed. The respponse from the server is: " +
client.getReplyString());
}
updateClientMode(client);
client.setFileType(fileType);
client.setBufferSize(bufferSize);
}
}
return invocation.proceed();
}
});
return (T) factory.getProxy();
// Connect
client.connect(host);
if (!FTPReply.isPositiveCompletion(client.getReplyCode())) {
throw new MessagingException("Connecting to server [" +
host + ":" + port + "] failed. Please check the connection.");
}
logger.debug("Connected to server [" + host + ":" + port + "]");
// Login
if (!client.login(username, password)) {
throw new IllegalStateException("Login failed. The respponse from the server is: " +
client.getReplyString());
}
this.postProcessClientAfterConnect(client);
this.updateClientMode(client);
client.setFileType(fileType);
client.setBufferSize(bufferSize);
return client;
}
/**

4
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java
View File

@@ -109,4 +109,8 @@ class FtpSession implements Session {
}
}
}
public boolean isOpen() {
return this.client.isConnected();
}
}

5
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java
View File

@@ -39,6 +39,7 @@ import com.jcraft.jsch.SftpException;
* @author Josh Long
* @author Mario Gray
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.0
*/
class SftpSession implements Session {
@@ -136,4 +137,8 @@ class SftpSession implements Session {
}
}
public boolean isOpen() {
return this.channel.isConnected() && this.jschSession.isConnected();
}
}