我的情况与描述的类似这个问题。区别在于我不使用WebFlux.outboundGateway
but an 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
根据一些参数选择一个实例。
根据加里·拉塞尔的评论,我希望重试失败的下载。但是,如果我中断下载服务器端(通过在 FileZilla 实例中发出“Kick user”),我只会立即获得堆栈跟踪,而无需重试:
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
,该断点是never hit.
我的配置有问题吗,谁能帮我解决一下RequestHandlerRetryAdvice
跟...共事Ftp.outboundGateway
/AbstractRemoteFileOutboundGateway.Command.GET
?