spring_integration_sample

Bootstrap.java

package com.example;

import java.util.concurrent.TimeUnit;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.message.GenericMessage;
import org.springframework.scheduling.support.PeriodicTrigger;

public class Bootstrap {
	 
    public static void main(String[] args) {
    	
    	ApplicationContext context = new ClassPathXmlApplicationContext("config.xml");
     
    	// 1. Create a pollable channel
        PollableChannel channel = (PollableChannel) context.getBean(PollableChannel.class);
        MessageHandler subscribingHandlerA = new SimpleMessageHandler("A");
   
        // Create a polling consumer
        PollingConsumer consumer = new PollingConsumer(channel, subscribingHandlerA);        
        PeriodicTrigger trigger = new PeriodicTrigger(10, TimeUnit.MILLISECONDS);
        consumer.setTrigger(trigger);
        consumer.setMaxMessagesPerPoll(10);
        consumer.setBeanFactory(context);    
       
        // Start the consumer.
        consumer.start();
        
        // Send a message. The consumer should be notified and use the MessageHandler
        // to handle the message. For us, that's SimpleMessageHandler.handleMessage()
        channel.send(new GenericMessage<String>("Message on the pollable channel"));
        
        consumer.stop();
        
        // 2. Create a subscribable channel.
        SubscribableChannel subscribableChannel = (SubscribableChannel) context.getBean(SubscribableChannel.class);
                
        // Set up two message handlers.
        MessageHandler subscribingHandlerB = new SimpleMessageHandler("B");
        MessageHandler subscribingHandlerC = new SimpleMessageHandler("C");
        
        // Subscribe the message handlers to the channel.
        subscribableChannel.subscribe(subscribingHandlerB);
        subscribableChannel.subscribe(subscribingHandlerC);
        
        // Send a message. The message handlers should be notified.
        subscribableChannel.send(new GenericMessage<String>("Message on the subscribable channel"));  
        
    }
}

config.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:si="http://www.springframework.org/schema/integration"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
        					http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
							http://www.springframework.org/schema/integration 
							http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
							http://www.springframework.org/schema/context
        					http://www.springframework.org/schema/context/spring-context-3.0.xsd">
 

	<context:annotation-config />
	<bean class="com.example.TestAppConfig"/>	
</beans>

SimpleMessageHandler.java

import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;

public class SimpleMessageHandler implements MessageHandler {

	String id;
	
	public SimpleMessageHandler(String id) {
		this.id = id;
	}
	
	public void handleMessage(Message<?> message) throws MessagingException {
		System.out.println("Handler " + id + "; " + message);
	}
}

TestAppConfig.java

@Configuration
public class TestAppConfig {

	@Bean
	public PollableChannel defaultChannel() {
		return new QueueChannel(10);
	}
	
	@Bean
	public SubscribableChannel subscribeChannel() {
		return new PublishSubscribeChannel();
	}
	
	/**
	 * A TaskScheduler has to be created for each endpoint or one has to be defined
	 * in the integration context.
	 * 
	 */
	@Bean
	public TaskScheduler taskScheduler() {
        return new ThreadPoolTaskScheduler();