Messaging - Vorteile

• Integration von Daten und Funktionen
• Kapselung
• Asynchronität
• Zuverlässigkeit
• lose Kopplung

Message

• Header
• Payload

public interface Message<T> {
    T getPayload();
    MessageHeaders getHeaders();
}

Message Headers

public final class MessageHeaders implements 
    Map<String, Object>, Serializable {
    ...
}

Object someValue = message.getHeaders().get("someKey");

CustomerId customerId = message.getHeaders()
    .get("customerId", CustomerId.class);

Message Implementation

new GenericMessage<T>(T payload);
new GenericMessage<T>(T payload,
                      Map<String, Object> headers);

ErrorMessage message = new ErrorMessage(someThrowable);
Throwable t = message.getPayload();

MessageBuilder

Message<String> message1 = MessageBuilder
        .withPayload("test")
        .setHeader("foo", "bar")
        .build();

Message<String> message2 = MessageBuilder
        .fromMessage(message1).build();

assertEquals("test", message2.getPayload());
assertEquals("bar", message2.getHeaders().get("foo"));

Message Channels

• verbindet Message Producer und Consumer

public interface MessageChannel {
    boolean send(Message message);
    boolean send(Message message, long timeout);
}

• Point-to-Point

public interface PollableChannel extends MessageChannel {
    Message<?> receive();
    Message<?> receive(long timeout);
}

• Publish/Subscribe

public interface SubscribableChannel extends 
    MessageChannel {
    boolean subscribe(MessageHandler handler);
    boolean unsubscribe(MessageHandler handler);
}

Message Channel Implementierungen

• PublishSubscribeChannel
• QueueChannel
• PriorityChannel
• RendezvousChannel
• DirectChannel
• ExecutorChannel
• Scoped Channel

Message Channel Konfiguration

<int:channel id="bspChannel"/>

<int:publish-subscribe-channel id="bspPSChannel"/>

<int:channel id="queueChannel">
    <queue capacity="25"/>
</int:channel>

Endpoints - Hintergrund

• EventDrivenConsumer
• PollingConsumer

<int:poller fixed-rate="1000"/>
<int:poller id="weekdayPoller" 
    cron="*/10 * * * * MON-FRI"/>
<int:poller id="defaultPoller" default="true" 
    max-messages-per-poll="5" fixed-rate="3000"/>

Typische Endpoints

• Service Activator
• Message Bridge
• Message Enricher
• Gateway

Service Activator

<int:service-activator input-channel="new-order-channel"
    ref="orderService" method="processNewOrder">
    <int:poller fixed-delay="1000" />
</int:service-activator>

<bean id="orderService"
    class="org.jsolutions.ex.si.OrderService" />

Message Enricher

<int:header-enricher input-channel="in"
                     output-channel="out">
    <int:priority value="HIGHEST"/>
    <int:header name="foo" value="123" />
    <int:header name="bar"
        expression="payload.toUpperCase()" />
    <int:header name="foobar" ref="aBean" />
</int:header-enricher>

Gateway

package org.cafeteria;

public interface Cafe {
    void placeOrder(Order order);
}

<int:gateway id="cafeService"
         service-interface="org.cafeteria.Cafe"
         default-request-channel="requestChannel"/>

public class NotSIAwareService {
    @Resource(name = "cafeService")
    Cafe cafeGateway;
    
    public void someMethod() {
    	Order order = constructOrderInSomeWay();
        cafeGateway.placeOrder(order);
    }
...

Scrippting Support

<int:channel id="debugLog"/>
<int-stream:stdout-channel-adapter id="stdout"
    channel="debugLog"/>

<int:service-activator id="stdoutLogger"
    input-channel="debugLog">
    <int-groovy:script>
        println '=' * 140
        headers.each { println it }
        println payload
        println '=' * 140
    </int-groovy:script>
</int:service-activator>

Message Transformation

• Messages im gewünschten Format
• Custom Transformer
• Built-In Transformer

Transformer konfigurieren

<int:transformer id="testTransformer"
    ref="testTransformerBean" method="transform"
    input-channel="inChannel"
    output-channel="outChannel"/>
    
<beans:bean id="testTransformerBean"
    class="org.foo.TestTransformer" />

Built-In Transformer

• String Transformer
• Map Transformer
• (De-)Serializing Transformer
• JSON Transformer
• XML Transformer
• XPath Transformer

Message Routing

• Router
• Filter
• Splitter
• Aggregator
• Resequencer

Router konfigurieren

<int:payload-type-router input-channel="srcChannel">
    <int:mapping type="org.foo.Invoice"
                 channel="invoiceChannel" />
    <int:mapping type="org.foo.Credit"
                 channel="creditChannel" />
</int:payload-type-router>

<int:header-value-router input-channel="productChannel"
                         header-name="sellState">
    <int:mapping value="buyable" channel="actChannel" />
    <int:mapping value="soldout" channel="oldChannel" />
</int:header-value-router>

Filter konfigurieren

<int:filter input-channel="entryQueue" ref="selector"
            output-channel="discoDancer"/>

<bean id="selector" class="org.disco.Tuersteher"/>

public class Tuersteher implements MessageSelector {
    public boolean accept(Message<?> message) {
        Person person = (Person)message.getPayload();
        return (person.getAge() >= 18);
    }
}

Integration Adapters

• AMPQ
• Feed
• File
• FTP
• GemFire
• HTTP
• TCP/UDP

• JDCB
• JMS
• Mail
• MongoDb
• Redis
• RMI
• Stream

• Twitter
• Web
  Services
• XML
• XMPP
• ...

Feed Adapter

<int-feed:inbound-channel-adapter id="feedAdapter" 
    channel="feedChannel" 
    url="http://joachim.weinbrenner.name/feed/">
    <int:poller fixed-rate="10000"
                max-messages-per-poll="100" />
</int-feed:inbound-channel-adapter>

Hinweis: Payloads der erzeugten Messages sind in diesem Fall vom Typ com.sun.syndication.feed.synd.SyndEntry

File Adapter

<file:inbound-channel-adapter id="filesIn"
    directory="file:${java.io.tmpdir}/input"
    filename-pattern="*.txt">
    <int:poller id="poller" fixed-delay="5000"/>
</file:inbound-channel-adapter>

<file:file-to-string-transformer
    input-channel="filesIn" output-channel="strings"/>

<file:outbound-channel-adapter id="filesOut"
    directory="file:${java.io.tmpdir}/output"/>

Spring Integration - Positionierung

Konkurrenz?

• Apache Camel
• Mule ESB