Spring 集成 Java DSL SFTP 如何在处理程序中获取远程 SFTP 服务器信息

Question

我正在尝试从多个 SFTP 服务器下载文件，然后处理这些文件。但我无法获取远程 SFTP 服务器的信息，例如：IpAddress、remoteDirectory，具体取决于 MessageHandler 处理的文件。相反，Payload 仅包含本地下载文件的信息。这是我从指南中使用的源代码：

SFTIntegration.java

import com.jcraft.jsch.ChannelSftp.LsEntry;
import java.io.File;
import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec;
import org.springframework.integration.expression.FunctionExpression;
import org.springframework.integration.file.remote.aop.RotatingServerAdvice;
import org.springframework.integration.file.remote.session.DelegatingSessionFactory;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.sftp.dsl.Sftp;
import org.springframework.integration.sftp.dsl.SftpInboundChannelAdapterSpec;
import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
import org.springframework.messaging.MessageChannel;
import org.springframework.stereotype.Component;

/**
 * flow.
 */
@Configuration
@Component
public class SFTIntegration {
    public static final String TIMEZONE_UTC = "UTC";
    public static final String TIMESTAMP_FORMAT_OF_FILES = "yyyyMMddHHmmssSSS";
    public static final String TEMPORARY_FILE_SUFFIX = ".part";
    public static final int POLLER_FIXED_PERIOD_DELAY = 60000;
    public static final int MAX_MESSAGES_PER_POLL = 100;

    private static final Logger LOG = LoggerFactory.getLogger(SFTIntegration.class);
    private static final String CHANNEL_INTERMEDIATE_STAGE = "intermediateChannel";

    @Autowired
    private ImportHandler importHandler;

    /** database access repository */
    private final SFTPServerConfigRepo SFTPServerConfigRepo;

    @Value("${sftp.local.directory.download:${java.io.tmpdir}/localDownload}")
    private String localTemporaryPath;

    public SFTIntegration(final SFTPServerConfigRepo SFTPServerConfigRepo) {
        this.SFTPServerConfigRepo = SFTPServerConfigRepo;
    }

    /**
     * The default poller with 5s, 100 messages, RotatingServerAdvice and transaction.
     *
     * @return default poller.
     */
    @Bean(name = PollerMetadata.DEFAULT_POLLER)
    public PollerMetadata poller() {
        return Pollers
                .fixedDelay(POLLER_FIXED_PERIOD_DELAY)
                .advice(advice())
                .maxMessagesPerPoll(MAX_MESSAGES_PER_POLL)
                .transactional()
                .get();
    }

    /**
     * The direct channel for the flow.
     *
     * @return MessageChannel
     */
    @Bean
    public MessageChannel stockIntermediateChannel() {
        return new DirectChannel();
    }

    /**
     * Get the files from a remote directory. Add a timestamp to the filename
     * and write them to a local temporary folder.
     *
     * @return IntegrationFlow
     */
    @Bean
    public IntegrationFlow collectionInboundFlowFromSFTPServer() {
        // Source definition
        final SftpInboundChannelAdapterSpec sourceSpec = Sftp.inboundAdapter(delegatingSFtpSessionFactory())

                .preserveTimestamp(true)
                .patternFilter("*.*")
                .deleteRemoteFiles(true)
                .maxFetchSize(MAX_MESSAGES_PER_POLL)
                .remoteDirectory("/")
                .localDirectory(new File(localTemporaryPath))
                .temporaryFileSuffix(TEMPORARY_FILE_SUFFIX)
                .localFilenameExpression(new FunctionExpression<String>(s -> {
                    final int fileTypeSepPos = s.lastIndexOf('.');
                    return
                            DateTimeFormatter
                                    .ofPattern(TIMESTAMP_FORMAT_OF_FILES)
                                    .withZone(ZoneId.of(TIMEZONE_UTC))
                                    .format(Instant.now())
                                    + "_"
                                    + s.substring(0, fileTypeSepPos)
                                    + s.substring(fileTypeSepPos);
                }));

        // Poller definition
        final Consumer<SourcePollingChannelAdapterSpec> collectionInboundPoller = endpointConfigurer -> endpointConfigurer
                .id("collectionInboundPoller")
                .autoStartup(true)
                .poller(poller());

        return IntegrationFlows
                .from(sourceSpec, collectionInboundPoller)
                .transform(File.class, p -> {
                    // log step
                    LOG.info("flow=collectionInboundFlowFromSFTPServer, message=incoming file: " + p);
                    return p;
                })
                .channel(CHANNEL_INTERMEDIATE_STAGE)
                .get();
    }

    @Bean
    public IntegrationFlow collectionIntermediateStageChannel() {
        return IntegrationFlows
                .from(CHANNEL_INTERMEDIATE_STAGE)
                .handle(importHandler)
                .channel(new NullChannel())
                .get();
    }

    public DefaultSftpSessionFactory createNewSftpSessionFactory(final SFTPServerConfig pc) {
        final DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(
                false);
        factory.setHost(pc.getServerIp());
        factory.setPort(pc.getPort());
        factory.setUser(pc.getUsername());
        factory.setPassword(pc.getPassword());
        factory.setAllowUnknownKeys(true);
        return factory;
    }

    @Bean
    public DelegatingSessionFactory<LsEntry> delegatingSFtpSessionFactory() {
        final List<SFTPServerConfig> partnerConnections = SFTPServerConfigRepo.findAll();

        if (partnerConnections.isEmpty()) {
            return null;
        }

        final Map<Object, SessionFactory<LsEntry>> factories = new LinkedHashMap<>(10);

        for (SFTPServerConfig pc : partnerConnections) {
            // create a factory for every key containing server type, url and port
            if (factories.get(pc.getKey()) == null) {
                factories.put(pc.getKey(), createNewSftpSessionFactory(pc));
            }
        }

        // use the first SF as the default
        return new DelegatingSessionFactory<>(factories, factories.values().iterator().next());
    }

    @Bean
    public RotatingServerAdvice advice() {
        final List<SFTPServerConfig> sftpConnections = SFTPServerConfigRepo.findAll();

        final List<RotatingServerAdvice.KeyDirectory> keyDirectories = new ArrayList<>();
        for (SFTPServerConfig pc : sftpConnections) {
            keyDirectories
                    .add(new RotatingServerAdvice.KeyDirectory(pc.getKey(), pc.getServerPath()));
        }

        return new RotatingServerAdvice(delegatingSFtpSessionFactory(), keyDirectories, true);
    }
}

ImportHandler.java

import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;


@Service
public class ImportHandler {

    public void handle(Message<?> message) {
        System.out.println("Hello " + message);
        System.out.println(message.getPayload());
        System.out.println(message.getHeaders());
        //How can I get the information of remote server Ip address, remoteDirectory here where the file comes from
    }
}

如果您有任何想法，请告诉我。非常感谢！

Answer 1

暂不支持；请open a new feature request.