测试中SFTP入站通道触发失败：receive方法挂起问题排查

我发现用Java DSL编写的IntegrationFlow可测试性不佳，于是改用Java配置方式拆分出@Bean定义。单元测试中使用内存SFTP服务器，尝试触发InboundChannelAdapter后调用通道的receive()方法，但测试在channel.receive()处挂起，调试发现sessionFactory.getSession()从未被调用。之前因为SFTP适配器文档没说明Channel类型困惑过，后来在测试示例里找到正确的QueueChannel，但现在还是卡壳了，请问哪里操作错了？

配置代码 #

@Bean
public PollableChannel sftpChannel() {
    return new QueueChannel();
}

@Bean
@EndpointId("sftpInboundAdapter")
@InboundChannelAdapter(channel = "sftpChannel", poller = @Poller(fixedDelay = "1000"))
public SftpInboundFileSynchronizingMessageSource sftpMessageSource() {
    SftpInboundFileSynchronizingMessageSource source =
            new SftpInboundFileSynchronizingMessageSource(sftpInboundFileSynchronizer);
    source.setLocalDirectory(new File("/local"));
    source.setAutoCreateLocalDirectory(true);
    source.setLocalFilter(new AcceptOnceFileListFilter<File>());
    source.setMaxFetchSize(6);
    return source;
}

@Bean
public SftpInboundFileSynchronizer sftpInboundFileSynchronizer() {
    SftpInboundFileSynchronizer fileSynchronizer = new SftpInboundFileSynchronizer(testSftpSessionFactory);
    fileSynchronizer.setDeleteRemoteFiles(false);
    fileSynchronizer.setPreserveTimestamp(true);
    fileSynchronizer.setRemoteDirectory("/remote");
    List<String> filterFileNameList = List.of("1.txt");
    fileSynchronizer.setFilter(new FilenameListFilter(filterFileNameList));

    return fileSynchronizer;
}

@Bean
private DefaultSftpSessionFactory testSftpSessionFactory(String username, String password, int port, String host) {
    DefaultSftpSessionFactory defaultSftpSessionFactory = new DefaultSftpSessionFactory();
    defaultSftpSessionFactory.setPassword("password");
    defaultSftpSessionFactory.setUser("username");
    defaultSftpSessionFactory.setHost("localhost");
    defaultSftpSessionFactory.setPort(777);
    defaultSftpSessionFactory.setAllowUnknownKeys(true);

    Properties config = new java.util.Properties();
    config.put( "StrictHostKeyChecking", "no" );
    defaultSftpSessionFactory.setSessionConfig(config);

    return defaultSftpSessionFactory;
}

测试代码 #

@ExtendWith(SpringExtension.class)
@ContextConfiguration(classes = {IntegrationFlowTestSupport.class, Synchronizer.class, Channel.class, Activator.class})
public class IntegrationFlowConfigTest {

    private static final String CONTENTS = "abcdef 1234567890";

    @Autowired
    PollableChannel sftpChannel;

    @Autowired
    DefaultSftpSessionFactory testSftpSessionFactory;

    @Autowired
    SftpInboundFileSynchronizer sftpInboundFileSynchronizer;

    @Autowired
    SftpInboundFileSynchronizingMessageSource sftpMessageSource;

    @Autowired
    SourcePollingChannelAdapter sftpInboundAdapter;

    @Test
    public void test() throws Exception {
        FileEntry f1 = new FileEntry("/remote/1.txt", CONTENTS);
        FileEntry f2 = new FileEntry("/remote/2.txt", CONTENTS);
        FileEntry f3 = new FileEntry("/remote/3.txt", CONTENTS);

        withSftpServer(server -> {
            server.setPort(777);
            server.addUser("username", "password");

            server.putFile(f1.getPath(), f1.createInputStream());
            server.putFile(f2.getPath(), f2.createInputStream());

            sftpInboundAdapter.start();
            Message<?> message = sftpChannel.receive();
        });
    }
}

问题分析与解决方案 #

核心问题及修复： #

1. SessionFactory Bean定义错误
   你的testSftpSessionFactory Bean用private修饰，还声明了未使用的方法参数(String username, String password, int port, String host)，Spring无法正确创建这个Bean，导致适配器无法获取SFTP会话。
   修复：改为public修饰，移除多余参数：

   @Bean
   public DefaultSftpSessionFactory testSftpSessionFactory() {
       DefaultSftpSessionFactory defaultSftpSessionFactory = new DefaultSftpSessionFactory();
       defaultSftpSessionFactory.setPassword("password");
       defaultSftpSessionFactory.setUser("username");
       defaultSftpSessionFactory.setHost("localhost");
       defaultSftpSessionFactory.setPort(777);
       defaultSftpSessionFactory.setAllowUnknownKeys(true);
   
       Properties config = new java.util.Properties();
       config.put( "StrictHostKeyChecking", "no" );
       defaultSftpSessionFactory.setSessionConfig(config);
   
       return defaultSftpSessionFactory;
   }
   

2. 轮询器初始延迟导致等待
   配置的@Poller仅设置了fixedDelay="1000"，默认初始延迟也是1000ms，测试中启动适配器后立即调用receive()，可能还没到第一次轮询时间。添加initialDelay="0"让轮询器启动后立即执行：

   @InboundChannelAdapter(channel = "sftpChannel", poller = @Poller(fixedDelay = "1000", initialDelay = "0"))
   

3. 避免无限等待
   channel.receive()默认无限等待，建议添加超时参数，比如channel.receive(5000)，5秒内无消息则返回null，方便排查问题。

4. 本地目录权限优化
   硬编码的/local目录可能在测试环境无写入权限，改用临时目录更可靠：

   source.setLocalDirectory(Files.createTempDirectory("sftp-local").toFile());
   

总结 #

优先修复SessionFactory的Bean定义问题，这是getSession()未被调用的核心原因。调整轮询器配置和receive()超时，能提升测试的稳定性和排查效率。

内容的提问来源于stack exchange，提问作者kacpr