配置了 SFTP 入站通道以从 SFTP 服务器获取文件，但我的测试用例无法激活它

【问题标题】：Configured a SFTP inbound channel to get a file from a SFTP server but my test case does not bring it alive配置了 SFTP 入站通道以从 SFTP 服务器获取文件，但我的测试用例无法激活它
【发布时间】：2021-08-16 14:32:59
【问题描述】：

我想从 SFTP 服务器获取文件，将 JSON 内容转换为我的一个类的实例数组。 因此，我曾考虑在 Spring Boot 应用程序中使用 Spring Integration，并且很高兴能够流畅地编程其中一个 IntegrationFlows 来实现这一目标。

到目前为止，我在 SO 中深入研究了许多文章、问题和答案，但没有任何帮助。几周以来，我一直在为此苦苦挣扎。 Spring Integration 文档本身很难理解。根据 Spring 版本和不同的编程范式（XML、Java 配置和 Java DSL），它带有很多不同的方面，这给问题带来了更多的复杂性，使得更难决定，哪些是罕见的例子，最终对新手毫无帮助。一个新手想要找到他的问题，并希望被推荐和引导到当前解决他的问题的最佳方法，被解释为什么和几率和偶数。如果他足够高兴，他将能够在其中一个示例中找到他的问题（从 SFTP 获取文件以进一步处理它，这不是一项奇特的任务）并且解决方案非常适合示例，他可以以最少的速度复制和粘贴适应。直到现在我才这么幸运。

SO 中的一个问题接近我可能需要的：配置了一个 sftpInboundFileSynchonizer，以及一个 MessageSource 和一个 MassageHandler。提问者想和我做的大致相同。答案是“那个代码正在做相反的事情（发送到 SFTP）。”让我瞠目结舌的想，我是不是把inbound和outbound的理解基本搞混了？

我认为我必须使用基于 SftpSessionFactory 的 InboundAdapter。我将 SessionFactory 放入一个额外的配置中，因为我还计划将它与其他适配器一起使用：

@Configuration
@ConfigurationProperties("test.job.common")
public class JobCommonConfiguration {
    private static final Logger LOG = LoggerFactory.getLogger(JobCommonConfiguration.class);

    private String hostname;
    private int port;
    private String username;
    private String password;

    @Bean
    public SessionFactory<LsEntry> sftpTest3SessionFactory() {
        DefaultSftpSessionFactory sf = new DefaultSftpSessionFactory();
        sf.setHost(hostname);
        sf.setPort(port);
        sf.setUser(username);
        sf.setPassword(password);
//      factory.setTestSession(true);
        return new CachingSessionFactory<LsEntry>(sf);
    }

    /* getters & setters */

}

第二个配置是配置一个 SftpInboundAdapter，假设我对“入站”的理解是正确的，它带有一个 SI Transformer 来将 JSON 转换为我的 Event 实例的数组。最后，实例应该通过请求发送到 HTTP REST 服务，我也可以将其包含到适配器中吗？如下：

@Configuration
@ConfigurationProperties("test.job.transfer3")
@Import({ JobCommonConfiguration.class })
public class Job3Configuration {
    private static final Logger LOG = LoggerFactory.getLogger(Job3Configuration.class);

    private boolean enabled = false;
    private String remoteDir;
    private String remoteFile;
    private String remoteFilePattern;
    private boolean remoteRemove;
    private String localDir;
    private String localFile;

    @Autowired
    private SessionFactory<LsEntry> sftpTest3SessionFactory;

    @Bean
    public FireOnceTrigger fireOnceTest3Trigger() {
        return new FireOnceTrigger();
    }

    @Bean
    public IntegrationFlow test3SftpInboundFlow() {
        return IntegrationFlows
            .from(Sftp.inboundAdapter(sftpTest3SessionFactory)
                .preserveTimestamp(true)
                .remoteDirectory(remoteDir)
                .regexFilter(remoteFilePattern)
                .localFilenameExpression(localFile)
                .localDirectory(new File(localDir)),
            e -> e.id("sftpTest3InboundAdapter")
                .autoStartup(true)
                .poller(Pollers.trigger(fireOnceTest3Trigger()))
            )
            .transform(Transformers.fromJson(Event[].class))
            .handle(m -> System.out.println(m.getPayload()))
            .get();
    }

    /* getters & setters */
}

我的实体事件很简单：

public class Event {

    private Integer crmId;
    private String eventType;
    private LocalDateTime dateFrom;
    private LocalDateTime dateTo;

    /* getters & setter & toString() */
}

我的测试还没有测试任何东西，因为到目前为止还没有发生任何事情。然后它应该断言我收到了正确数量的事件。如下所示：

@RunWith(SpringRunner.class)
@SpringBootTest
@ContextConfiguration(classes=SftpTransferTestApplication.class)
@Import(Job3Configuration.class)
public class GetFileIntegrationTest {
    private static final Logger LOG = LoggerFactory.getLogger(GetFileIntegrationTest.class);

    @Test
    public void testGetFile() throws Exception {
        LOG.info("GetIntegrationTest testgetFile");

//      assertThat(fileReceived, is(sb.toString()));
    }
}

Test 应用程序很简单：

@SpringBootApplication
@EnableIntegration
@IntegrationComponentScan
public class SftpTransferTestApplication {

    public static void main(String[] args) {
        new SpringApplicationBuilder(SftpTransferTestApplication.class).web(
            NONE).run(args);
    }
}

我的项目有 2.3.0.RELEASE 版本的 spring-boot-starter-parent 作为父项目，并使用 5.3.0.RELEASE 版本的 spring-integration-sftp。

请帮助我激活这个测试用例。我哪里错了？

如何将 Logger 功能包含到 IntegrationFlow 中以查看更多正在发生（或未发生）的事情？

编辑 1

我尝试稍微剥离我的代码以避免配置问题：

@Configuration
public class JobConfiguration {
    private static final Logger LOG = LoggerFactory.getLogger(JobConfiguration.class);
    @Bean
    public TransferChannel getTransferChannel() {
        TransferChannel channel = new TransferChannel();
        channel.setHost("myHost");
        channel.setPort(0);
        channel.setUser("test");
        channel.setPassword("xxx");
        return channel;
    }

    @Bean
    public TransferContext getTransferContext() {
        TransferContext context = new TransferContext();
        context.setEnabled(false);
        context.setChannel(getTransferChannel());
        context.setRemoteDir("data");
        context.setRemoteFilename("GetMessage3.json");
        context.setRemoteFilenameFilter("GetMessage\\.json$");
        context.setLocalDir("sftp-inbound");
        context.setLocalFile("GetMessage3.json");
        return context;
    }

    @Bean
    public SessionFactory<LsEntry> getSftpTestSessionFactory(TransferChannel transferChannel) {
        DefaultSftpSessionFactory sf = new DefaultSftpSessionFactory();
        sf.setHost(transferChannel.getHost());
        sf.setPort(transferChannel.getPort());
        sf.setUser(transferChannel.getUser());
        sf.setPassword(transferChannel.getPassword());
//      factory.setTestSession(true);
        return new CachingSessionFactory<LsEntry>(sf);
    }

    @Bean
    public FireOnceTrigger fireOnceTestTrigger() {
        return new FireOnceTrigger();
    }

    @Bean
    public IntegrationFlow testSftpInboundFlow(TransferContext context) {
        return IntegrationFlows
            .from(Sftp.inboundAdapter(getSftpTestSessionFactory(context.getChannel()))
                    .preserveTimestamp(true)
                    .remoteDirectory(context.getRemoteDir())
                    .regexFilter(context.getRemoteFilenameFilter())
                    .localFilenameExpression(context.getLocalFile())
                    .localDirectory(new File(context.getLocalDir())),
                e -> e.id("sftpTestInboundAdapter")
                    .autoStartup(true)
                    .poller(Pollers.trigger(fireOnceTestTrigger()))
//                  .poller(Pollers.fixedDelay(5000))
                )
            .transform(Transformers.fromJson(Event[].class))
            .channel("sftpTestMessageChannel")
//          .logAndReply(Level.DEBUG);
//          .handle(m -> System.out.println("myHandler: " + m.getPayload().toString()))
            .get();
    }

    @Bean("someService.handler")             
    @EndpointId("someService")               
    @ServiceActivator(inputChannel = "sftpTestMessageChannel")
    public MessageHandler someHandler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                System.out.println("myThirdHandler: " + message.getPayload());
                System.out.println("myThirdHandler: " + message.getHeaders());
                Event ev = ((Event[]) message.getPayload())[1];
                System.out.println("myThirdHandler: " + ev);
                throw new IllegalStateException("Want to see the next MessageHandler");
            }
        };
    }
}

为了让ftp实现Jsch多说一点，我有一个配置“application-junit.yaml”：

logging:
  level:
    org.springframework.integration: debug
    ### filter warning
    ### org.springframework.integration.expression.ExpressionUtils:
    ### Creating EvaluationContext with no beanFactory
    org.springframework.integration.expression.ExpressionUtils: error
    ### filter info
    ###    because jsch is very talkative
    com.jcraft.jsch: debug
    com.harry.potter: debug
    com.harry.potter.filetransfer: trace

我无法让它工作。日志输出为：

  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::        (v2.3.0.RELEASE)

2021-08-22 23:29:27.958  INFO 9916 --- [           main] c.l.c.f.service.GetFileIntegrationTest   : Starting GetFileIntegrationTest on myClient with PID 9916 (started by test in C:\Users\test\git\tis\sftp-client)
2021-08-22 23:29:27.959 DEBUG 9916 --- [           main] c.l.c.f.service.GetFileIntegrationTest   : Running with Spring Boot v2.3.0.RELEASE, Spring v5.2.6.RELEASE
2021-08-22 23:29:27.959  INFO 9916 --- [           main] c.l.c.f.service.GetFileIntegrationTest   : No active profile set, falling back to default profiles: default
2021-08-22 23:29:28.752  INFO 9916 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
2021-08-22 23:29:28.759  INFO 9916 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'taskScheduler' has been explicitly defined. Therefore, a default ThreadPoolTaskScheduler will be created.
2021-08-22 23:29:28.763 DEBUG 9916 --- [           main] faultConfiguringBeanFactoryPostProcessor : SpEL function '#xpath' isn't registered: there is no spring-integration-xml.jar on the classpath.
2021-08-22 23:29:28.766  INFO 9916 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'integrationHeaderChannelRegistry' has been explicitly defined. Therefore, a default DefaultHeaderChannelRegistry will be created.
2021-08-22 23:29:28.837  INFO 9916 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationChannelResolver' of type [org.springframework.integration.support.channel.BeanFactoryChannelResolver] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2021-08-22 23:29:28.841  INFO 9916 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationDisposableAutoCreatedBeans' of type [org.springframework.integration.config.annotation.Disposables] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2021-08-22 23:29:28.873  INFO 9916 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.integration.config.IntegrationManagementConfiguration' of type [org.springframework.integration.config.IntegrationManagementConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2021-08-22 23:29:30.298  INFO 9916 --- [           main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService 'applicationTaskExecutor'
2021-08-22 23:29:30.802  INFO 9916 --- [           main] o.s.s.c.ThreadPoolTaskScheduler          : Initializing ExecutorService 'taskScheduler'
2021-08-22 23:29:30.894 DEBUG 9916 --- [           main] faultConfiguringBeanFactoryPostProcessor : 
Spring Integration global properties:
    
    spring.integration.endpoints.noAutoStartup=
    spring.integration.taskScheduler.poolSize=10
    spring.integration.channels.maxUnicastSubscribers=0x7fffffff
    spring.integration.channels.autoCreate=true
    spring.integration.channels.maxBroadcastSubscribers=0x7fffffff
    spring.integration.readOnly.headers=
    spring.integration.messagingTemplate.throwExceptionOnLateReply=false
  
2021-08-22 23:29:30.901 DEBUG 9916 --- [           main] .s.i.c.GlobalChannelInterceptorProcessor : No global channel interceptors.
2021-08-22 23:29:30.904  INFO 9916 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
2021-08-22 23:29:30.904  INFO 9916 --- [           main] o.s.i.channel.PublishSubscribeChannel    : Channel 'application.errorChannel' has 1 subscriber(s).
2021-08-22 23:29:30.904  INFO 9916 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean '_org.springframework.integration.errorLogger'
2021-08-22 23:29:30.905  INFO 9916 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {json-to-object-transformer} as a subscriber to the 'testSftpInboundFlow.channel#0' channel
2021-08-22 23:29:30.905  INFO 9916 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'application.testSftpInboundFlow.channel#0' has 1 subscriber(s).
2021-08-22 23:29:30.905  INFO 9916 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean 'testSftpInboundFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0'; defined in: 'class path resource [com/harry/potter/filetransfer/config/JobConfiguration.class]'; from source: 'bean method testSftpInboundFlow'
2021-08-22 23:29:30.905  INFO 9916 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {message-handler:someService} as a subscriber to the 'sftpTestMessageChannel' channel
2021-08-22 23:29:30.905  INFO 9916 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'application.sftpTestMessageChannel' has 1 subscriber(s).
2021-08-22 23:29:30.905  INFO 9916 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean 'someService'
2021-08-22 23:29:30.910  INFO 9916 --- [           main] o.s.i.e.SourcePollingChannelAdapter      : started bean 'sftpTestInboundAdapter'; defined in: 'class path resource     [com/harry/potter/filetransfer/config/JobConfiguration.class]'; from source: 'bean method testSftpInboundFlow'
2021-08-22 23:29:30.922  INFO 9916 --- [           main] c.l.c.f.service.GetFileIntegrationTest   : Started GetFileIntegrationTest in 3.323 seconds (JVM running for 4.13)
2021-08-22 23:29:30.935  INFO 9916 --- [ask-scheduler-1] com.jcraft.jsch                          : Connecting to myHost port 22
2021-08-22 23:29:30.968  INFO 9916 --- [ask-scheduler-1] com.jcraft.jsch                          : Connection established
2021-08-22 23:29:31.041  INFO 9916 --- [ask-scheduler-1] com.jcraft.jsch                          : Remote version string: SSH-2.0-OpenSSH_for_Windows_8.1
2021-08-22 23:29:31.042  INFO 9916 --- [ask-scheduler-1] com.jcraft.jsch                          : Local version string: SSH-2.0-JSCH-0.1.54
2021-08-22 23:29:31.042  INFO 9916 --- [ask-scheduler-1] com.jcraft.jsch                          : CheckCiphers: aes256-ctr,aes192-ctr,aes128-ctr,aes256-cbc,aes192-cbc,aes128-cbc,3des-ctr,arcfour,arcfour128,arcfour256
2021-08-22 23:29:31.216  INFO 9916 --- [           main] c.l.c.f.service.GetFileIntegrationTest   : GetIntegrationTest testgetFile
2021-08-22 23:29:31.226  INFO 9916 --- [extShutdownHook] o.s.i.e.SourcePollingChannelAdapter      : stopped bean 'sftpTestInboundAdapter'; defined in: 'class path resource [com/harry/potter/filetransfer/config/JobConfiguration.class]'; from source: 'bean method testSftpInboundFlow'
2021-08-22 23:29:31.227  INFO 9916 --- [extShutdownHook] o.s.i.endpoint.EventDrivenConsumer       : Removing {json-to-object-transformer} as a subscriber to the 'testSftpInboundFlow.channel#0' channel
2021-08-22 23:29:31.227  INFO 9916 --- [extShutdownHook] o.s.integration.channel.DirectChannel    : Channel 'application.testSftpInboundFlow.channel#0' has 0 subscriber(s).
2021-08-22 23:29:31.227  INFO 9916 --- [extShutdownHook] o.s.i.endpoint.EventDrivenConsumer       : stopped bean 'testSftpInboundFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0'; defined in: 'class path resource [com/harry/potter/filetransfer/config/JobConfiguration.class]'; from source: 'bean method testSftpInboundFlow'
2021-08-22 23:29:31.227  INFO 9916 --- [extShutdownHook] o.s.i.endpoint.EventDrivenConsumer       : Removing {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
2021-08-22 23:29:31.227  INFO 9916 --- [extShutdownHook] o.s.i.channel.PublishSubscribeChannel    : Channel 'application.errorChannel' has 0 subscriber(s).
2021-08-22 23:29:31.227  INFO 9916 --- [extShutdownHook] o.s.i.endpoint.EventDrivenConsumer       : stopped bean '_org.springframework.integration.errorLogger'
2021-08-22 23:29:31.227  INFO 9916 --- [extShutdownHook] o.s.i.endpoint.EventDrivenConsumer       : Removing {message-handler:someService} as a subscriber to the 'sftpTestMessageChannel' channel
2021-08-22 23:29:31.227  INFO 9916 --- [extShutdownHook] o.s.integration.channel.DirectChannel    : Channel 'application.sftpTestMessageChannel' has 0 subscriber(s).
2021-08-22 23:29:31.227  INFO 9916 --- [extShutdownHook] o.s.i.endpoint.EventDrivenConsumer       : stopped bean 'someService'
2021-08-22 23:29:31.227  INFO 9916 --- [extShutdownHook] o.s.s.c.ThreadPoolTaskScheduler          : Shutting down ExecutorService 'taskScheduler'
2021-08-22 23:29:31.228  INFO 9916 --- [extShutdownHook] o.s.s.concurrent.ThreadPoolTaskExecutor  : Shutting down ExecutorService 'applicationTaskExecutor'

我用 Pollers.fixedDelay(5000) 替换了 FireOnceTestTrigger，什么都没有。

我做错了什么？

【问题讨论】：

  • 这个问题有问题吗？文字太多？无法解决的问题？我应该不使用 SI 来解决这个问题吗？要求看不懂？我会改变它。每个提示都非常受欢迎。
  • 第一步是开启DEBUG日志；您应该看到轮询器正在运行；如果你没有看到它，那么根本就错了。
  • 我到现在调试的是，只要文件已经在本地目录localDir中，处理就可以了。第一步——从远程主机获取它——还没有完成。我什至没有看到 jsch 在做什么。
  • 可能与您的FireOnceTrigger 有关系您是否尝试过使用普通（例如Pollers.fixedDelay()）触发器？
  • 我会检查的。目前我无法让我的项目响应 DEBUG 级别的更改。

标签： java spring spring-integration spring-integration-dsl spring-integration-sftp


【解决方案1】：

测试正在退出，JVM 在获取完成之前关闭。

您需要等待文件被提取。一种方法是在您的测试中添加一个通道拦截器。像这样的：

public class GetFileIntegrationTest {
    private static final Logger LOG = LoggerFactory.getLogger(GetFileIntegrationTest.class);

    @Autowired
    AbstractMessageChannel sftpTestMessageChannel;

    @Test
    public void testGetFile() throws Exception {
        LOG.info("GetIntegrationTest testgetFile");
        CountDownLatch latch = new CountDownLatch(1);
        this.sftpTestMessageChannel.addInterceptor(new ChannelInterceptor() {
                        
             // override preSend and/or postSend, capture the message and 
             // count down the latch.
        });

        assertTrue(latch.await(10, TimeUnit.SECONDS));
//      assertThat(fileReceived, is(sb.toString()));
    }
}

【讨论】：

  • 非常感谢，加里！我有点惭愧，你的一个回答会在我脑海中引发十几个问题。直到现在我才用过 Latch。为什么它从未在您的 SftpTests 中使用？我喜欢测试，除了验证代码之外，还展示了我可以用令人印象深刻的 SI 包做的事情的用例，以及如何以正确的方式使用对象。文档主要展示了如何配置东西，而不是为了我需要它们以及我如何使用它们。我认为消息和接收文件的处理应该由我的 MessageHandler 来处理？
  • 重写 preSend- 和/或 postSend 方法是发送消息的一部分。我以为我会收到一条消息，说我请求的文件是从 SFTP 服务器获取的？我可能愿意发送什么样的按摩？
  • 不清楚你的意思；该框架在其测试中广泛使用了闩锁；它们是测试异步应用程序的重要工具。我不确定您在第二条评论中的意思；当检索到文件时，会向通道发送一条消息。拦截器允许测试拦截该消息并对其进行检查。
  • 我明白了。从 Spring Integration 的角度来看，这确实是一个发送过程。 Spring Integration 向我（我的 SFTP 客户端）发送一条消息，通知我请求的文件现在可用。对吗？
  • 谢谢你，加里，你的回答！你能用简短的话解释一下谁在这里等待什么吗？这个线程用1创建latch来等待一个事件？在覆盖的通道拦截器中填充的事件？另一个线程的拦截器倒计时一，向这个线程发出事件发生的信号？该线程正在其latch.await() 方法中等待事件？超时等待不是永远。是这个吗？
猜你喜欢
  • 1970-01-01
  • 2016-03-10
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
相关资源
最近更新 更多
热门标签
Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode