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

INT-1631 polishing, moved checkIfRemoteDirExist to synchronier other minor cleanup

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-11-17 23:53:42 -05:00
parent d4ebf43833
commit 9f1d9e87ce
6 changed files with 61 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -69,7 +69,6 @@ public class SftpInboundChannelAdapterParser extends AbstractPollingInboundChann
BeanDefinitionBuilder messageSourceBuilder =
BeanDefinitionBuilder.rootBeanDefinition("org.springframework.integration.sftp.inbound.SftpInboundSynchronizingMessageSource");
messageSourceBuilder.addConstructorArgReference(sessionPollName);
messageSourceBuilder.addPropertyValue("synchronizer", synchronizerBuilder.getBeanDefinition());
if (hasFileNamePattern){

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

@@ -33,6 +33,7 @@ import org.springframework.scheduling.Trigger;
import org.springframework.util.Assert;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.SftpATTRS;
/**
* Gandles the synchronization between a remote SFTP endpoint and a local mount.
@@ -47,6 +48,8 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
* the path on the remote mount
*/
private volatile String remotePath;
private volatile boolean autoCreateDirectories;
/**
* the pool of {@link org.springframework.integration.sftp.session.SftpSessionPool} SFTP sessions
@@ -58,6 +61,9 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
this.sessionPool = sessionPool;
}
public void setAutoCreateDirectories(boolean autoCreateDirectories) {
this.autoCreateDirectories = autoCreateDirectories;
}
public void setRemotePath(String remotePath) {
this.remotePath = remotePath;
@@ -66,7 +72,6 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
@Override
protected Trigger getTrigger() {
throw new UnsupportedOperationException("This method curently is not implemented");
//return new PeriodicTrigger(10 * 1000);
}
@Override
@@ -76,6 +81,48 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
this.entryAcknowledgmentStrategy = new DeletionEntryAcknowledgmentStrategy();
}
}
/**
* This method will check to ensure that the remote directory exists. If the directory
* doesnt exist, and autoCreatePath is 'true,' then this method makes a few reasonably sane attempts
* to create it. Otherwise, it fails fast.
*
* @param remotePath the path on the remote SSH / SFTP server to create.
* @return whether or not the directory is there (regardless of whether we created it in this method or it already
* 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;
}
catch (Throwable th) {
if (this.autoCreateDirectories && (this.sessionPool != null) && (session != null)) {
try {
if (channelSftp != null) {
channelSftp.mkdir(remotePath);
if (channelSftp.stat(remotePath).isDir()) {
return true;
}
}
}
catch (RuntimeException re) {
throw re;
}
catch (Exception e){
throw new MessagingException("Failed to auto-create remote directory", e);
}
}
}
return false;
}
private boolean copyFromRemoteToLocalDirectory(SftpSession sftpSession, ChannelSftp.LsEntry entry, Resource localDir) throws Exception {
File fileForLocalDir = localDir.getFile();
@@ -116,7 +163,9 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
SftpSession session = null;
try {
session = sessionPool.getSession();
logger.trace("Pooled SftpSession " + this.sessionPool + " from the pool");
session.start();
this.checkThatRemotePathExists(remotePath, session);
ChannelSftp channelSftp = session.getChannel();
Collection<ChannelSftp.LsEntry> beforeFilter = channelSftp.ls(remotePath);
ChannelSftp.LsEntry[] entries = (beforeFilter == null) ? new ChannelSftp.LsEntry[0] :
@@ -133,6 +182,7 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
}
finally {
this.sessionPool.release(session);
logger.trace("Putting SftpSession " + this.sessionPool + " back into the pool");
}
}

82
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizingMessageSource.java
View File

@@ -24,11 +24,8 @@ import org.springframework.integration.MessagingException;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.file.synchronization.AbstractInboundRemoteFileSystemSynchronizingMessageSource;
import org.springframework.integration.sftp.filters.SftpPatternMatchingFileListFilter;
import org.springframework.integration.sftp.session.SftpSession;
import org.springframework.integration.sftp.session.SftpSessionPool;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.SftpATTRS;
/**
@@ -40,31 +37,13 @@ import com.jcraft.jsch.SftpATTRS;
*/
public class SftpInboundSynchronizingMessageSource extends
AbstractInboundRemoteFileSystemSynchronizingMessageSource<ChannelSftp.LsEntry, SftpInboundSynchronizer> {
/**
* the pool of sessions
*/
private final SftpSessionPool sessionPool;
/**
* the remote path on the server
*/
private volatile String remoteDirectory;
private volatile Pattern filenamePattern;
public SftpInboundSynchronizingMessageSource(SftpSessionPool sessionPool){
this.sessionPool = sessionPool;
}
public void setFilenamePattern(Pattern filenamePattern) {
this.filenamePattern = filenamePattern;
}
public void setRemoteDirectory(String remoteDirectory) {
this.remoteDirectory = remoteDirectory;
}
public String getComponentType(){
return "sftp:inbound-channel-adapter";
}
@@ -77,69 +56,12 @@ public class SftpInboundSynchronizingMessageSource extends
*/
Message<File> message = this.fileSource.receive();
if (message == null){
this.checkThatRemotePathExists(this.remoteDirectory);
this.synchronizer.syncRemoteToLocalFileSystem();
message = this.fileSource.receive();
}
return message;
}
/**
* This method will check to ensure that the remote directory exists. If the directory
* doesnt exist, and autoCreatePath is 'true,' then this method makes a few reasonably sane attempts
* to create it. Otherwise, it fails fast.
*
* @param remotePath the path on the remote SSH / SFTP server to create.
* @return whether or not the directory is there (regardless of whether we created it in this method or it already
* existed.)
*/
private boolean checkThatRemotePathExists(String remotePath) {
SftpSession session = null;
ChannelSftp channelSftp = null;
try {
session = this.sessionPool.getSession();
session.start();
channelSftp = session.getChannel();
}
catch (RuntimeException re) {
throw re;
}
catch (Exception e){
throw new MessagingException("Failed to get SftpSession while checking for existance of the remote directory", e);
}
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;
}
catch (Throwable th) {
if (this.autoCreateDirectories && (this.sessionPool != null) && (session != null)) {
try {
if (channelSftp != null) {
channelSftp.mkdir(remotePath);
if (channelSftp.stat(remotePath).isDir()) {
return true;
}
}
}
catch (RuntimeException re) {
throw re;
}
catch (Exception e){
throw new MessagingException("Failed to auto-create remote directory", e);
}
}
}
finally {
this.sessionPool.release(session);
}
return false;
}
@Override
protected void onInit() {
try {
@@ -172,7 +94,9 @@ public class SftpInboundSynchronizingMessageSource extends
if (filenamePattern != null) {
SftpPatternMatchingFileListFilter sftpFilePatternMatchingEntryListFilter =
new SftpPatternMatchingFileListFilter(filenamePattern);
//TODO refactor the design so values don't need to be duplicated
this.synchronizer.setFilter(sftpFilePatternMatchingEntryListFilter);
this.synchronizer.setAutoCreateDirectories(this.autoCreateDirectories);
}
}
}

3
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/QueuedSftpSessionPool.java
View File

@@ -74,9 +74,6 @@ public class QueuedSftpSessionPool implements SftpSessionPool, SmartLifecycle {
SftpSession session = this.queue.poll();
if (null == session) {
session = this.sftpSessionFactory.getSession();
if (this.queue.size() < this.maxPoolSize) {
this.queue.add(session);
}
}
return session;
}

6
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpInboundReceiveSample-ignored-context.xml
View File

@@ -3,8 +3,10 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-sftp="http://www.springframework.org/schema/integration/sftp"
xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd
http://www.springframework.org/schema/integration/sftp http://www.springframework.org/schema/integration/sftp/spring-integration-sftp-2.0.xsd">
@@ -29,7 +31,7 @@
auto-create-directories="false"
auto-delete-remote-files-on-sync="false"
filename-pattern=".*\.txt$">
<int:poller fixed-rate="10000" max-messages-per-poll="10"/>
<int:poller fixed-rate="1000" max-messages-per-poll="10" task-executor="executor"/>
</int-sftp:inbound-channel-adapter>
@@ -40,5 +42,7 @@
<bean id="filter" class="org.springframework.integration.sftp.filters.SftpPatternMatchingFileListFilter">
<constructor-arg value=".*\.txt$"/>
</bean>
<task:executor id="executor" pool-size="15"/>
</beans>

16
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpInboundReceiveSample.java
View File

@@ -29,22 +29,10 @@ public class SftpInboundReceiveSample {
@Test
@Ignore
public void testInbound() throws Exception{
ClassPathXmlApplicationContext ac =
new ClassPathXmlApplicationContext("SftpInboundReceiveSample-ignored.xml", SftpInboundReceiveSample.class);
new ClassPathXmlApplicationContext("SftpInboundReceiveSample-ignored-context.xml", SftpInboundReceiveSample.class);
System.out.println("Done");
System.in.read();
ac.stop();
// File file = new File("/Users/ozhurakousky/workspace-sts-2.3.3.M2/si/spring-integration/spring-integration-sftp/foo.txt");
// if (file.exists()){
// Message<File> message = MessageBuilder.withPayload(file).build();
// MessageChannel inputChannel = ac.getBean("inputChannel", MessageChannel.class);
// inputChannel.send(message);
// Thread.sleep(2000);
// }
// System.out.println("Done");
// ac.stop();
Thread.sleep(3000 * 300);
}
}