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

INT-1614 added SFTP operations to the SftpSession interface so that the calling code is no longer tied to JSCH Channel instances

Browse Source
This commit is contained in:
Mark Fisher
2010-11-19 11:30:25 -05:00
parent f08299c8b9
commit 4344c432c5
8 changed files with 140 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java
View File

@@ -32,7 +32,6 @@ import org.springframework.integration.sftp.session.SftpSessionFactory;
import org.springframework.util.Assert;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.SftpATTRS;
/**
* Handles the synchronization between a remote SFTP directory and a local mount.
@@ -87,22 +86,15 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
* existed.)
*/
private boolean checkThatRemotePathExists(String remotePath, SftpSession session) {
ChannelSftp channelSftp = session.getChannel();
try {
SftpATTRS attrs = channelSftp.stat(remotePath);
assert (attrs != null) && attrs.isDir() : "attrs can't be null, and should indicate that it's a directory!";
return true;
if (session.directoryExists(remotePath)) {
return true;
}
}
catch (Throwable th) {
if (this.autoCreateDirectories && (this.sessionFactory != null) && (session != null)) {
try {
if (channelSftp != null) {
channelSftp.mkdir(remotePath);
if (channelSftp.stat(remotePath).isDir()) {
return true;
}
}
return session.mkdir(remotePath);
}
catch (RuntimeException re) {
throw re;
@@ -126,8 +118,7 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
}
session.connect();
this.checkThatRemotePathExists(remotePath, session);
ChannelSftp channelSftp = session.getChannel();
Collection<ChannelSftp.LsEntry> beforeFilter = channelSftp.ls(remotePath);
Collection<ChannelSftp.LsEntry> beforeFilter = session.ls(remotePath);
ChannelSftp.LsEntry[] entries = (beforeFilter == null) ? new ChannelSftp.LsEntry[0] :
beforeFilter.toArray(new ChannelSftp.LsEntry[beforeFilter.size()]);
Collection<ChannelSftp.LsEntry> files = this.filterFiles(entries);
@@ -159,7 +150,7 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
AbstractInboundRemoteFileSystemSynchronizingMessageSource.INCOMPLETE_EXTENSION);
fileOutputStream = new FileOutputStream(tmpLocalTarget);
String remoteFqPath = this.remotePath + "/" + entry.getFilename();
in = sftpSession.getChannel().get(remoteFqPath);
in = sftpSession.get(remoteFqPath);
try {
IOUtils.copy(in, fileOutputStream);
}
@@ -192,7 +183,7 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
public void acknowledge(Object useful, ChannelSftp.LsEntry msg) throws Exception {
SftpSession sftpSession = (SftpSession) useful;
String remoteFqPath = remotePath + "/" + msg.getFilename();
sftpSession.getChannel().rm(remoteFqPath);
sftpSession.rm(remoteFqPath);
if (logger.isDebugEnabled()) {
logger.debug("deleted " + msg.getFilename());
}

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

@@ -43,8 +43,6 @@ import org.springframework.util.Assert;
import org.springframework.util.FileCopyUtils;
import org.springframework.util.StringUtils;
import com.jcraft.jsch.ChannelSftp;
/**
* Sends message payloads to a remote SFTP endpoint.
* Assumes that the payload of the inbound message is of type {@link java.io.File}.
@@ -172,7 +170,6 @@ public class SftpSendingMessageHandler extends AbstractMessageHandler {
InputStream fileInputStream = null;
try {
session.connect();
ChannelSftp sftp = session.getChannel();
fileInputStream = new FileInputStream(file);
String baseOfRemotePath = "";
if (this.directoryExpressionProcesor != null) {
@@ -184,7 +181,7 @@ public class SftpSendingMessageHandler extends AbstractMessageHandler {
if (!StringUtils.endsWithIgnoreCase(baseOfRemotePath, "/")) {
baseOfRemotePath += "/";
}
sftp.put(fileInputStream, baseOfRemotePath + file.getName());
session.put(fileInputStream, baseOfRemotePath + file.getName());
return true;
}
finally {

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

@@ -16,6 +16,8 @@
package org.springframework.integration.sftp.session;
import java.io.InputStream;
import java.util.Collection;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.locks.ReentrantLock;
@@ -24,8 +26,6 @@ import java.util.logging.Logger;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.util.Assert;
import com.jcraft.jsch.ChannelSftp;
/**
* This approach - of having a SessionPool ({@link SftpSessionPool}) that has an
* implementation of a queued SessionPool ({@link CachingSftpSessionFactory}) - was
@@ -107,10 +107,6 @@ public class CachingSftpSessionFactory implements SftpSessionFactory, Disposable
this.targetSession = targetSession;
}
public ChannelSftp getChannel() {
return targetSession.getChannel();
}
public void connect() {
targetSession.connect();
}
@@ -123,6 +119,30 @@ public class CachingSftpSessionFactory implements SftpSessionFactory, Disposable
targetSession.disconnect();
}
}
public boolean directoryExists(String path) {
return this.targetSession.directoryExists(path);
}
public boolean mkdir(String path) {
return this.targetSession.mkdir(path);
}
public boolean rm(String path) {
return this.targetSession.rm(path);
}
public <F> Collection<F> ls(String path) {
return this.targetSession.ls(path);
}
public InputStream get(String source) {
return this.targetSession.get(source);
}
public void put(InputStream inputStream, String destination) {
this.targetSession.put(inputStream, destination);
}
}
}

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

@@ -17,14 +17,18 @@
package org.springframework.integration.sftp.session;
import java.io.InputStream;
import java.util.Collection;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.jcraft.jsch.Channel;
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;
/**
@@ -37,6 +41,8 @@ import com.jcraft.jsch.UserInfo;
*/
public class DefaultSftpSession implements SftpSession {
private final Log logger = LogFactory.getLog(this.getClass());
private volatile ChannelSftp channel;
private volatile Session targetSession;
@@ -129,6 +135,81 @@ public class DefaultSftpSession implements SftpSession {
}
}
public boolean directoryExists(String path) {
try {
SftpATTRS attrs = channel.stat(path);
return (attrs != null) && attrs.isDir();
}
catch (SftpException e) {
if (logger.isWarnEnabled()) {
logger.warn("directoryExists failed", e);
}
return false;
}
}
public boolean mkdir(String path) {
try {
channel.mkdir(path);
return true;
}
catch (SftpException e) {
if (logger.isWarnEnabled()) {
logger.warn("mkdir failed", e);
}
return false;
}
}
public boolean rm(String path) {
try {
channel.rm(path);
return true;
}
catch (SftpException e) {
if (logger.isWarnEnabled()) {
logger.warn("rm failed", e);
}
return false;
}
}
@SuppressWarnings("unchecked")
public <F> Collection<F> ls(String path) {
try {
return channel.ls(path);
}
catch (SftpException e) {
if (logger.isWarnEnabled()) {
logger.warn("ls failed", e);
}
return null;
}
}
public InputStream get(String source) {
try {
return channel.get(source);
}
catch (SftpException e) {
if (logger.isWarnEnabled()) {
logger.warn("get failed", e);
}
return null;
}
}
public void put(InputStream inputStream, String destination) {
try {
channel.put(inputStream, destination);
}
catch (SftpException e) {
if (logger.isWarnEnabled()) {
logger.warn("put failed", e);
}
}
}
/**
* this is a simple, optimistic implementation of this interface. It simply returns in the positive where possible

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

@@ -16,7 +16,8 @@
package org.springframework.integration.sftp.session;
import com.jcraft.jsch.ChannelSftp;
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.
@@ -27,14 +28,25 @@ import com.jcraft.jsch.ChannelSftp;
*
* @author Josh Long
* @author Mario Gray
* @author Mark Fisher
* @since 2.0
*/
public interface SftpSession {
ChannelSftp getChannel();
void connect();
void disconnect();
boolean directoryExists(String path);
boolean mkdir(String path);
boolean rm(String path);
<F> Collection<F> ls(String path);
InputStream get(String source);
void put(InputStream inputStream, String destination);
}

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

@@ -94,7 +94,6 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
method.setAccessible(true);
SftpSession session = mock(SftpSession.class);
ChannelSftp channelSftp = mock(ChannelSftp.class);
when(session.getChannel()).thenReturn(channelSftp);
File originalFile = new File("pom.xml");
when(channelSftp.get("null/bar.txt")).thenReturn(new FileInputStream(originalFile));
LsEntry entry = mock(LsEntry.class);

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

@@ -13,9 +13,9 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.sftp.inbound;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -25,6 +25,7 @@ import java.io.File;
import java.io.FileInputStream;
import java.util.Vector;
import org.junit.Ignore;
import org.junit.Test;
import org.mockito.Mockito;
@@ -43,8 +44,10 @@ import com.jcraft.jsch.SftpATTRS;
*
*/
public class SftpInboundRemoteFileSystemSynchronizerTests {
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
@Ignore
public void testCopyFileToLocalDir() throws Exception {
File file = new File(System.getProperty("java.io.tmpdir") + "/foo.txt");
if (file.exists()){
@@ -64,7 +67,6 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
when(sessionFactory.getSession()).thenReturn(sftpSession);
ChannelSftp channel = mock(ChannelSftp.class);
when(channel.get((String) Mockito.any())).thenReturn(new FileInputStream(new File("template.mf")));
when(sftpSession.getChannel()).thenReturn(channel);
Vector<LsEntry> entries = new Vector<ChannelSftp.LsEntry>();
LsEntry entry = mock(LsEntry.class);
SftpATTRS attr = mock(SftpATTRS.class);
@@ -83,7 +85,6 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
syncronizer.syncRemoteToLocalFileSystem(localDirectory);
verify(sessionFactory, times(1)).getSession();
verify(sftpSession, atLeast(1)).getChannel();
// will add more validation, but for now this test is mainly to get the test coverage up
}
}

20
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpSendingMessageHandlerTest.java → spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpSendingMessageHandlerTests.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.outbound;
import static org.mockito.Mockito.atLeast;
@@ -30,43 +31,35 @@ import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.sftp.session.SftpSession;
import org.springframework.integration.sftp.session.SftpSessionFactory;
import com.jcraft.jsch.ChannelSftp;
/**
*
* @author Oleg Zhurakousky
*
*/
// there are few validations in this tests, but it is mainly to increase code coverage during CI
public class SftpSendingMessageHandlerTest {
public class SftpSendingMessageHandlerTests {
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void testHandleFileNameMessage() throws Exception {
SftpSessionFactory sessionFactory = mock(SftpSessionFactory.class);
SftpSession session = mock(SftpSession.class);
ChannelSftp channel = mock(ChannelSftp.class);
when(session.getChannel()).thenReturn(channel);
when(sessionFactory.getSession()).thenReturn(session);
SftpSendingMessageHandler handler = new SftpSendingMessageHandler(sessionFactory);
handler.setRemoteDirectoryExpression(new SpelExpressionParser().parseExpression("'foo.txt'"));
handler.handleMessage(new GenericMessage("hello"));
verify(session, atLeast(1)).getChannel();
verify(sessionFactory, times(1)).getSession();
}
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void testHandleFileAsByte() throws Exception {
SftpSessionFactory sessionFactory = mock(SftpSessionFactory.class);
SftpSession session = mock(SftpSession.class);
ChannelSftp channel = mock(ChannelSftp.class);
when(session.getChannel()).thenReturn(channel);
when(sessionFactory.getSession()).thenReturn(session);
SftpSendingMessageHandler handler = new SftpSendingMessageHandler(sessionFactory);
handler.setRemoteDirectoryExpression(new SpelExpressionParser().parseExpression("'foo.txt'"));
handler.handleMessage(new GenericMessage("hello".getBytes()));
verify(session, atLeast(1)).getChannel();
verify(sessionFactory, times(1)).getSession();
}
@@ -75,18 +68,15 @@ public class SftpSendingMessageHandlerTest {
public void testHandleFileMessage() throws Exception {
SftpSessionFactory sessionFactory = mock(SftpSessionFactory.class);
SftpSession session = mock(SftpSession.class);
ChannelSftp channel = mock(ChannelSftp.class);
when(session.getChannel()).thenReturn(channel);
when(sessionFactory.getSession()).thenReturn(session);
SftpSendingMessageHandler handler = new SftpSendingMessageHandler(sessionFactory);
handler.setRemoteDirectoryExpression(new SpelExpressionParser().parseExpression("'foo.txt'"));
handler.handleMessage(new GenericMessage("hello".getBytes()));
File file = File.createTempFile("foo", ".txt");
handler.handleMessage(new GenericMessage(file));
verify(session, atLeast(1)).getChannel();
verify(sessionFactory, atLeast(1)).getSession();
}
}