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

INT-1614 moved Session and SessionFactory from SFTP into more general 'remote' package in the file module

Browse Source
This commit is contained in:
Mark Fisher
2010-11-19 15:47:45 -05:00
parent 67bac6d61c
commit c7c0972ce2
11 changed files with 93 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java → spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java
View File

@@ -14,24 +14,20 @@
* limitations under the License.
*/
package org.springframework.integration.sftp.session;
package org.springframework.integration.file.remote.session;
import java.io.InputStream;
import java.util.Collection;
/**
* There are many ways to create a {@link SftpSession} just as there are many ways to SSH into a remote system.
* You may use a username and password, you may use a username and private key, you may use a username and a private key with a password, etc.
* <p/>
* This object represents the connection to the remote server, and to use it you must provide it with all the components you'd normally provide an
* incantation of the <code>ssh</code> command.
* Common abstraction for a Session with a remote File system.
*
* @author Josh Long
* @author Mario Gray
* @author Mark Fisher
* @since 2.0
*/
public interface SftpSession {
public interface Session {
void connect();

8
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSessionFactory.java → spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SessionFactory.java
View File

@@ -14,14 +14,16 @@
* limitations under the License.
*/
package org.springframework.integration.sftp.session;
package org.springframework.integration.file.remote.session;
/**
* Factory for acquiring {@link Session} instances.
*
* @author Mark Fisher
* @since 2.0
*/
public interface SftpSessionFactory {
public interface SessionFactory {
SftpSession getSession();
Session getSession();
}

20
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java
View File

@@ -25,10 +25,10 @@ import org.apache.commons.io.IOUtils;
import org.springframework.core.io.Resource;
import org.springframework.integration.MessagingException;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.file.synchronizer.AbstractInboundFileSynchronizer;
import org.springframework.integration.file.synchronizer.AbstractInboundFileSynchronizingMessageSource;
import org.springframework.integration.sftp.session.SftpSession;
import org.springframework.integration.sftp.session.SftpSessionFactory;
import org.springframework.util.Assert;
import com.jcraft.jsch.ChannelSftp;
@@ -48,12 +48,12 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer
private volatile String remotePath;
/**
* the pool of {@link org.springframework.integration.sftp.session.SftpSessionPool} SFTP sessions
* the {@link SessionFactory} for acquiring SFTP Sessions.
*/
private final SftpSessionFactory sessionFactory;
private final SessionFactory sessionFactory;
public SftpInboundFileSynchronizer(SftpSessionFactory sessionFactory) {
public SftpInboundFileSynchronizer(SessionFactory sessionFactory) {
Assert.notNull(sessionFactory, "sessionFactory must not be null");
this.sessionFactory = sessionFactory;
}
@@ -71,7 +71,7 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer
}
public void synchronizeToLocalDirectory(Resource localDirectory) {
SftpSession session = null;
Session session = null;
try {
session = this.sessionFactory.getSession();
if (logger.isTraceEnabled()) {
@@ -100,7 +100,7 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer
}
}
private boolean copyFromRemoteToLocalDirectory(SftpSession sftpSession, ChannelSftp.LsEntry entry, Resource localDir) throws Exception {
private boolean copyFromRemoteToLocalDirectory(Session session, ChannelSftp.LsEntry entry, Resource localDir) throws Exception {
File fileForLocalDir = localDir.getFile();
File localFile = new File(fileForLocalDir, entry.getFilename());
if (!localFile.exists()) {
@@ -111,7 +111,7 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer
AbstractInboundFileSynchronizingMessageSource.INCOMPLETE_EXTENSION);
fileOutputStream = new FileOutputStream(tmpLocalTarget);
String remoteFqPath = this.remotePath + "/" + entry.getFilename();
in = sftpSession.get(remoteFqPath);
in = session.get(remoteFqPath);
try {
IOUtils.copy(in, fileOutputStream);
}
@@ -120,7 +120,7 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer
IOUtils.closeQuietly(fileOutputStream);
}
if (tmpLocalTarget.renameTo(localFile)) {
this.acknowledge(sftpSession, entry);
this.acknowledge(session, entry);
}
return true;
}
@@ -142,7 +142,7 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer
private class DeletionEntryAcknowledgmentStrategy implements AbstractInboundFileSynchronizer.EntryAcknowledgmentStrategy<ChannelSftp.LsEntry> {
public void acknowledge(Object useful, ChannelSftp.LsEntry msg) throws Exception {
SftpSession sftpSession = (SftpSession) useful;
Session sftpSession = (Session) useful;
String remoteFqPath = remotePath + "/" + msg.getFilename();
sftpSession.rm(remoteFqPath);
if (logger.isDebugEnabled()) {

10
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/outbound/SftpSendingMessageHandler.java
View File

@@ -35,10 +35,10 @@ import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.MessagingException;
import org.springframework.integration.file.DefaultFileNameGenerator;
import org.springframework.integration.file.FileNameGenerator;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.sftp.session.SftpSession;
import org.springframework.integration.sftp.session.SftpSessionFactory;
import org.springframework.util.Assert;
import org.springframework.util.FileCopyUtils;
import org.springframework.util.StringUtils;
@@ -55,7 +55,7 @@ public class SftpSendingMessageHandler extends AbstractMessageHandler {
private static final String TEMPORARY_FILE_SUFFIX = ".writing";
private final SftpSessionFactory sessionFactory;
private final SessionFactory sessionFactory;
private volatile ExpressionEvaluatingMessageProcessor<String> directoryExpressionProcesor;
@@ -70,7 +70,7 @@ public class SftpSendingMessageHandler extends AbstractMessageHandler {
private volatile String charset = Charset.defaultCharset().name();
public SftpSendingMessageHandler(SftpSessionFactory sessionFactory) {
public SftpSendingMessageHandler(SessionFactory sessionFactory) {
Assert.notNull(sessionFactory, "sessionFactory must not be null");
this.sessionFactory = sessionFactory;
}
@@ -163,7 +163,7 @@ public class SftpSendingMessageHandler extends AbstractMessageHandler {
}
private boolean sendFileToRemoteEndpoint(Message<?> message, File file) throws Exception {
SftpSession session = this.sessionFactory.getSession();
Session session = this.sessionFactory.getSession();
if (session == null) {
throw new MessagingException("The session returned from the pool is null, cannot proceed.");
}

28
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/CachingSftpSessionFactory.java
View File

@@ -24,6 +24,8 @@ import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.util.Assert;
/**
@@ -35,14 +37,14 @@ import org.springframework.util.Assert;
* @author Oleg Zhurakousky
* @since 2.0
*/
public class CachingSftpSessionFactory implements SftpSessionFactory, DisposableBean {
public class CachingSftpSessionFactory implements SessionFactory, DisposableBean {
private static Logger logger = Logger.getLogger(CachingSftpSessionFactory.class.getName());
public static final int DEFAULT_POOL_SIZE = 10;
private final Queue<SftpSession> queue;
private final Queue<Session> queue;
private final SimpleSftpSessionFactory sftpSessionFactory;
@@ -58,15 +60,15 @@ public class CachingSftpSessionFactory implements SftpSessionFactory, Disposable
public CachingSftpSessionFactory(SimpleSftpSessionFactory sessionFactory, int maxPoolSize) {
this.sftpSessionFactory = sessionFactory;
this.maxPoolSize = maxPoolSize;
this.queue = new ArrayBlockingQueue<SftpSession>(this.maxPoolSize, true);
this.queue = new ArrayBlockingQueue<Session>(this.maxPoolSize, true);
}
public SftpSession getSession() {
public Session getSession() {
Assert.notNull(this.queue, "SftpSession is unavailable since the pool component is not started");
this.lock.lock();
try {
SftpSession session = this.queue.poll();
Session session = this.queue.poll();
if (null == session) {
session = sftpSessionFactory.getSession();
}
@@ -80,16 +82,16 @@ public class CachingSftpSessionFactory implements SftpSessionFactory, Disposable
public void destroy() {
if (this.queue != null) {
for (SftpSession sftpSession : this.queue) {
this.destroySftpSession(sftpSession);
for (Session session : this.queue) {
this.destroySession(session);
}
}
}
private void destroySftpSession(SftpSession sftpSession) {
private void destroySession(Session session) {
try {
if (sftpSession != null) {
sftpSession.disconnect();
if (session != null) {
session.disconnect();
}
}
catch (Throwable e) {
@@ -99,11 +101,11 @@ public class CachingSftpSessionFactory implements SftpSessionFactory, Disposable
}
private class PooledSftpSession implements SftpSession {
private class PooledSftpSession implements Session {
private final SftpSession targetSession;
private final Session targetSession;
private PooledSftpSession(SftpSession targetSession) {
private PooledSftpSession(Session targetSession) {
this.targetSession = targetSession;
}

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

@@ -23,10 +23,11 @@ import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.file.remote.session.Session;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.Session;
import com.jcraft.jsch.SftpATTRS;
import com.jcraft.jsch.SftpException;
import com.jcraft.jsch.UserInfo;
@@ -39,13 +40,13 @@ import com.jcraft.jsch.UserInfo;
* @author Mark Fisher
* @since 2.0
*/
public class DefaultSftpSession implements SftpSession {
public class DefaultSftpSession implements Session {
private final Log logger = LogFactory.getLog(this.getClass());
private volatile ChannelSftp channel;
private volatile Session targetSession;
private volatile com.jcraft.jsch.Session jschSession;
private String privateKey;
@@ -101,14 +102,14 @@ public class DefaultSftpSession implements SftpSession {
jSch.addIdentity(this.privateKey);
}
}
this.targetSession = jSch.getSession(userName, hostName, port);
this.jschSession = jSch.getSession(userName, hostName, port);
if (!StringUtils.isEmpty(userPassword)) {
this.targetSession.setPassword(userPassword);
this.jschSession.setPassword(userPassword);
}
this.userInfo = new OptimisticUserInfoImpl(userPassword);
this.targetSession.setUserInfo(userInfo);
this.targetSession.connect();
this.channel = (ChannelSftp) this.targetSession.openChannel("sftp");
this.jschSession.setUserInfo(userInfo);
this.jschSession.connect();
this.channel = (ChannelSftp) this.jschSession.openChannel("sftp");
}
public ChannelSftp getChannel() {
@@ -127,8 +128,8 @@ public class DefaultSftpSession implements SftpSession {
}
public void disconnect() {
if (targetSession.isConnected()) {
targetSession.disconnect();
if (jschSession.isConnected()) {
jschSession.disconnect();
if (channel.isConnected()) {
channel.disconnect();
}

6
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SimpleSftpSessionFactory.java
View File

@@ -17,6 +17,8 @@
package org.springframework.integration.sftp.session;
import org.springframework.core.io.Resource;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -28,7 +30,7 @@ import org.springframework.util.StringUtils;
* @author Mario Gray
* @since 2.0
*/
public class SimpleSftpSessionFactory implements SftpSessionFactory {
public class SimpleSftpSessionFactory implements SessionFactory {
private volatile String host;
@@ -73,7 +75,7 @@ public class SimpleSftpSessionFactory implements SftpSessionFactory {
this.privateKeyPassphrase = privateKeyPassphrase;
}
public SftpSession getSession() {
public Session getSession() {
Assert.hasText(this.host, "host must not be empty");
Assert.hasText(this.user, "user must not be empty");
Assert.isTrue(this.port >= 0, "port must be a positive number");

16
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizerTests.java
View File

@@ -27,10 +27,10 @@ import org.junit.Test;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.file.synchronizer.AbstractInboundFileSynchronizer.EntryAcknowledgmentStrategy;
import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer;
import org.springframework.integration.sftp.session.SftpSession;
import org.springframework.integration.sftp.session.SftpSessionFactory;
import org.springframework.util.ReflectionUtils;
import com.jcraft.jsch.ChannelSftp;
@@ -65,11 +65,11 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
*/
@Test
public void testCopyAndRenameWhenLocalFileExists() throws Exception {
SftpInboundFileSynchronizer synchronizer = new SftpInboundFileSynchronizer(mock(SftpSessionFactory.class));
SftpInboundFileSynchronizer synchronizer = new SftpInboundFileSynchronizer(mock(SessionFactory.class));
Method method =
ReflectionUtils.findMethod(synchronizer.getClass(), "copyFromRemoteToLocalDirectory", SftpSession.class, LsEntry.class, Resource.class);
ReflectionUtils.findMethod(synchronizer.getClass(), "copyFromRemoteToLocalDirectory", Session.class, LsEntry.class, Resource.class);
method.setAccessible(true);
SftpSession session = mock(SftpSession.class);
Session session = mock(Session.class);
LsEntry entry = mock(LsEntry.class);
when(entry.getFilename()).thenReturn("foo.txt");
Resource localDir = new FileSystemResource(new File("target"));
@@ -87,12 +87,12 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
@org.junit.Ignore
@Test
public void testCopyAndRenameWhenLocalFileDoesntExist() throws Exception {
SftpInboundFileSynchronizer synchronizer = new SftpInboundFileSynchronizer(mock(SftpSessionFactory.class));
SftpInboundFileSynchronizer synchronizer = new SftpInboundFileSynchronizer(mock(SessionFactory.class));
synchronizer.setEntryAcknowledgmentStrategy(mock(EntryAcknowledgmentStrategy.class));
Method method =
ReflectionUtils.findMethod(synchronizer.getClass(), "copyFromRemoteToLocalDirectory", SftpSession.class, LsEntry.class, Resource.class);
ReflectionUtils.findMethod(synchronizer.getClass(), "copyFromRemoteToLocalDirectory", Session.class, LsEntry.class, Resource.class);
method.setAccessible(true);
SftpSession session = mock(SftpSession.class);
Session session = mock(Session.class);
ChannelSftp channelSftp = mock(ChannelSftp.class);
File originalFile = new File("pom.xml");
when(channelSftp.get("null/bar.txt")).thenReturn(new FileInputStream(originalFile));

10
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java
View File

@@ -36,8 +36,8 @@ import org.mockito.stubbing.Answer;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.integration.sftp.session.SftpSession;
import org.springframework.integration.sftp.session.SftpSessionFactory;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.ChannelSftp.LsEntry;
@@ -45,7 +45,7 @@ import com.jcraft.jsch.SftpATTRS;
/**
* @author Oleg Zhurakousky
*
* @since 2.0
*/
public class SftpInboundRemoteFileSystemSynchronizerTests {
@@ -57,7 +57,7 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
if (file.exists()){
file.delete();
}
SftpSessionFactory sessionFactory = mock(SftpSessionFactory.class);
SessionFactory sessionFactory = mock(SessionFactory.class);
SftpInboundFileSynchronizer syncronizer = new SftpInboundFileSynchronizer(sessionFactory);
syncronizer.setRemotePath("foo/bar");
@@ -66,7 +66,7 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
syncronizer.setFilter(filter);
SftpSession sftpSession = mock(SftpSession.class);
Session sftpSession = mock(Session.class);
when(sessionFactory.getSession()).thenReturn(sftpSession);
final ChannelSftp channel = mock(ChannelSftp.class);

19
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/Temp.java Normal file
View File

@@ -0,0 +1,19 @@
package org.springframework.integration.sftp.inbound;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
public class Temp {
public static void main(String[] args) throws Exception {
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
//client.retrieveFile("path", outputStream);
ByteArrayInputStream inputStream = new ByteArrayInputStream(outputStream.toByteArray());
System.out.println(inputStream.read());
System.out.println(inputStream.read());
System.out.println(inputStream.read());
}
}

16
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpSendingMessageHandlerTests.java
View File

@@ -27,9 +27,9 @@ import java.io.File;
import org.junit.Test;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.sftp.session.SftpSession;
import org.springframework.integration.sftp.session.SftpSessionFactory;
/**
* @author Oleg Zhurakousky
@@ -40,8 +40,8 @@ public class SftpSendingMessageHandlerTests {
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void testHandleFileNameMessage() throws Exception {
SftpSessionFactory sessionFactory = mock(SftpSessionFactory.class);
SftpSession session = mock(SftpSession.class);
SessionFactory sessionFactory = mock(SessionFactory.class);
Session session = mock(Session.class);
when(sessionFactory.getSession()).thenReturn(session);
SftpSendingMessageHandler handler = new SftpSendingMessageHandler(sessionFactory);
handler.setRemoteDirectoryExpression(new SpelExpressionParser().parseExpression("'foo.txt'"));
@@ -53,8 +53,8 @@ public class SftpSendingMessageHandlerTests {
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void testHandleFileAsByte() throws Exception {
SftpSessionFactory sessionFactory = mock(SftpSessionFactory.class);
SftpSession session = mock(SftpSession.class);
SessionFactory sessionFactory = mock(SessionFactory.class);
Session session = mock(Session.class);
when(sessionFactory.getSession()).thenReturn(session);
SftpSendingMessageHandler handler = new SftpSendingMessageHandler(sessionFactory);
handler.setRemoteDirectoryExpression(new SpelExpressionParser().parseExpression("'foo.txt'"));
@@ -66,8 +66,8 @@ public class SftpSendingMessageHandlerTests {
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void testHandleFileMessage() throws Exception {
SftpSessionFactory sessionFactory = mock(SftpSessionFactory.class);
SftpSession session = mock(SftpSession.class);
SessionFactory sessionFactory = mock(SessionFactory.class);
Session session = mock(Session.class);
when(sessionFactory.getSession()).thenReturn(session);
SftpSendingMessageHandler handler = new SftpSendingMessageHandler(sessionFactory);
handler.setRemoteDirectoryExpression(new SpelExpressionParser().parseExpression("'foo.txt'"));