Spring FTP入站通道适配器每隔一定时间对FTP进行日志记录

Question

@Bean
    @InboundChannelAdapter(value = "stream", poller = @Poller(fixedRate = "1000"))
    public MessageSource<InputStream> ftpMessageSource() {

        FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(), null);

        messageSource.setRemoteDirectory(remotedirectory);
        messageSource.setFilter(filter());

        return messageSource;
    }

 @Bean(name = PollerMetadata.DEFAULT_POLLER)
        public PollerMetadata defaultPoller() {
            PollerMetadata pollerMetadata = new PollerMetadata();
            pollerMetadata.setTrigger(new PeriodicTrigger(5000));
            return pollerMetadata;
        }

public static void main(String[] args) {
        SpringApplication.run(FtpinboundApp.class, args);
    }

    @Bean
    public SessionFactory<FTPFile> ftpSessionFactory() {
        DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
        sf.setHost(remotehost);
        sf.setPort(remoteport);
        sf.setUsername(remoteuser);
        sf.setPassword(remotepassword);
        return new CachingSessionFactory<FTPFile>(sf);
    }

    @Bean
    @ServiceActivator(inputChannel = "data", adviceChain = "after")
    public MessageHandler handler() {
        return new MessageHandler() {

            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                try {
                    httpposthgfiles.getHGFilesfromRestful(message.getPayload().toString());
                    httppost990.get990fromRestful(message.getPayload().toString());

                } catch (IOException e) {
                    logger.error(e);
                } catch (Exception e) {
                    logger.error(e);
                }
            }
        };
    }

    @Bean
    public ExpressionEvaluatingRequestHandlerAdvice after() {
        ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
        advice.setOnSuccessExpression("@template.remove(headers['file_remoteDirectory'] + headers['file_remoteFile'])");
        advice.setPropagateEvaluationFailures(true);
        return advice;
    }

    @Bean
    @InboundChannelAdapter(value = "stream", poller = @Poller(fixedRate = "1000"))
    public MessageSource<InputStream> ftpMessageSource() {

        FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(), null);

        messageSource.setRemoteDirectory(remotedirectory);
        messageSource.setFilter(filter());

        return messageSource;
    }

    public FileListFilter<FTPFile> filter() {
        CompositeFileListFilter<FTPFile> filter = new CompositeFileListFilter<>();
        filter.addFilter(new FtpSimplePatternFileListFilter("xxxx_aaa204*"));
        filter.addFilter(acceptOnceFilter());
        return filter;
    }

    @Bean
    public FtpPersistentAcceptOnceFileListFilter acceptOnceFilter() {
        FtpPersistentAcceptOnceFileListFilter filter = new FtpPersistentAcceptOnceFileListFilter(meta(), "xxxx_aaa204");
        filter.setFlushOnUpdate(true);
        return filter;
    }

    @Bean
    public ConcurrentMetadataStore meta() {
        PropertiesPersistingMetadataStore meta = new PropertiesPersistingMetadataStore();
        meta.setBaseDirectory("/tmp/foo");
        meta.setFileName("ftpStream.properties");
        return meta;
    }

    @Bean
    @Transformer(inputChannel = "stream", outputChannel = "data")
    public org.springframework.integration.transformer.Transformer transformer() {
        return new StreamTransformer("UTF-8");
    }

    @Bean
    public FtpRemoteFileTemplate template() {
        return new FtpRemoteFileTemplate(ftpSessionFactory());
    }

    @Bean(name = PollerMetadata.DEFAULT_POLLER)
    public PollerMetadata defaultPoller() {
        PollerMetadata pollerMetadata = new PollerMetadata();
        pollerMetadata.setTrigger(new PeriodicTrigger(5000));
        return pollerMetadata;
    }

Answer 1

只有在使用CachingSessionFactory时，它才会保持登录状态。

最好不要像那样Hibernate和绑定线程，而是使用任务调度器（这是轮询器所做的）。

New PerioadTrigger(600_000)将调度一个任务，每10分钟登录并检查一次文件。