Spring 集成：如何访问最后一个订阅者的返回值

Question

我正在尝试实现 2 个文件的 SFTP 文件上传，这必须按特定顺序进行 - 首先是一个 pdf 文件，然后成功上传一个包含有关 pdf 的元信息的文本文件。

我遵循了线程中的建议，但无法正常工作。

我的 Spring 引导配置：

@Bean
public SessionFactory<LsEntry> sftpSessionFactory() {
    final DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
    final Properties jschProps = new Properties();
    jschProps.put("StrictHostKeyChecking", "no");
    jschProps.put("PreferredAuthentications", "publickey,password");
    factory.setSessionConfig(jschProps);

    factory.setHost(sftpHost);
    factory.setPort(sftpPort);
    factory.setUser(sftpUser);
    if (sftpPrivateKey != null) {
        factory.setPrivateKey(sftpPrivateKey);
        factory.setPrivateKeyPassphrase(sftpPrivateKeyPassphrase);
    } else {
        factory.setPassword(sftpPasword);
    }
    factory.setAllowUnknownKeys(true);
    return new CachingSessionFactory<>(factory);
}

@Bean
@BridgeTo
public MessageChannel toSftpChannel() {
    return new PublishSubscribeChannel();
}


@Bean
@ServiceActivator(inputChannel = "toSftpChannel")
@Order(0)
public MessageHandler handler() {
    final SftpMessageHandler handler = new SftpMessageHandler(sftpSessionFactory());
    handler.setRemoteDirectoryExpression(new LiteralExpression(sftpRemoteDirectory));
    handler.setFileNameGenerator(message -> {
        if (message.getPayload() instanceof byte[]) {
            return (String) message.getHeaders().get("filename");
        } else {
            throw new IllegalArgumentException("File expected as payload.");
        }
    });
    return handler;
}

@ServiceActivator(inputChannel = "toSftpChannel")
@Order(1)
public String transferComplete(@Payload byte[] file, @Header("filename") String filename) {
    return "The SFTP transfer complete for file: " + filename;
}

@MessagingGateway
public interface UploadGateway {

    @Gateway(requestChannel = "toSftpChannel")
    String upload(@Payload byte[] file, @Header("filename") String filename);

}

我的测试用例：

final String pdfStatus = uploadGateway.upload(content, documentName);
log.info("Upload of {} completed, {}.", documentName, pdfStatus);

从网关上传调用的 return 我希望得到确认上传的字符串，例如"The SFTP transfer complete for file:..." 但我在 byte[]:

中得到上传文件的 returned 内容

Upload of 123456789.1.pdf completed, 37,80,68,70,45,49,46,54,13,37,-30,-29,-49,-45,13,10,50,55,53,32,48,32,111,98,106,13,60,60,47,76,105,110,101,97,114,105,122,101,100,32,49,47,76,32,50,53,52,55,49,48,47,79,32,50,55,55,47,69,32,49,49,49,55,55,55,47,78,32,49,47,84,32,50,53,52,51,53,57,47,72,32,91,32,49,49,57,55,32,53,51,55,93,62,62,13,101,110,100,111,98,106,13,32,32,32,32,32,32,32,32,32,32,32,32,13,10,52,55,49,32,48,32,111,98,106,13,60,60,47,68,101,99,111,100,101,80,97,114,109,115,60,60,47,67,111,108,117,109,110,115,32,53,47,80,114,101,100,105,99,116,111,114,32,49,50,62,62,47,70,105,108,116,101,114,47,70,108,97,116,101,68,101,99,111,100,101,47,73,68,91,60,57,66,53,49,56,54,69,70,53,66,56,66,49,50,52,49,65,56,50,49,55,50,54,56,65,65,54,52,65,57,70,54,62,60,68,52,50,68,51,55,54,53,54,65,67,48,55,54,52,65,65,53,52,66,52,57,51,50,56,52,56,68,66 etc.

我错过了什么？

Answer 1

我认为 @Order(0) 不能与 @Bean 一起使用。

要修复它，您应该在那个 bean 定义中这样做：

final SftpMessageHandler handler = new SftpMessageHandler(sftpSessionFactory());
handler.setOrder(0);

有关详细信息，请参阅 Reference Manual：

When using these annotations on consumer @Bean definitions, if the bean definition returns an appropriate MessageHandler (depending on the annotation type), attributes such as outputChannel, requiresReply etc, must be set on the MessageHandler @Bean definition itself.

换句话说：如果可以使用 setter，则必须使用。我们不处理这种情况的注释，因为不能保证应该优先处理什么。所以，为了避免这样的混淆，我们只给你留下了 setter 的选择。

更新

我看到了你的问题，它就在这里：

@Bean
@BridgeTo
public MessageChannel toSftpChannel() {
    return new PublishSubscribeChannel();
}

日志证实了这一点：

Adding {bridge:dmsSftpConfig.toSftpChannel.bridgeTo} as a subscriber to the 'toSftpChannel' channel
Channel 'org.springframework.context.support.GenericApplicationContext@b3d0f7.toSftpChannel' has 3 subscriber(s).
started dmsSftpConfig.toSftpChannel.bridgeTo

因此，您实际上还有一个 toSftpChannel 的订阅者，它是一个 BridgeHandler，输出到 replyChannel header。默认订单就像 private volatile int order = Ordered.LOWEST_PRECEDENCE; 这个成为第一个订阅者，正是这个 returns 你那个 byte[] 只是因为它是一个 payload 请求。

你需要决定你是否真的需要这样一座桥。尽管 @Order 没有解决方法...

Spring 集成：如何访问最后一个订阅者的返回值

Spring Integration: how to access the returned values from last Subscriber

spring-boot

spring-integration

spring-integration-sftp