在一台 FTP 服务器上丰富 Header 并在另一台 FTP 服务器上获取报头

Question

我已经成功地将文件从一台 FTP 服务器 (source) 发送到另一台 FTP 服务器 (target)。我首先使用入站适配器将文件从源发送到本地目录，然后使用出站适配器将文件从本地目录发送到目标。到目前为止，这一切正常。

我想要实现的是：用哈希码（使用source上传输的文件生成）丰富source处的消息头，然后在target 处获取该标头并将其与哈希码（使用target 上的文件生成）匹配

这是我迄今为止尝试过的：

Application.java

@SpringBootApplication
public class Application {

    @Autowired
    private Hashing hashing;

    public static ConfigurableApplicationContext context;

    public static void main(String[] args) {
        context = new SpringApplicationBuilder(Application.class)
                .web(false)
                .run(args);
    }

    @Bean
    @ServiceActivator(inputChannel = "ftpChannel")
    public MessageHandler sourceHandler() {
        return new MessageHandler() {

            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                System.out.println("Reply channel isssss:"+message.getHeaders().getReplyChannel());
                Object payload = message.getPayload();
                System.out.println("Payload: " + payload);
                File file = (File) payload;

                // enrich header with hash code before sending to target FTP
                Message<?> messageOut = MessageBuilder
                        .withPayload(message.getPayload())
                        .copyHeadersIfAbsent(message.getHeaders())
                        .setHeaderIfAbsent("hashCode", hashing.getHashCode(file)).build();

                // send to target FTP
                System.out.println("Trying to send " + file.getName() + " to target");
                MyGateway gateway = context.getBean(MyGateway.class);
                gateway.sendToFtp(messageOut);
            }

        };
    }
}

FileTransferServiceConfig.java

@Configuration
@Component
public class FileTransferServiceConfig {

    @Autowired
    private ConfigurationService configurationService;

    @Autowired
    private Hashing hashing;

    public static final String FILE_POLLING_DURATION = "5000";

    @Bean
    public SessionFactory<FTPFile> sourceFtpSessionFactory() {
        DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
        sf.setHost(configurationService.getSourceHostName());
        sf.setPort(Integer.parseInt(configurationService.getSourcePort()));
        sf.setUsername(configurationService.getSourceUsername());
        sf.setPassword(configurationService.getSourcePassword());
        return new CachingSessionFactory<>(sf);
    }

    @Bean
    public SessionFactory<FTPFile> targetFtpSessionFactory() {
        DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
        sf.setHost(configurationService.getTargetHostName());
        sf.setPort(Integer.parseInt(configurationService.getTargetPort()));
        sf.setUsername(configurationService.getTargetUsername());
        sf.setPassword(configurationService.getTargetPassword());
        return new CachingSessionFactory<>(sf);
    }

    @MessagingGateway
    public interface MyGateway {

        @Gateway(requestChannel = "toFtpChannel")
        void sendToFtp(Message message);

    }

    @Bean
    public FtpInboundFileSynchronizer ftpInboundFileSynchronizer() {
        FtpInboundFileSynchronizer fileSynchronizer = new FtpInboundFileSynchronizer(sourceFtpSessionFactory());
        fileSynchronizer.setDeleteRemoteFiles(false);
        fileSynchronizer.setRemoteDirectory(configurationService.getSourceDirectory());
        fileSynchronizer.setFilter(new FtpSimplePatternFileListFilter(
                configurationService.getFileMask()));
        return fileSynchronizer;
    }

    @Bean
    public AcceptOnceFileListFilter<File> acceptOnceFileListFilter() {
        return new AcceptOnceFileListFilter<>();
    }

    @Bean
    @InboundChannelAdapter(channel = "ftpChannel",
            poller = @Poller(fixedDelay = FILE_POLLING_DURATION))
    public MessageSource<File> ftpMessageSource() {
        FtpInboundFileSynchronizingMessageSource source
                = new FtpInboundFileSynchronizingMessageSource(ftpInboundFileSynchronizer());
        source.setLocalDirectory(new File(configurationService.getLocalDirectory()));
        source.setAutoCreateLocalDirectory(true);
        source.setLocalFilter(acceptOnceFileListFilter());
        return source;
    }

    // makes sure transfer continues on connection reset
    @Bean
    public Advice expressionAdvice() {
        ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
        advice.setTrapException(true);
        advice.setOnFailureExpression("@acceptOnceFileListFilter.remove(payload)");
        return advice;
    }

    @Bean
    @ServiceActivator(inputChannel = "toFtpChannel")
    public void listenOutboundMessage() {
        // tried to subscribe to "toFtpChannel" but this was not triggered
        System.out.println("Message received");
    }

    @Bean
    @ServiceActivator(inputChannel = "ftpChannel", adviceChain = "expressionAdvice")
    public MessageHandler targetHandler() {
        FtpMessageHandler handler = new FtpMessageHandler(targetFtpSessionFactory());
        handler.setRemoteDirectoryExpression(new LiteralExpression(
                configurationService.getTargetDirectory()));
        return handler;
    }

}

Hashing.java

public interface Hashing {
    public String getHashCode(File payload);
}

我已设法丰富sourceHandler() 中的消息，构建消息并将其发送到target，但我无法弄清楚如何在target 上接收该消息，以便我可以从消息？

如果需要更多信息，请告诉我。非常感谢您的帮助。

Answer 1

ftpChannel 上有两个订阅者 - 目标处理程序和您的 sourceHandler；除非将 ftpChannel 声明为 pubsub 频道，否则他们将收到备用消息。

您订阅toFtpChannel应该没有问题。

打开 DEBUG 日志记录以在应用程序上下文启动时查看所有订阅活动。

编辑

从@ServiceActivator 中删除@Bean - 这样的bean 必须是MessageHandler。

@ServiceActivator(inputChannel = "toFtpChannel")
public void listenOutboundMessage(Message message) {
    // tried to subscribe to "toFtpChannel" but this was not triggered
    System.out.println("Message received:" + message);
}

对我来说很好......

Payload: /tmp/foo/baz.txt
Trying to send baz.txt to target
Message received:GenericMessage [payload=/tmp/foo/baz.txt, headers={hashCode=foo, id=410eb9a2-fe8b-ea8a-015a-d5896387cf00, timestamp=1509115006278}]

再次； ftpChannel 上必须只有一个订阅者，除非您将其设为 pubsub。