Watch multiple Remote SFTP directories

[ulmermark]

I am looking for solution/example that points me to how to watch multiple SFTP remote directories....

currently using a Spring Boot application with an @Configuration class that watches a single directory as such:

   @Bean
    public SftpInboundFileSynchronizer sftpInboundFileSynchronizer() throws Exception {
        SftpInboundFileSynchronizer fileSynchronizer = new SftpInboundFileSynchronizer(sftpSessionFactory());
        fileSynchronizer.setDeleteRemoteFiles(false);
        fileSynchronizer.setRemoteDirectory(this.serviceProps.getSftp().getRead().getRemoteDirectory());
        fileSynchronizer.setFilter(new SftpSimplePatternFileListFilter(this.serviceProps.getSftp().getRead().getRemotePattern()));
        return fileSynchronizer;
    }
    
    /**
     * Inbound Message Handler for files retrieved from the SFTP server
     * @return
     * @throws Exception
     */
    @Bean
    @InboundChannelAdapter(channel = "spn.vendor.gateway.sftp.inbound.channel")
    public MessageSource<File> sftpMessageSource() throws Exception {
        SftpInboundFileSynchronizingMessageSource source =
                new SftpInboundFileSynchronizingMessageSource(sftpInboundFileSynchronizer());
        source.setLocalDirectory(new File(this.serviceProps.getSftp().getRead().getLocalDirectory()));
        source.setAutoCreateLocalDirectory(true);
        
        FileSystemPersistentAcceptOnceFileListFilter localFilter = new FileSystemPersistentAcceptOnceFileListFilter(new RedisMetadataStore(this.redis), "spn-vendor-gateway-svc");
        
        Collection<FileListFilter<File>> filters = new ArrayList<>();
		filters.add(localFilter);
        CompositeFileListFilter<File> composite = new CompositeFileListFilter<File>(filters);
        
        source.setLocalFilter(composite); // TODO - need to use persistent a distributed Redis AcceptOnceFileFilter...
        return source;
    }

I do not see a way to instantiate multiple SftpInboundFileSynchronizer and MessageSource<File> instances all tied to the same input channel using the @Bean annotations.

Thanks for any help that can be given on this.

Mark

[artembilan]

Well, since SftpInboundFileSynchronizer doesn't support several directories to scan you don't have choice unless configure several SftpInboundFileSynchronizer @Beans for all remote dirs and the same number of SftpInboundFileSynchronizingMessageSource for each those SftpInboundFileSynchronizer.

And yes, you are correct: you have to declare all @InboundChannelAdapter to the same spn.vendor.gateway.sftp.inbound.channel. And subscriber(s) on that channel will receive messages based on all your remote directory.

So, your problem isn't clear.

Plus be advised that this kind of questions we support only from the [StackOverflow|http://stackoverflow.com/questions/tagged/spring-integration].

There is a trick in the recent Java DSL like IntegrationFlowContext. With that you can declare IntegrationFlows manually at runtime. Exactly what fits your requirements.
More info is here: https://spring.io/blog/2016/10/14/java-dsl-for-spring-integration-1-2-release-is-available

[garyrussell]

You can also use a Smart Poller with a custom message source advice to change the remote directory - e.g. when a poll returns null, change the directory so the next poll will check the next directory; when you've cycled through all directories, start again at the beginning.

[ulmermark]

I took the approach of configuring multiple SftpInboundFileSynchronizer and all appears to be working as long as I am inside the Evlipse IDE debugging the app

As soon as I package the Spring Boot app as an Uber jar. I am running into threading issues with Connectivity to the SFTP server

The app uses a single poller and a Caching session factory per the Spring Integration SFTP adapter docs

In Wxlipse I see multiple threads downloading the files from the remote directories

Is the CachingSessionFactory thread safe?

[artembilan]

Glad to see that we have a progress!

Regarding your threading issue, it would be better to some description how it is behaving . Plus some stack trace and logs would be grate help for us to determine the issue.

The global Poller shares 10 threads for tasks by default. So, your adapters doesn't work in single thread.

Yes, CachingSessionFactory is designed to be thread-safe. Therefore we want to know more context about your app.

Maybe even you can come up with some simple Boot application to reproduce an issue from our side...

[ulmermark]

I have attached logs from the 2 different configurations...with a
single SftpInboundFileSynchronizingMessageSource
and SftpInboundFileSynchronizer and one with 2 as well as a stripped down
application that exhibits the behavior I am experiencing.

The application.yml will drive the number of "SFTP" inbound synchronizers
that are configured....just configure 1 or 2 clients in the
application.yml....

clients:
- name: SPP
id: "01791100"
read:
remoteDirectory: SPP/downloads
remotePattern: ""
localDirectory: Downloads/SPP
fileType: spp
write:
remoteDirectory: SPP/uploads
- name: TEST
id: "01791100"
read:
remoteDirectory: TEST/downloads
remotePattern: ""
localDirectory: Downloads/TEST
fileType: test
write:
remoteDirectory: TEST/uploads

Will need to be running a local SFTP server to test or point to a remote
one as well...credentials go in the application.yml

On Sun, Nov 20, 2016 at 8:16 PM, Artem Bilan notifications@github.com
wrote:

Glad to see that we have a progress!

Regarding your threading issue, it would be better to some description how
it is behaving . Plus some stack trace and logs would be grate help for us
to determine the issue.

The global Poller shares 10 threads for tasks by default. So, your
adapters doesn't work in single thread.

Yes, CachingSessionFactory is designed to be thread-safe. Therefore we
want to know more context about your app.

Maybe even you can come up with some simple Boot application to reproduce
an issue from our side...

—
You are receiving this because you authored the thread.
Reply to this email directly, view it on GitHub
#1980 (comment),
or mute the thread
https://github.com/notifications/unsubscribe-auth/ADVJndkmHsB_oL_JWJBBtvS7NPwa4qraks5rAP8ZgaJpZM4K2-ci
.

[ulmermark]

After much debugging, the issue definitely comes down to a threading issue in DefaultSftpSessionfactory or classes it interacts with.

I added the following code to that class as shown, and the issues go away.

I am not saying this is the correct solution, but somehow the use of the ReadWriteLock is not quite correct in this class.

@Override
_**synchronized**_ public SftpSession getSession() {

Also, made a more fined tune change to getSession() and it appears to work as well.....

		SftpSession sftpSession = new SftpSession(jschSession);
		**_this.sharedSessionLock.writeLock().lock();_**
		sftpSession.connect();
		jschSession.addChannel();
		return sftpSession;
	}
	catch (Exception e) {
		throw new IllegalStateException("failed to create SFTP Session", e);
	}
	**_finally {
		this.sharedSessionLock.writeLock().unlock();
	}_**

[artembilan]

I have attached logs from the 2 different configurations

Where they are? The email reply doesn't attache files to the GH issue.

And what is your threading problem exactly?

Usage of this.sharedSessionLock around sftpSession.connect(); isn't correct. Because it is exactly for the this.sharedJschSession.

[artembilan]

Please, note https://jira.spring.io/browse/INT-4175.

That still isn't clear that we are on the same page, but at least that is some fix because we have to release tomorrow.

[artembilan]

See #1987