1. intro
In shorts, Spring integration is a framework wich can help in creation application with complex data flows, with a lot of different data transformation, filtering, routing and others.
Schema of flow in this example :
Description :
we have collection of payments, this collection is splitting and sending "one by one" into next step : Filter. Filter just allows to go ahead payments with amount!=nll. After that(router), we have 2 types of workflow regarding property isVip: one workflow for regular payments(regular transformer) , another one for vip payments : vip transformer. Results of transformation are sending into paymentProcessor in which has to main logic of our application. In case when we have errors in processing - we are processing errors in fail processor.
2. Component description.
- Inbound Gateways ---
bring data from an external system to the integration network, and expect
a response message to be provided by some other components, through some
channel, to be forward to the external system
- Channels --- the connectors between the other types of components, i.e. the
endpoints. (These are the "pipes" in the
"pipes&filters" architectural characterization.) Channels provide
a simple API defined in-terms of message passing operations, such as send() and recv() methods.
- message router is an endpoint that
forwards incoming messages to one (or more) among several configured output
channels. The criteria for selecting the output channel is usually based on the
message payload and/or the message headers.
Message filters are endpoints that selectively relay messages from
an input to an output channel. They can be used to discard messages based on
payload or header values, detect invalid messages, or relay non-conforming
messages to a separte channel.
Message transformers are endpoints that transform
messages payload and/or headers. The can be used to perform arbitrary
transformations on the payload, convert payload types, and add, change, or
remove headers.
The Splitter is a component whose role is to partition a message in several parts, and send the resulting messages to be processed independently. Very often, they are upstream producers in a pipeline that includes an Aggregator.
3. Project structure
4. Main project (pom) file
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.demien</groupId>
<artifactId>spring-integration-demo</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>3.2.1.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
<version>2.2.2.RELEASE</version>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.16</version>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>13.0.1</version>
</dependency>
<dependency>
<groupId>org.testng</groupId>
<artifactId>testng</artifactId>
<version>6.5.2</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.6</source>
<target>1.6</target>
</configuration>
</plugin>
</plugins>
</build>
</project>
5. Context file
Here described all component of application. Please take a look how implemented logging: interceptors in channels.
<?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:task="http://www.springframework.org/schema/task"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/task
http://www.springframework.org/schema/task/spring-task.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd"
>
<bean id="auditInterceptor" class="com.demien.spring.integration.interceptor.LogInterceptor"/>
<int:channel id = "newPaymentChannel">
<int:interceptors>
<ref bean="auditInterceptor"/>
</int:interceptors>
</int:channel>
<int:gateway id="inPaymentGateway"
service-interface="com.demien.spring.integration.gateways.LoadPaymentsGateway">
<int:method name="loadPayments" request-channel="newPaymentChannel" />
</int:gateway>
<int:splitter
input-channel="newPaymentChannel"
output-channel="singlePaymentChannel" />
<int:channel id = "singlePaymentChannel">
<int:interceptors>
<ref bean="auditInterceptor"/>
</int:interceptors>
</int:channel>
<int:filter
input-channel="singlePaymentChannel"
output-channel="filteredPaymentChannel"
ref="paymentFilter" />
<int:channel id = "filteredPaymentChannel">
<int:interceptors>
<ref bean="auditInterceptor"/>
</int:interceptors>
</int:channel>
<bean id="paymentFilter" class="com.demien.spring.integration.filters.PaymentFilter"/>
<int:recipient-list-router input-channel="filteredPaymentChannel">
<int:recipient channel = "regularPayments" selector-expression="!payload.isVip()"/>
<int:recipient channel = "vipPayments" selector-expression="payload.isVip()" />
</int:recipient-list-router>
<int:channel id = "regularPayments">
<int:interceptors>
<ref bean="auditInterceptor"/>
</int:interceptors>
</int:channel>
<int:channel id = "vipPayments">
<int:interceptors>
<ref bean="auditInterceptor"/>
</int:interceptors>
</int:channel>
<int:transformer
input-channel="regularPayments"
output-channel="processingChannel"
ref="regularPaymentTransformer" />
<int:transformer
input-channel="vipPayments"
output-channel="processingChannel"
ref="vipPaymentTransformer" />
<int:channel id = "processingChannel">
<int:queue capacity="10" />
<int:interceptors>
<ref bean="auditInterceptor"/>
</int:interceptors>
</int:channel>
<bean id="regularPaymentTransformer" class="com.demien.spring.integration.transformers.RegularPaymentTransformer"/>
<bean id="vipPaymentTransformer" class="com.demien.spring.integration.transformers.VipPaymentTransformer"/>
<int:service-activator input-channel="processingChannel" ref="paymentProcessor">
<int:poller fixed-rate="100" error-channel="failedPaymentsChannel" />
</int:service-activator>
<bean id="paymentProcessor" class="com.demien.spring.integration.activators.PaymentServiceActivator"/>
<int:channel id = "failedPaymentsChannel">
<int:interceptors>
<ref bean="auditInterceptor"/>
</int:interceptors>
</int:channel>
<int:service-activator input-channel="failedPaymentsChannel" ref="failProcessor" />
<bean id="failProcessor" class="com.demien.spring.integration.activators.FailedPaymentActivator"/>
</beans>
6. Test dto(Payment)
Very simple object. Please take a look at method isVip- we will use it later.
package com.demien.spring.integration.dto;
import java.math.BigDecimal;
import java.util.Date;
public class Payment {
BigDecimal amount;
String description;
public Payment(String description, BigDecimal amount) {
this.description = description;
this.amount = amount;
}
public boolean isVip() {
if (amount!=null && amount.compareTo(new BigDecimal(10000))==1) {
return true;
} else {
return false;
}
}
public BigDecimal getAmount() {
return amount;
}
public void setAmount(BigDecimal amount) {
this.amount = amount;
}
public String getDescription() {
return description;
}
public void setDescription(String description) {
this.description = description;
}
@Override
public String toString() {
return "Payment{" +
"amount=" + amount +
", description='" + description + '\'' +
'}';
}
}
7. Interceptors
As I mentioned before - we will use interceptor for logging messages which came into channel.
package com.demien.spring.integration.interceptor;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter;
public class LogInterceptor extends ChannelInterceptorAdapter {
public Message<?> preSend(Message<?> message,
MessageChannel channel) {
System.out.println("[["+channel.toString()+"]] "+message.getPayload());
return message;
}
}
8. Filters
We can filter object which came into channel. In this example, if received object don't have amount - it will be filtered (and will not move forward).
package com.demien.spring.integration.filters;
import com.demien.spring.integration.dto.Payment;
import org.springframework.integration.annotation.Filter;
public class PaymentFilter {
@Filter
public boolean checkMandatoryFields(Payment payment) {
if (payment==null || payment.getAmount()==null) {
System.out.println("REJECTED:"+payment);
return false;
} else {
return true;
}
}
}
9. Transformers
I made 2 branches of workflow here : 2 transformers : for regular and vip payments.
package com.demien.spring.integration.transformers;
import com.demien.spring.integration.dto.Payment;
public interface PaymentTransformer {
String paymentToSting(Payment payment);
}
package com.demien.spring.integration.transformers;
import com.demien.spring.integration.dto.Payment;
import org.springframework.integration.annotation.Transformer;
public class RegularPaymentTransformer implements PaymentTransformer {
@Override
@Transformer
public String paymentToSting(Payment payment) {
return payment.getDescription()+" "+payment.getAmount();
}
}
package com.demien.spring.integration.transformers;
import com.demien.spring.integration.dto.Payment;
import org.springframework.integration.annotation.Transformer;
public class VipPaymentTransformer implements PaymentTransformer {
@Override
@Transformer
public String paymentToSting(Payment payment) {
return "!!! "+ " VIP PAYMENT:"+payment.getDescription()+" "+payment.getAmount();
}
}
10. Activators
Usually, the main processing logic. In this example I just printed in-object.
package com.demien.spring.integration.activators;
import org.springframework.integration.annotation.ServiceActivator;
public class PaymentServiceActivator {
@ServiceActivator
public void processPayment(String payment) throws Exception {
if (payment.toUpperCase().contains("TEST")) {
throw new Exception("Test payment was not processed");
}
// some logic have to be here
System.out.println("PROCESSED:"+payment);
}
}
package com.demien.spring.integration.activators;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.annotation.ServiceActivator;
public class FailedPaymentActivator {
@ServiceActivator
public void handleFailedOrder(Message<MessageHandlingException> message) {
String payment=(String)message.getPayload().getFailedMessage().getPayload();
System.out.println("FAILED:"+payment+" WITH ERROR:"+message.getPayload().getMessage());
}
}
11. Main starter file
Here I'm creating 4 test objects and sending them into workflow by gateway.
package com.demien.spring.integration;
import com.demien.spring.integration.dto.Payment;
import com.demien.spring.integration.gateways.LoadPaymentsGateway;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
public class App {
static LoadPaymentsGateway gateway;
public static void main(String[] args) {
ClassPathXmlApplicationContext context=new ClassPathXmlApplicationContext("/expences-context.xml");
Payment payment1=new Payment("First payment", new BigDecimal(666) );
Payment payment2=new Payment("Second payment", new BigDecimal(20000) );
Payment payment3=new Payment("Null payment", null );
Payment payment4=new Payment("Test payment", new BigDecimal(1) );
List<Payment> payments=new ArrayList<Payment>();
payments.add(payment1);
payments.add(payment2);
payments.add(payment3);
payments.add(payment4);
gateway=(LoadPaymentsGateway)context.getBean("inPaymentGateway");
gateway.loadPayments(payments);
context.close();
}
}
12. Results
I ordered strings in log in order of processing. So we can see what newPaymentChannel received collection of 4 objects. After that they were sent one-by-one into singlePaymentChannel. Filtered payment channel got only 3 objects : one was rejected because of null amount. Later based on value of amount (and property isVip) 2objects were sent into regularPayments and 1 - vipPayments, and transformed by corresponding transformers into another objects(into Strings in this example). At the end transformed objects were processed by PaymentServiceActivators(processingChannel). Rejected objects are moving forward to failedPaymentChannel and processed by FailedServiceActivator.
[[newPaymentChannel]] [Payment{amount=666, description='First payment'}, Payment{amount=20000, description='Second payment'}, Payment{amount=null, description='Null payment'}, Payment{amount=1, description='Test payment'}]
[[singlePaymentChannel]] Payment{amount=666, description='First payment'}
[[singlePaymentChannel]] Payment{amount=20000, description='Second payment'}
[[singlePaymentChannel]] Payment{amount=null, description='Null payment'}
[[singlePaymentChannel]] Payment{amount=1, description='Test payment'}
[[filteredPaymentChannel]] Payment{amount=666, description='First payment'}
[[filteredPaymentChannel]] Payment{amount=20000, description='Second payment'}
[[filteredPaymentChannel]] Payment{amount=1, description='Test payment'}
REJECTED:Payment{amount=null, description='Null payment'}
[[regularPayments]] Payment{amount=666, description='First payment'}
[[regularPayments]] Payment{amount=1, description='Test payment'}
[[vipPayments]] Payment{amount=20000, description='Second payment'}
[[processingChannel]] First payment 666
PROCESSED:First payment 666
[[processingChannel]] !!! VIP PAYMENT:Second payment 20000
PROCESSED:!!! VIP PAYMENT:Second payment 20000
[[processingChannel]] Test payment 1
FAILED:Test payment 1 WITH ERROR:java.lang.Exception: Test payment was not processed
[[failedPaymentsChannel]] org.springframework.integration.MessageHandlingException: java.lang.Exception: Test payment was not processed
13. The end
Source code can be downloaded from
here.
