Main.java

package test;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.concurrent.TimeUnit;

import javax.sql.DataSource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.core.Pollers;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;


@Configuration
@IntegrationComponentScan
@ComponentScan(basePackages="test.casutilisation.cuxx")
@EnableIntegration
@PropertySource(value = {"classpath:app.properties","classpath:datasource.properties"}, ignoreResourceNotFound = true)
public class Main {

  private final static Logger LOGGER = LoggerFactory.getLogger(Main.class);

  public static void main(String[] args) throws IOException, InterruptedException {
    final ConfigurableApplicationContext ctx = new AnnotationConfigApplicationContext(Main.class);
    
    PublishSubscribeChannel channel = ctx.getBean("bridgeToOutput", PublishSubscribeChannel.class);
    String[] beanNamesForType = ctx.getBeanNamesForType(MessageHandler.class);
    for (String string : beanNamesForType) {
		System.out.println(string);
	}
    MessageHandler handler= ctx.getBean("testAdvice.logging.serviceActivator.handler", MessageHandler.class);
    LOGGER.info("Appuyez sur entrée pour quitter ...");
    while(true){
    	String line = getReader().readLine();
    	if(line.isEmpty()){
    		break;
    	}
    	if(line.equalsIgnoreCase("demarrer")){
    		channel.subscribe(handler);
    	}
    	if(line.equalsIgnoreCase("arreter")){
    		channel.unsubscribe(handler);
    	}
    }
    ctx.close();
  }

  private static BufferedReader getReader() {
    final BufferedReader result = new BufferedReader(new InputStreamReader(System.in));
    return result;
  }

  @Bean
  public static PropertySourcesPlaceholderConfigurer placeholderConfigurer() {
    final PropertySourcesPlaceholderConfigurer result = new PropertySourcesPlaceholderConfigurer();
    return result;
  }
  
  @Bean(destroyMethod="close")
  public static DataSource datasource(@Value("${datasource.properties:src/main/resources/datasource.properties}")  String propertiesFile){
    HikariConfig config = new HikariConfig(propertiesFile);
    HikariDataSource result = new HikariDataSource(config);
    return result;
  }
  
}

Test.java

package test.casutilisation.cuxx;

import java.io.File;
import java.util.concurrent.TimeUnit;

import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.BridgeFrom;
import org.springframework.integration.annotation.BridgeTo;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.stereotype.Component;

@Component
public class TestAdvice {
	
	@Bean
	public PollableChannel bridgeFromInput() {
		return new QueueChannel();
	}

	@Bean
	@BridgeFrom(value = "bridgeFromInput", poller = @Poller(fixedDelay = "1000"))
	public MessageChannel bridgeFromOutput() {
		return new DirectChannel();
	}
	@Bean
	public PublishSubscribeChannel bridgeToOutput() {
		PublishSubscribeChannel result = new PublishSubscribeChannel();
		result.setMinSubscribers(1);
		result.setMinSubscribers(1);
		return result;
	}

	@Bean
	@BridgeTo("bridgeToOutput")
	public MessageChannel bridgeToInput() {
		return new DirectChannel();
	}

	
    @Bean
    @InboundChannelAdapter(value = "bridgeFromInput", poller = @Poller(fixedRate="5000"),
        autoStartup = "true")
    public MessageSource<File> input() {
    	FileReadingMessageSource result = new FileReadingMessageSource();
    	result.setDirectory(new File("src/main/java/test/casutilisation/cu01"));
		return result;
    }
    
    @Bean
    @ServiceActivator(inputChannel="bridgeToOutput")
    public MessageHandler logging(){
    	return new MessageHandler() {
			
			@Override
			public void handleMessage(Message<?> message) throws MessagingException {
				System.out.println("DEBUT DU TRAITEMENT DU HANDLER");
				try {
					//Simuler une opération lente
					for (int i = 0; i < 5; i++) {
						System.out.println("...");
						TimeUnit.SECONDS.sleep(1);
					}
				} catch (InterruptedException e) {
					throw new MessagingException(e.getMessage(),e);
				}
				System.out.println("Fin du traitement de "+message.getPayload());
			}
		};
    }
}