CODEWITHSUNDEEP
spring-integrationspring-integration-sftp
Bastien verschaete
Bastien verschaete

Reputation: 445

StreamingMessageSource keeps firing when a filter is applied

I am trying to poll an FTP directory for a certain kind of file, the polling of a directory works, but whenever I try to apply a filter to filter the files by extension, the messagesource keeps spamming messages about the file with no regard to the polling delay. Without the filters everything works fine, once I enable them my application authenticates with the FTP, downloads the file and sends the message nonstop over and over again. I have the following beans:

/**
 * Factory that creates the remote connection
 *
 * @return DefaultSftpSessionFactory
 */
@Bean
public DefaultSftpSessionFactory sftpSessionFactory(@Value("${ftp.host}") String ftpHost,
                                                    @Value("${ftp.port}") int ftpPort,
                                                    @Value("${ftp.user}") String ftpUser,
                                                    @Value("${ftp.pass}") String ftpPass) {
    DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory();
    factory.setAllowUnknownKeys(true);
    factory.setHost(ftpHost);
    factory.setPort(ftpPort);
    factory.setUser(ftpUser);
    factory.setPassword(ftpPass);

    return factory;
}

/**
 * Template to handle remote files
 *
 * @param sessionFactory SessionFactory bean
 * @return SftpRemoteFileTemplate
 */
@Bean
public SftpRemoteFileTemplate fileTemplate(DefaultSftpSessionFactory sessionFactory) {
    SftpRemoteFileTemplate template = new SftpRemoteFileTemplate(sessionFactory);
    template.setAutoCreateDirectory(true);
    template.setUseTemporaryFileName(false);
    return template;
}

/**
 * To listen to multiple directories, declare multiples of this bean with the same inbound channel
 *
 * @param fileTemplate FileTemplate bean
 * @return MessageSource
 */
@Bean
@InboundChannelAdapter(channel = "deeplinkAutomated", poller = @Poller(fixedDelay = "6000", maxMessagesPerPoll = "-1"))
public MessageSource inboundChannelAdapter(SftpRemoteFileTemplate fileTemplate) {
    SftpStreamingMessageSource source = new SftpStreamingMessageSource(fileTemplate);
    source.setRemoteDirectory("/upload");
    source.setFilter(new CompositeFileListFilter<>(
            Arrays.asList(new AcceptOnceFileListFilter<>(), new SftpSimplePatternFileListFilter("*.trg"))
    ));

    return source;
}

/**
 * Listener that activates on new messages on the specified input channel
 *
 * @return MessageHandler
 */
@Bean
@ServiceActivator(inputChannel = "deeplinkAutomated")
public MessageHandler handler(JobLauncher jobLauncher, Job deeplinkBatch) {
    return message -> {
        Gson gson = new Gson();
        SFTPFileInfo info = gson.fromJson((String) message.getHeaders().get("file_remoteFileInfo"), SFTPFileInfo.class);
        System.out.println("File to download: " + info.getFilename().replace(".trg", ".xml"));
    };
}

Upvotes: 1

Views: 384

Answers (1)

Artem Bilan
Artem Bilan

Reputation: 121272

I think AcceptOnceFileListFilter is not suitable for SFTP files. The returned LsEntry doesn't match previously stored in the HashSet: just their hashes are different!

Consider to use a SftpPersistentAcceptOnceFileListFilter instead.

Also it would be better to configure a DefaultSftpSessionFactory for the isSharedSession:

/**
 * @param isSharedSession true if the session is to be shared.
 */
public DefaultSftpSessionFactory(boolean isSharedSession) {

To avoid session recreation on each polling task.

you don't have a 6 seconds delay between calls because you have a maxMessagesPerPoll = "-1". That means poll remote files until they are there in remote dir. In your case with the AcceptOnceFileListFilter you always end up with the same file by the hash reason.

Upvotes: 1

Related Questions