RequestHandlerRetryAdvice нельзя заставить работать с Ftp.outboundGateway в Spring Integration

Question

Моя ситуация аналогична описанной в этом вопросе .Разница в том, что я использую не WebFlux.outboundGateway, а Ftp.outboundGateway, для которого я вызываю команду AbstractRemoteFileOutboundGateway.Command.GET, общая проблема в том, что я не могу получить определенный RequestHandlerRetryAdvice для использования.

Конфигурация выглядит следующим образом (разделена на соответствующие части):

@RestController
@RequestMapping( value = "/somepath" )
public class DownloadController
{
   private DownloadGateway downloadGateway;

   public DownloadController( DownloadGateway downloadGateway )
   {
      this.downloadGateway = downloadGateway;
   }

   @PostMapping( "/downloads" )
   public void download( @RequestParam( "filename" ) String filename )
   {
      Map<String, Object> headers = new HashMap<>();

      downloadGateway.triggerDownload( filename, headers );
   }
}    

@MessagingGateway
public interface DownloadGateway
{
   @Gateway( requestChannel = "downloadFiles.input" )
   void triggerDownload( Object value, Map<String, Object> headers );
}

@Configuration
@EnableIntegration
public class FtpDefinition
{
   private FtpProperties ftpProperties;

   public FtpDefinition( FtpProperties ftpProperties )
   {
      this.ftpProperties = ftpProperties;
   }

   @Bean
   public DirectChannel gatewayDownloadsOutputChannel()
   {
      return new DirectChannel();
   }

   @Bean
   public IntegrationFlow downloadFiles( RemoteFileOutboundGatewaySpec<FTPFile, FtpOutboundGatewaySpec> getRemoteFile )
   {
      return f -> f.handle( getRemoteFile, getRetryAdvice() )
                   .channel( "gatewayDownloadsOutputChannel" );
   }

   private Consumer<GenericEndpointSpec<AbstractRemoteFileOutboundGateway<FTPFile>>> getRetryAdvice()
   {
      return e -> e.advice( ( (Supplier<RequestHandlerRetryAdvice>) () -> {
         RequestHandlerRetryAdvice advice = new RequestHandlerRetryAdvice();
         advice.setRetryTemplate( getRetryTemplate() );
         return advice;
      } ).get() );
   }

   private RetryTemplate getRetryTemplate()
   {
      RetryTemplate result = new RetryTemplate();

      FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
      backOffPolicy.setBackOffPeriod( 5000 );

      result.setBackOffPolicy( backOffPolicy );
      return result;
   }

   @Bean
   public RemoteFileOutboundGatewaySpec<FTPFile, FtpOutboundGatewaySpec> getRemoteFile( SessionFactory sessionFactory )
   {
      return 
         Ftp.outboundGateway( sessionFactory,
                              AbstractRemoteFileOutboundGateway.Command.GET,
                              "payload" )
            .fileExistsMode( FileExistsMode.REPLACE )
            .localDirectoryExpression( "'" + ftpProperties.getLocalDir() + "'" )
            .autoCreateLocalDirectory( true );
   }

   @Bean
   public SessionFactory<FTPFile> ftpSessionFactory()
   {
      DefaultFtpSessionFactory sessionFactory = new DefaultFtpSessionFactory();
      sessionFactory.setHost( ftpProperties.getServers().get( 0 ).getHost() );
      sessionFactory.setPort( ftpProperties.getServers().get( 0 ).getPort() );
      sessionFactory.setUsername( ftpProperties.getServers().get( 0 ).getUser() );
      sessionFactory.setPassword( ftpProperties.getServers().get( 0 ).getPassword() );
      return sessionFactory;
   }
}

@SpringBootApplication
@EnableIntegration
@IntegrationComponentScan
public class FtpTestApplication {

    public static void main(String[] args) {
        SpringApplication.run( FtpTestApplication.class, args );
    }
}

@Configuration
@PropertySource( "classpath:ftp.properties" )
@ConfigurationProperties( prefix = "ftp" )
@Data
public class FtpProperties
{
   @NotNull
   private String localDir;

   @NotNull
   private List<Server> servers;

   @Data
   public static class Server
   {
      @NotNull
      private String host;

      @NotNull
      private int port;

      @NotNull
      private String user;

      @NotNull
      private String password;
   }
}

Контроллер в основном используется только для тестирования, вФактическая реализация есть опросник.Мой FtpProperties содержит список серверов, потому что в реальной реализации я использую DelegatingSessionFactory для выбора экземпляра на основе некоторых параметров.

Согласно комментарию Гари Рассела , я 'Я ожидаю, что неудачная загрузка будет повторена.Но если я прерву загрузку на стороне сервера (введя «Kick user» в экземпляре FileZilla), я просто получу немедленную трассировку стека и не буду повторять попытки:

org.apache.commons.net.ftp.FTPConnectionClosedException: FTP response 421 received.  Server closed connection.
[...]

Мне также нужно загрузить файлы, длякоторый я использую Ftp.outboundAdapter.В этом случае и с тем же RetryTemplate, если я прерву загрузку на стороне сервера, Spring Integration выполнит еще две попытки с задержкой в ​​5 с каждая, и только после этого регистрирует java.net.SocketException: Connection reset, все как положено.

Я попытался немного отладить и заметил, что перед первой попыткой загрузки через Ftp.outboundAdapter достигнута точка останова на RequestHandlerRetryAdvice.doInvoke().Но при загрузке через Ftp.outboundGateway эта точка останова никогда не удалась.

Есть ли проблема с моей конфигурацией, может кто-то заставить RequestHandlerRetryAdvice работать с Ftp.outboundGateway / AbstractRemoteFileOutboundGateway.Command.GET * * 1040

Answer 1

Извините за задержку;На этой неделе мы находимся на SpringOne Platform.

Проблема связана с тем, что спецификация шлюза является компонентом - шлюз заканчивается инициализацией до применения рекомендации.

Я изменил вашкод вроде этого ...

@Bean
public IntegrationFlow downloadFiles(SessionFactory<FTPFile> sessionFactory) {
    return f -> f.handle(getRemoteFile(sessionFactory), getRetryAdvice())
            .channel("gatewayDownloadsOutputChannel");
}

...

private RemoteFileOutboundGatewaySpec<FTPFile, FtpOutboundGatewaySpec> getRemoteFile(SessionFactory<FTPFile> sessionFactory) {
    return Ftp.outboundGateway(sessionFactory,
            AbstractRemoteFileOutboundGateway.Command.GET,
            "payload")
            .fileExistsMode(FileExistsMode.REPLACE)
            .localDirectoryExpression("'/tmp'")
            .autoCreateLocalDirectory(true);
}

... и это сработало.

Как правило, лучше не иметь дело со спецификациями напрямую, а просто встроить их в определение потока ...

@Bean
public IntegrationFlow downloadFiles(SessionFactory<FTPFile> sessionFactory) {
    return f -> f.handle(Ftp.outboundGateway(sessionFactory,
            AbstractRemoteFileOutboundGateway.Command.GET,
            "payload")
            .fileExistsMode(FileExistsMode.REPLACE)
            .localDirectoryExpression("'/tmp'")
            .autoCreateLocalDirectory(true), getRetryAdvice())
        .channel("gatewayDownloadsOutputChannel");
}