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

INT-1663, INT-1664, INT-1665, INT-1666 refactored Session to encapsulate 'copy' functionality as a whole, fixed FTP stale connection issue, fixed FTP default buffer size, better error reporting, removed remoteWorkingDirectory attribute from session factory, other polishing, adjusted tests

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-12-05 19:24:42 -05:00
parent bcef49aaf2
commit 0f761f3adb
19 changed files with 122 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-file/src/main/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandler.java
View File

@@ -178,7 +178,7 @@ public class FileTransferringMessageHandler extends AbstractMessageHandler {
remoteDirectory += File.separatorChar;
}
String remoteFilePath = remoteDirectory + file.getName();
session.put(fileInputStream, remoteFilePath);
session.copy(fileInputStream, remoteFilePath);
fileInputStream.close();
return true;
}

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

@@ -16,7 +16,9 @@
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.locks.ReentrantLock;
@@ -116,20 +118,20 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean {
}
}
public boolean rm(String path) {
return this.targetSession.rm(path);
public boolean remove(String path) {
return this.targetSession.remove(path);
}
public <F> F[] ls(String path) {
return this.targetSession.<F>ls(path);
public <F> F[] list(String path) {
return this.targetSession.<F>list(path);
}
public InputStream get(String source) {
return this.targetSession.get(source);
public void copy(String source, OutputStream os) throws IOException{
this.targetSession.copy(source, os);
}
public void put(InputStream inputStream, String destination) {
this.targetSession.put(inputStream, destination);
public void copy(InputStream inputStream, String destination) throws IOException{
this.targetSession.copy(inputStream, destination);
}
}

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

@@ -16,7 +16,9 @@
package org.springframework.integration.file.remote.session;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
/**
* Common abstraction for a Session with a remote File system.
@@ -24,18 +26,18 @@ import java.io.InputStream;
* @author Josh Long
* @author Mario Gray
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.0
*/
public interface Session {
boolean rm(String path);
boolean remove(String path);
<F> F[] ls(String path);
<F> F[] list(String path);
void copy(String source, OutputStream outputStream) throws IOException;
InputStream get(String source);
void put(InputStream inputStream, String destination);
void copy(InputStream inputStream, String destination) throws IOException;
void close();
}

10
spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java
View File

@@ -33,7 +33,6 @@ import org.springframework.integration.file.filters.FileListFilter;
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.FileCopyUtils;
import org.springframework.util.ObjectUtils;
/**
@@ -117,7 +116,7 @@ public abstract class AbstractInboundFileSynchronizer<F> implements InboundFileS
try {
session = this.sessionFactory.getSession();
Assert.state(session != null, "failed to acquire a Session");
F[] files = session.<F>ls(this.remoteDirectory);
F[] files = session.<F>list(this.remoteDirectory);
if (!ObjectUtils.isEmpty(files)) {
Collection<F> filteredFiles = this.filterFiles(files);
for (F file : filteredFiles) {
@@ -160,10 +159,7 @@ public abstract class AbstractInboundFileSynchronizer<F> implements InboundFileS
InputStream inputStream = null;
FileOutputStream fileOutputStream = new FileOutputStream(tempFile);
try {
inputStream = session.get(remoteFilePath);
if (inputStream != null) {
FileCopyUtils.copy(inputStream, fileOutputStream);
}
session.copy(remoteFilePath, fileOutputStream);
}
catch (Exception e) {
if (e instanceof RuntimeException){
@@ -189,7 +185,7 @@ public abstract class AbstractInboundFileSynchronizer<F> implements InboundFileS
}
if (tempFile.renameTo(localFile)) {
if (this.deleteRemoteFiles) {
session.rm(remoteFilePath);
session.remove(remoteFilePath);
if (logger.isDebugEnabled()) {
logger.debug("deleted " + remoteFilePath);
}

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

@@ -13,12 +13,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
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;
@@ -26,6 +27,7 @@ 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;
@@ -36,13 +38,13 @@ import org.springframework.util.Assert;
*
* @author Iwein Fuld
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.0
*/
public abstract class AbstractFtpSessionFactory<T extends FTPClient> implements SessionFactory {
public static final String DEFAULT_REMOTE_WORKING_DIRECTORY = "/";
private final Log logger = LogFactory.getLog(this.getClass());
protected FTPClientConfig config;
@@ -54,8 +56,8 @@ public abstract class AbstractFtpSessionFactory<T extends FTPClient> implements
protected String password;
protected int port = FTP.DEFAULT_PORT;
protected String remoteWorkingDirectory = DEFAULT_REMOTE_WORKING_DIRECTORY;
protected int bufferSize = 2048; //see https://issues.apache.org/jira/browse/NET-207
protected int clientMode = FTPClient.ACTIVE_LOCAL_DATA_CONNECTION_MODE;
@@ -80,6 +82,10 @@ public abstract class AbstractFtpSessionFactory<T extends FTPClient> implements
Assert.notNull(config);
this.config = config;
}
public void setBufferSize(int bufferSize) {
this.bufferSize = bufferSize;
}
public void setHost(String host) {
Assert.hasText(host);
@@ -100,12 +106,6 @@ public abstract class AbstractFtpSessionFactory<T extends FTPClient> implements
Assert.notNull(pass, "password should not be null");
this.password = pass;
}
public void setRemoteWorkingDirectory(String remoteWorkingDirectory) {
Assert.notNull(remoteWorkingDirectory, "remote directory should not be null");
this.remoteWorkingDirectory = remoteWorkingDirectory.replaceAll("^$", "/");
}
/**
* ACTIVE_LOCAL_DATA_CONNECTION_MODE = 0 <br>
* A constant indicating the FTP session is expecting all transfers
@@ -139,41 +139,46 @@ public abstract class AbstractFtpSessionFactory<T extends FTPClient> implements
}
}
@SuppressWarnings("unchecked")
private T createClient() throws SocketException, IOException {
T client = this.createClientInstance();
final T client = this.createClientInstance();
Assert.notNull(client, "client must not be null");
client.configure(this.config);
Assert.hasText(this.username, "username is required");
client.connect(this.host);
this.afterConnect(client);
if (!FTPReply.isPositiveCompletion(client.getReplyCode())) {
throw new MessagingException("Connecting to server [" +
this.host + ":" + this.port + "] failed. Please check the connection.");
}
if (logger.isDebugEnabled()) {
logger.debug("Connected to server [" + this.host + ":" + this.port + "]");
}
if (!client.login(username, password)) {
throw new MessagingException(
"Login failed. Please check the username and password.");
}
this.updateClientMode(client);
client.setFileType(this.fileType);
if (logger.isDebugEnabled()) {
logger.debug("login successful");
}
if (!this.remoteWorkingDirectory.equals(client.printWorkingDirectory()) &&
!client.changeWorkingDirectory(this.remoteWorkingDirectory)) {
throw new MessagingException("Could not change directory to '" +
remoteWorkingDirectory + "'. Please check the path.");
}
if (logger.isDebugEnabled()) {
logger.debug("working directory is: " + client.printWorkingDirectory());
}
if (client != null) {
this.postProcessClient(client);
}
return client;
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();
}
/**
@@ -195,18 +200,16 @@ public abstract class AbstractFtpSessionFactory<T extends FTPClient> implements
protected abstract T createClientInstance();
/**
* this is a hook to setup the state of the {@link org.apache.commons.net.ftp.FTPClient} impl *after* the
* implementation's {@link org.apache.commons.net.ftp.FTPClient#connect(String)} method's been called but before any
* action's been taken.
*
* @param t the ftp client instance on which to act
* @throws IOException if anything should go wrong
* Will handle additional initialization after client.connect() method was invoked,
* but before any action on the client has been taken
*/
protected void afterConnect(T t) throws IOException {
protected void postProcessClientAfterConnect(T t) throws IOException {
// NOOP
}
protected void postProcessClient(T t) throws IOException {
/**
* Will handle additional initialization before client.connect() method was invoked.
*/
protected void postProcessClientBeforeConnect(T client) throws IOException {
// NOOP
}

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

@@ -123,13 +123,13 @@ public class DefaultFtpsSessionFactory extends AbstractFtpSessionFactory<FTPSCli
}
@Override
protected void afterConnect(FTPSClient ftpsClient) throws IOException {
protected void postProcessClientAfterConnect(FTPSClient ftpsClient) throws IOException {
ftpsClient.execPBSZ(0);
ftpsClient.execPROT(this.prot);
}
@Override
protected void postProcessClient(FTPSClient ftpsClient) throws IOException {
protected void postProcessClientBeforeConnect(FTPSClient ftpsClient) throws IOException {
if (StringUtils.hasText(this.authValue)) {
ftpsClient.setAuthValue(authValue);
}

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

@@ -18,6 +18,7 @@ package org.springframework.integration.ftp.session;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -31,6 +32,7 @@ import org.springframework.util.Assert;
* Implementation of {@link Session} for FTP.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.0
*/
class FtpSession implements Session {
@@ -46,21 +48,26 @@ class FtpSession implements Session {
}
public boolean rm(String path) {
public boolean remove(String path) {
Assert.hasText(path, "path must not be null");
boolean completed = false;
try {
this.client.deleteFile(path);
return true;
completed = this.client.deleteFile(path);
if (!completed){
throw new IOException("Failed to delete '" + path + "'. Server replied with: " + client.getReplyString());
}
}
catch (IOException e) {
if (logger.isWarnEnabled()) {
logger.warn("failed to delete file", e);
}
return false;
}
return completed;
}
@SuppressWarnings({"unchecked"})
public FTPFile[] ls(String path) {
public FTPFile[] list(String path) {
Assert.hasText(path, "path must not be null");
try {
return this.client.listFiles(path);
}
@@ -72,29 +79,24 @@ class FtpSession implements Session {
}
}
public InputStream get(String path) {
try {
InputStream inputStream = this.client.retrieveFileStream(path);
this.client.completePendingCommand();
return inputStream;
}
catch (IOException e) {
if (logger.isWarnEnabled()) {
logger.warn("failed to retrieve file", e);
}
return null;
public void copy(String path, OutputStream fos) throws IOException{
Assert.hasText(path, "path must not be null");
Assert.notNull(fos, "outputStream must not be null");
boolean completed = this.client.retrieveFile(path, fos);
if (!completed){
throw new IOException("Failed to copy '" + path + "'. Server replied with: " + client.getReplyString());
}
logger.info("File have been successfully transfered to: " + path);
}
public void put(InputStream inputStream, String path) {
public void copy(InputStream inputStream, String path) throws IOException{
Assert.notNull(inputStream, "inputStream must not be null");
Assert.notNull(path, "path must not be null");
try {
this.client.storeFile(path, inputStream);
}
catch (IOException e) {
throw new IllegalStateException("failed to copy file", e);
Assert.hasText(path, "path must not be null");
boolean completed = client.storeFile(path, inputStream);
if (!completed){
throw new IOException("Failed to copy '" + path + "'. Server replied with: " + client.getReplyString());
}
logger.info("File have been successfully transfered to: " + path);
}
public void close() {
@@ -107,5 +109,4 @@ class FtpSession implements Session {
}
}
}
}

1
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests-context.xml
View File

@@ -17,7 +17,6 @@
<property name="password" value="password"/>
<property name="clientMode" value="2"/>
<property name="fileType" value="2"/>
<property name="remoteWorkingDirectory" value="foo/bar"/>
</bean>
<ftp:inbound-channel-adapter id="adapterFtp"

1
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests-fail-context.xml
View File

@@ -17,7 +17,6 @@
<property name="password" value="password"/>
<property name="clientMode" value="0"/>
<property name="fileType" value="2"/>
<property name="remoteWorkingDirectory" value="foo/bar"/>
</bean>
<ftp:inbound-channel-adapter id="adapterFtpDontAutoCreate"

1
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserOutboundTests-context.xml
View File

@@ -16,7 +16,6 @@
<property name="password" value="password"/>
<property name="clientMode" value="0"/>
<property name="fileType" value="2"/>
<property name="remoteWorkingDirectory" value="temp"/>
</bean>
<int-ftp:outbound-channel-adapter id="ftpOutboundAdapter"

3
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterSample-context.xml
View File

@@ -9,9 +9,8 @@
<bean id="ftpClientFactory" class="org.springframework.integration.ftp.session.DefaultFtpSessionFactory">
<property name="host" value="localhost"/>
<property name="username" value="ozhurakousky"/>
<property name="username" value="oleg"/>
<property name="password" value="xxxx"/>
<property name="remoteWorkingDirectory" value="/Users/ozhurakousky/workspace-sts-2.3.3.M2/si/spring-integration/spring-integration-ftp/remote-test-dir"/>
</bean>
<int-ftp:inbound-channel-adapter id="ftpInbound"

1
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundChannelAdapterParserTests-context.xml
View File

@@ -14,7 +14,6 @@
<property name="password" value="password"/>
<property name="clientMode" value="0"/>
<property name="fileType" value="2"/>
<property name="remoteWorkingDirectory" value="foo/bar"/>
</bean>
<int-ftp:outbound-channel-adapter id="ftpOutbound"

8
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundChannelAdapterSample-context.xml
View File

@@ -9,9 +9,8 @@
<bean id="ftpSessionFactory" class="org.springframework.integration.ftp.session.DefaultFtpSessionFactory">
<property name="host" value="localhost"/>
<property name="username" value="ozhurakousky"/>
<property name="username" value="oleg"/>
<property name="password" value="xxxx"/>
<property name="remoteWorkingDirectory" value="/Users/ozhurakousky/workspace-sts-2.3.3.M2/si/spring-integration/spring-integration-ftp/remote-target-dir"/>
</bean>
<int:channel id="ftpChannel"/>
@@ -19,8 +18,5 @@
<int-ftp:outbound-channel-adapter
session-factory="ftpSessionFactory"
remote-directory="/Users/ozhurakousky/workspace-sts-2.3.3.M2/si/spring-integration/spring-integration-ftp/remote-target-dir"
channel="ftpChannel"/>
channel="ftpChannel"/>
</beans>

1
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpsInboundChannelAdapterParserTests-context.xml
View File

@@ -14,7 +14,6 @@
<property name="password" value="password"/>
<property name="clientMode" value="0"/>
<property name="fileType" value="2"/>
<property name="remoteWorkingDirectory" value="foo/bar"/>
</bean>
<int-ftp:inbound-channel-adapter id="ftpInbound"

1
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpsOutboundChannelAdapterParserTests-context.xml
View File

@@ -21,7 +21,6 @@
<property name="password" value="password"/>
<property name="clientMode" value="0"/>
<property name="fileType" value="2"/>
<property name="remoteWorkingDirectory" value="foo/bar"/>
<property name="useClientMode" value="true"/>
<property name="cipherSuites" value="a,b.c"/>
<property name="keyManager" ref="keyManager"/>

1
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/ftp-message-history-context.xml
View File

@@ -18,7 +18,6 @@
<property name="password" value="password"/>
<property name="clientMode" value="2"/>
<property name="fileType" value="2"/>
<property name="remoteWorkingDirectory" value="foo/bar"/>
</bean>
<ftp:inbound-channel-adapter id="adapterFtp"

9
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTest.java
View File

@@ -28,7 +28,7 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.File;
import java.io.FileInputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collection;
@@ -71,8 +71,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTest {
ftpSessionFactory.setUsername("kermit");
ftpSessionFactory.setPassword("frog");
ftpSessionFactory.setHost("foo.com");
ftpSessionFactory.setRemoteWorkingDirectory("remote-test-dir");
FtpInboundFileSynchronizer synchronizer = spy(new FtpInboundFileSynchronizer(ftpSessionFactory));
synchronizer.setDeleteRemoteFiles(true);
synchronizer.setRemoteDirectory("remote-test-dir");
@@ -117,10 +116,10 @@ public class FtpInboundRemoteFileSystemSynchronizerTest {
file.setName(fileName);
file.setType(FTPFile.FILE_TYPE);
ftpFiles.add(file);
when(ftpClient.retrieveFileStream(fileName)).thenReturn(new FileInputStream("remote-test-dir/" + fileName));
when(ftpClient.retrieveFile(Mockito.eq("remote-test-dir/" + fileName) , Mockito.any(OutputStream.class))).thenReturn(true);
}
when(ftpClient.listFiles("remote-test-dir")).thenReturn(ftpFiles.toArray(new FTPFile[]{}));
when(ftpClient.deleteFile(Mockito.anyString())).thenReturn(true);
return ftpClient;
} catch (Exception e) {
throw new RuntimeException("Failed to create mock client", e);

2
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpSendingMessageHandlerTest.java
View File

@@ -55,7 +55,7 @@ public class FtpSendingMessageHandlerTest {
sessionFactory.setUsername("kermit");
sessionFactory.setPassword("frog");
sessionFactory.setHost("foo.com");
sessionFactory.setRemoteWorkingDirectory("remote-test-dir");
//sessionFactory.setRemoteWorkingDirectory("remote-test-dir");
}
@SuppressWarnings({ "unchecked", "rawtypes" })

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

@@ -16,7 +16,9 @@
package org.springframework.integration.sftp.session;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Vector;
import org.apache.commons.logging.Log;
@@ -24,6 +26,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.util.Assert;
import org.springframework.util.FileCopyUtils;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.ChannelSftp.LsEntry;
@@ -53,7 +56,7 @@ class SftpSession implements Session {
}
public boolean rm(String path) {
public boolean remove(String path) {
Assert.state(this.channel != null, "session is not connected");
try {
this.channel.rm(path);
@@ -68,7 +71,7 @@ class SftpSession implements Session {
}
@SuppressWarnings("unchecked")
public LsEntry[] ls(String path) {
public LsEntry[] list(String path) {
Assert.state(this.channel != null, "session is not connected");
try {
Vector<?> lsEntries = this.channel.ls(path);
@@ -90,28 +93,25 @@ class SftpSession implements Session {
return new LsEntry[0];
}
public InputStream get(String source) {
public void copy(String source, OutputStream os) throws IOException{
Assert.state(this.channel != null, "session is not connected");
try {
return this.channel.get(source);
InputStream is = this.channel.get(source);
FileCopyUtils.copy(is, os);
}
catch (SftpException e) {
if (logger.isWarnEnabled()) {
logger.warn("failed to retrieve file", e);
}
return null;
throw new IOException("failed to copy file", e);
}
}
public void put(InputStream inputStream, String destination) {
public void copy(InputStream inputStream, String destination) throws IOException{
Assert.state(this.channel != null, "session is not connected");
try {
this.channel.put(inputStream, destination);
}
catch (SftpException e) {
if (logger.isWarnEnabled()) {
logger.warn("failed to copy file", e);
}
throw new IOException("failed to copy file", e);
}
}