Lancer le travail par lots à partir de Spring Integration

Question

J'ai besoin de télécharger un fichier depuis le serveur SFTP distant et de le traiter à l'aide du lot de printemps. J'ai déjà implémenté du code en utilisant Spring Integration pour télécharger des fichiers. Mais je ne peux pas lancer le travail Spring Batch à partir de composants d'intégration de printemps. J'ai le code suivant:

@Autowired 
private JobLauncher jobLauncher; 

public String OUTPUT_DIR = "temp_dir"; 

@Value("${sftp.remote.host}") 
private String sftpRemoteHost; 

@Value("${sftp.remote.user}") 
private String sftpUsername; 

@Value("${sftp.remote.password}") 
private String sftpPassword; 

@Value("${sftp.remote.folder}") 
private String sftpFolder; 

@Bean 
public DefaultSftpSessionFactory sftpSessionFactory() { 
    final DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(); 
    factory.setHost(sftpRemoteHost); 
    factory.setAllowUnknownKeys(true); 
    factory.setUser(sftpUsername); 
    factory.setPassword(sftpPassword); 
    return factory; 
} 

@Bean 
public SftpInboundFileSynchronizer sftpInboundFileSynchronizer() { 
    final SftpInboundFileSynchronizer fileSynchronizer = new SftpInboundFileSynchronizer(sftpSessionFactory()); 
    fileSynchronizer.setDeleteRemoteFiles(false); 
    fileSynchronizer.setRemoteDirectory(sftpFolder); 
    fileSynchronizer.setFilter(new SftpSimplePatternFileListFilter("*.csv")); 
    return fileSynchronizer; 
} 

@Bean 
@InboundChannelAdapter(channel = "sftpChannel", poller = @Poller(fixedDelay = "5000")) 
public MessageSource<File> sftpMessageSource() { 
    final SftpInboundFileSynchronizingMessageSource source = 
      new SftpInboundFileSynchronizingMessageSource(sftpInboundFileSynchronizer()); 
    source.setLocalDirectory(new File(OUTPUT_DIR)); 
    source.setAutoCreateLocalDirectory(true); 
    source.setLocalFilter(new AcceptOnceFileListFilter<>()); 
    return source; 
} 

@Bean 
@ServiceActivator(inputChannel = "sftpChannel") 
public MessageHandler handler() { 
    final FileWritingMessageHandler handler = new FileWritingMessageHandler(new File(OUTPUT_DIR)); 
    handler.setFileExistsMode(FileExistsMode.REPLACE); 
    handler.setExpectReply(true); 
    handler.setOutputChannelName("parse-csv-channel"); 
    return handler; 
} 

@ServiceActivator(inputChannel = "parse-csv-channel", outputChannel = "job-channel") 
public JobLaunchRequest adapt(final File file) throws Exception { 
    final JobParameters jobParameters = new JobParametersBuilder().addString(
      "input.file", file.getAbsolutePath()).toJobParameters(); 
    return new JobLaunchRequest(batchConfiguration.job(), jobParameters); 
} 

@ServiceActivator(inputChannel = "job-channel", outputChannel = "finish") 
public JobLaunchingMessageHandler jobHandler(JobLaunchRequest request) throws JobExecutionException { 
    return new JobLaunchingMessageHandler(jobLauncher);//.launch(request); 
} 

@ServiceActivator(inputChannel = "finish") 
public void finish() { 
    System.out.println("FINISH"); 
} 

Mais cela ne fonctionne pas (erreur dans la dernière méthode adapt), parce que non haricot de fichier de type trouvé. Je ne peux pas comprendre ces deux parties ensemble. Comment câbler l'intégration et le traitement par lots?

Answer 1

Vous devez simplement supprimer l'annotation @Bean de votre méthode adapt(). Nous avons besoin @Bean si nous construisons vraiment MessageHandler haricot, par exemple JobLaunchingMessageHandler pour accepter que JobLaunchRequest charge utile: https://docs.spring.io/spring-batch/trunk/reference/html/springBatchIntegration.html#launching-batch-jobs-through-messages.

Voir plus d'informations sur la messagerie annotations dans le Manuel de référence: https://docs.spring.io/spring-integration/docs/4.3.12.RELEASE/reference/html/configuration.html#annotations_on_beans

MISE À JOUR

@Bean 
@ServiceActivator(inputChannel = "sftpChannel") 
public MessageHandler handler() { 
    final FileWritingMessageHandler handler = new FileWritingMessageHandler(new File(OUTPUT_DIR)); 
    handler.setFileExistsMode(FileExistsMode.REPLACE); 
    handler.setExpectReply(true); 
    handler.setOutputChannelName("parse-csv-channel"); 
    return handler; 
} 

@ServiceActivator(inputChannel = "parse-csv-channel", outputChannel = "job-channel") 
public JobLaunchRequest adapt(final File file) throws Exception { 
    final JobParameters jobParameters = new JobParametersBuilder().addString(
      "input.file", file.getAbsolutePath()).toJobParameters(); 
    return new JobLaunchRequest(batchConfiguration.job(), jobParameters); 
} 

@Bean 
@ServiceActivator(inputChannel = "job-channel") 
public JobLaunchingGateway jobHandler() { 
    JobLaunchingGateway jobLaunchingGateway = new JobLaunchingGateway(jobLauncher); 
    jobLaunchingGateway.setOutputChannelName("finish"); 
    return jobLaunchingGateway; 
}