java - 如何为 Spring Integration SFTP 入站适配器动态定义文件过滤器模式？

Question

我需要将特定文件从不同 sftp 服务器的不同目录动态拉取到 spring boot 应用程序的本地服务器目录。

我将路径和文件模式存储在 postgres 数据库中。我一切正常，但我不知道如何根据 spring 集成 sftp 入站适配器的远程目录动态定义文件过滤器模式，以便不会提取该特定目录中的所有 xml 文件。

我使用 RotatingServerAdvice 和 DelegatingSessionFactory 来访问动态目录和服务器。

对于动态文件模式过滤器，我尝试使用

.filterFunction(f -> do_some_filtering_based_on_filename(f)

我想读取该文件的远程目录，但 f 来自 ChannelSftp.LsEntry 类型，并且没有包含远程目录的字段。否则我会从数据库加载配置数据，搜索路径并应用文件模式。

我的场景有更好的方法吗？

SFTP 示例:

127.0.0.1:22

目录:root/partner1

... test_p1_2343545.xml

... test_p1_453453.xml

... don_t_pull_this_file_453453.xml

127.0.0.2:22

目录:root/partner2

... companyname_2343545.xml

... companyname_453453.xml

... don_t_pull_this_file_3434.xml

数据库配置示例:

| URL       | PATH      | FILE_PATTERN      |
|-----------|-----------|-------------------|
| 127.0.0.1 | /partner1 | test_p1_*.xml     |
| 127.0.0.2 | /partner2 | companyname_*.xml |

我的 spring boot 应用程序的适配器类具有工作代码但由于 .patternFilter("*.xml") 而提取所有 xml 文件:

import com.jcraft.jsch.ChannelSftp;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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;

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;

/**
 * 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 = 5000;
  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";

  /** database access repository */
  private final PartnerConfigRepo partnerConfigRepo;

  @Value("${app.tmp-dir}")
  private String localTemporaryPath;

  public SFTIntegration(final PartnerConfigRepo partnerConfigRepo) {
    this.partnerConfigRepo = partnerConfigRepo;
  }

  /**
   * 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 stockInboundFlowFromSFTPServer() {
    // Source definition
    final SftpInboundChannelAdapterSpec sourceSpec = Sftp.inboundAdapter(delegatingSFtpSessionFactory())

        .preserveTimestamp(true)
        .patternFilter("*.xml")
        //.filterFunction(f -> do_some_filtering_based_on_filename(f, delegatingSFtpSessionFactory().getSession()))
        //.filter(new ModifiedFilter())
        //.filterExpression("#remoteDirectory")
        .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> stockInboundPoller = endpointConfigurer -> endpointConfigurer
        .id("stockInboundPoller")
        .autoStartup(true)
        .poller(poller());

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

  @Bean
  public IntegrationFlow stockIntermediateStageChannel() {
    return IntegrationFlows
        .from(CHANNEL_INTERMEDIATE_STAGE)
        .transform(p -> {
          //log step
          LOG.info("flow=stockIntermediateStageChannel, message=rename file: " + p);
          return p;
        })
        //TODO
        .channel(new NullChannel())
        .get();
  }

  public DefaultSftpSessionFactory createNewSftpSessionFactory(final PartnerConfigEntity 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<ChannelSftp.LsEntry> delegatingSFtpSessionFactory() {
    final List<PartnerConfigEntity> partnerConnections = partnerConfigRepo
        .findByTypeAndActiveIsTrue(PartnerConfigType.STOCK);

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

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

    for (PartnerConfigEntity 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<PartnerConfigEntity> partnerConnections = partnerConfigRepo.findByTypeAndActiveIsTrue("A_TYPE");

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

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

  //  private Boolean do_some_filtering_based_on_filename(final ChannelSftp.LsEntry f,
  //      final Session<ChannelSftp.LsEntry> session) {
  //    Object instance = session.getClientInstance();
  //    System.out.println(f);
  //    return true;
  //  }

  //
  //  private class ModifiedFilter<F> extends AbstractFileListFilter<F> {
  //    private final Logger log = LoggerFactory.getLogger(ModifiedFilter.class);
  //
  //    @Override public boolean accept(final F file) {
  //      log.info(file.toString());
  //      return false;
  //    }
  //  }
}

Answer 1

我建议您实现 SftpSimplePatternFileListFilter 逻辑的自定义变体，并有一个用于在运行时更改模式的 setter 。

然后您转到 RotatingServerAdvice 并在那里注入(inject)自定义 RotationPolicy，假设作为 StandardRotationPolicy 的副本，注入(inject)您的自定义FileListFilter 然后在 configureSource() 之后:

((AbstractInboundFileSynchronizingMessageSource<?>) source).getSynchronizer()
                    .setRemoteDirectory(this.current.getDirectory());

您执行数据库查询以通过 this.current.getDirectory() 获取模式并将其设置到您的 FileListFilter 中。

必须在代码中提到的 Sftp.inboundAdapter() 中使用相同的过滤器，而不是 .patternFilter("*.xml")。

我认为我们需要将 StandardRotationPolicy 中的所有属性设置为 protected 或使用 getter 以允许继承者访问。随时就此事提出 JIRA:https://jira.spring.io/browse/INT !

希望这能有所帮助。