title: How to dynamically define file filter pattern for Spring Integration SFTP Inbound Adapter?
body: <p>I need to dynamically pull specific files from different directories from different sftp servers to a local server directory of a spring boot application.</p>
  
  <p>I store the paths and a file pattern in a postgres database. I got everything working but i don't know how to dynamically define a file filter pattern depending on the remote directory for spring integration sftp inbound adapter so that not all xml files in that specific directory are pulled. </p>
  
  <p>I use a RotatingServerAdvice and a DelegatingSessionFactory for accessing dynamic directories and servers.</p>
  
  <p>For the dynamic file pattern filter I tried for instance to use the </p>
  
  <pre><code>.filterFunction(f -&gt; do_some_filtering_based_on_filename(f)
  </code></pre>
  
  <p>I wanted to read the remote directory of that file but f is from the type ChannelSftp.LsEntry and doesn't have a field with the remote directory. Otherwise I would have loaded the config data from the database, searched for the path and applied the file pattern.</p>
  
  <p>Is there a better way for my scenario?</p>
  
  <p><strong>SFTP example:</strong></p>
  
  <p>127.0.0.1:22</p>
  
  <p>directory: root/partner1</p>
  
  <blockquote>
    <p>... test_p1_2343545.xml</p>
    
    <p>... test_p1_453453.xml</p>
    
    <p>... don_t_pull_this_file_453453.xml</p>
  </blockquote>
  
  <p>127.0.0.2:22</p>
  
  <p>directory: root/partner2</p>
  
  <blockquote>
    <p>... companyname_2343545.xml</p>
    
    <p>... companyname_453453.xml </p>
    
    <p>... don_t_pull_this_file_3434.xml</p>
  </blockquote>
  
  <p><strong>Database configuration example:</strong></p>
  
  <pre><code>| URL       | PATH      | FILE_PATTERN      |
  |-----------|-----------|-------------------|
  | 127.0.0.1 | /partner1 | test_p1_*.xml     |
  | 127.0.0.2 | /partner2 | companyname_*.xml |
  </code></pre>
  
  <p><strong>Adapter Class of my spring boot application with working code but pulling all xml files because of .patternFilter("*.xml"):</strong></p>
  
  <pre class="lang-java prettyprint-override"><code>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 -&gt; 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&lt;String&gt;(s -&gt; {
            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&lt;SourcePollingChannelAdapterSpec&gt; stockInboundPoller = endpointConfigurer -&gt; endpointConfigurer
          .id("stockInboundPoller")
          .autoStartup(true)
          .poller(poller());
  
      return IntegrationFlows
          .from(sourceSpec, stockInboundPoller)
          .transform(File.class, p -&gt; {
            // 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 -&gt; {
            //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&lt;ChannelSftp.LsEntry&gt; delegatingSFtpSessionFactory() {
      final List&lt;PartnerConfigEntity&gt; partnerConnections = partnerConfigRepo
          .findByTypeAndActiveIsTrue(PartnerConfigType.STOCK);
  
      if (partnerConnections.isEmpty()) {
        return null;
      }
  
      final Map&lt;Object, SessionFactory&lt;ChannelSftp.LsEntry&gt;&gt; factories = new LinkedHashMap&lt;&gt;(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&lt;&gt;(factories, factories.values().iterator().next());
    }
  
    @Bean
    public RotatingServerAdvice advice() {
      final List&lt;PartnerConfigEntity&gt; partnerConnections = partnerConfigRepo.findByTypeAndActiveIsTrue("A_TYPE");
  
      final List&lt;RotatingServerAdvice.KeyDirectory&gt; keyDirectories = new ArrayList&lt;&gt;();
      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&lt;ChannelSftp.LsEntry&gt; session) {
    //    Object instance = session.getClientInstance();
    //    System.out.println(f);
    //    return true;
    //  }
  
    //
    //  private class ModifiedFilter&lt;F&gt; extends AbstractFileListFilter&lt;F&gt; {
    //    private final Logger log = LoggerFactory.getLogger(ModifiedFilter.class);
    //
    //    @Override public boolean accept(final F file) {
    //      log.info(file.toString());
    //      return false;
    //    }
    //  }
  }
  </code></pre>