Travel of Software Developer

8.5.13

Why and How with Message Handler Chain in Spring Integration?

5.6 Message Handler Chain <int:chain/> is not a pattern in EIP. It is a convenient usage in Spring Integration to make the configuration simpler.

The MessageHandlerChain is an implementation of MessageHandler that can be configured as a single Message Endpoint while actually delegating to a chain of other handlers, such as Filters, Transformers, Splitters, and so on. This can lead to a much simpler configuration when several handlers need to be connected in a fixed, linear progression. For example, it is fairly common to provide a Transformer before other components. Similarly, when providing a Filter before some other component in a chain, you are essentially creating a Selective Consumer. In either case, the chain only requires a single input-channel and a single output-channel eliminating the need to define channels for each individual component.

This is a good one to explain Message Handler Chain with an example.

java - Spring integration Message handler chain usage? - Stack Overflow

Understand Spring Integration's Gateway

Why am I interested in Gateway?

TechiQuest by Matt Vickery: Spring Integration - Payload Storage via Claim-check

When I was learning the Claim-check pattern using Spring Integration on the above article, I was lost at Spring Integration's Gateway.

It seemed that Spring Integration Gateway was much more than I would have expected to be: I sent out a message via the Gateway, and received a very final message through out the whole procedure.

Is that what Messaging Gateway supposed to be? Or is this just some trick by Spring Integration?

What is Messaging Gateway in EIP?

An application accesses another system via Messaging.

Enterprise Integration Patterns - Messaging Gateway

From this description, it sounds like that a Messaging Gateway is an application. It doesn't seem right here with Spring Integration. But maybe it was right at the time EIP was developed.

How do you encapsulate access to the messaging system from the rest of the application.

So, this is a question for the reader, me here. Encapsulation means we don't want to expose the rest of the system to the underlying Messaging implementation. That makes sense to a Gateway.

Use a Messaging Gateway, a class that wraps messaging-specific method calls and exposes domain-specific methods to the application.

Ok, I know this is the answer for the above question. In a word, Messaging Gateway is a class for Messaging system encapsulation.

The Messaging Gateway encapsulates messaging-specific code (e.g., the code required to send or receive a message) and separates it from the rest of the application code. This way, only the Messaging Gateway code knows about the messaging systems; the rest of the application code does not.
The Messaging Gateway exposes a business function to the rest of the application so that instead of requiring the application to set properties like Message.MessageReadPropertyFilter.AppSpecific, a Messaging Gateway exposes methods such as GetCreditScore that accept strongly typed parameters just like any other method.
A Message Gateway is a messing-specific version of the more general Gateway pattern.

Still, no more than the literal meaning of "Gateway".

Let's take a look at the combination of Request-Reply pattern and Messaging Gateway pattern:

Many Messaging Gateways send a message to another component and expect a reply message (see Request-Reply [154]). Such a Messaging Gateway can be implemented in two different ways:

Blocking (Synchronous) Messaging Gateway.
Event-Driven (Asynchronous) Messaging Gateway

In the Claim-Check example, the Messaging Gateway works as a blocking Messaging Gateway.

With all these information, I think what Spring Integration implements about Messaging Gateway fits the definition in EIP very well. But I still need to look closer into Spring Integration's implementation to understand more about its configurations.

@Gateway and <int:gateway/> in Spring Integration

There are many Gateway Things in Spring Integration, such as JDBC Outbound Gateways, JMS Inbound and Outbound Gateways, Web Service Inbound and Outbound Gateways, etc. What I am really interested in this moment is the @Gateway annotation, and <int:gateway/> namespace, which is defined as Messaging Gateways (7. Messaging Endpoints), while previously named Inbound Messaging Gateways (16. Inbound Messaging Gateways).

I don't have time to find out the change history. But Messaging Gateways makes more sense to me than Inbound Messaging Gateways, mostly because I have problem to tell the directions, I guess.

default-request-channel and request-channel

According to the above reading with EIP, Spring Integration's Messaging Gateways are Request-Reply Messaging Gateways. In that case, I could easily understand the request-channel and reply-channel, which are part of its XML Namespace Support or @Gateway annotation.

However, things are tricky with request-channel. Literally, the request-channel is the channel that the Messaging Endpoint, which is Messaging Gateway here, receives requests with. However, with Messaging Gateway, the interface, will immediately delegate those request to the next hop, which is the first component on the other side, the Messaging System in this case.

public interface ClaimCheckGateway {

public static final String CLAIM_CHECK_ID = "ClaimCheckID";

@Gateway(requestChannel = "claim-check-in-channel")

public Message<String> send(Message<String> message);

}

<int:gateway id="claimCheckGateway" service-interface="simple.demo.springintegration.demo.chapter5.ClaimCheckGateway"/>

<int:chain input-channel="claim-check-in-channel" output-channel="processing-channel">

<int:claim-check-in message-store="simpleMessageStore"/>

<int:header-enricher>

<int:header

name="#{T(simple.demo.springintegration.demo.chapter5.ClaimCheckGateway).CLAIM_CHECK_ID}"

expression="payload"/>

</int:header-enricher>

</int:chain>

From the above example, the requestChannel in ClaimCheckGateway is exactly the same channel for the first <int:chain/> component. It confused me in the first place, because I would have expected the input-channel of <int:chain/> was some output-channel from the previous component.

So as the reply-channel.

In the Claim-check example, no reply-channel was assigned to the Gateway, neither to the last component, which was the last <int:chain/> with "claim-check-out-channel" as its input-channel. Here is the explanation:

7. Messaging Endpoints

Typically you don't have to specify the default-reply-channel, since a Gateway will auto-create a temporary, anonymous reply channel, where it will listen for the reply.

A Gateway will create a temporary point-to-point reply channel which is anonymous and is added to the Message Headers with the name replyChannel. When providing an explicit default-reply-channel (reply-channel with remote adapter gateways), you have the option to point to a publish-subscribe channel, which is so named because you can add more than one subscriber to it. Internally Spring Integration will create a Bridge between the temporary replyChannel and the explicitly defined default-reply-channel.

Now I think I begin to understand how Spring Integration's Gateway works. But I still want to take a look at how these things are implemented.

Implementation of Messaging Gateway in Spring Integration

The XML Namespace Parser

@Override
protected String getBeanClassName(Element element) {
return IntegrationNamespaceUtils.BASE_PACKAGE + ".gateway.GatewayProxyFactoryBean";
}

There are other important information for this XML namespace, but I am good with this at this moment.

GatewayProxyFactoryBean was mentioned a couple of time in Spring Integration's documentation.

Same as other FactoryBeans, here is the entry point I would take a look at:

public Object getObject() throws Exception {
if (this.serviceProxy == null) {
this.onInit();
Assert.notNull(this.serviceProxy, "failed to initialize proxy");
}
return this.serviceProxy;
}

And in turn:

@Override
protected void onInit() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;
}
BeanFactory beanFactory = this.getBeanFactory();
if (this.channelResolver == null && beanFactory != null) {
this.channelResolver = new BeanFactoryChannelResolver(beanFactory);
}
Class<?> proxyInterface = this.determineServiceInterface();
Method[] methods = ReflectionUtils.getAllDeclaredMethods(proxyInterface);
for (Method method : methods) {
MethodInvocationGateway gateway = this.createGatewayForMethod(method);
this.gatewayMap.put(method, gateway);
}
this.serviceProxy = new ProxyFactory(proxyInterface, this).getProxy(this.beanClassLoader);
this.start();
this.initialized = true;
}
}

In method createGatewayForMethod(...):

MethodInvocationGateway gateway = new MethodInvocationGateway(messageMapper);
gateway.setErrorChannel(this.errorChannel);
if (this.getTaskScheduler() != null) {
gateway.setTaskScheduler(this.getTaskScheduler());
}
gateway.setBeanName(this.getComponentName());
gateway.setRequestChannel(requestChannel);
gateway.setReplyChannel(replyChannel);
if (requestTimeout == null) {
gateway.setRequestTimeout(-1);
}
else {
gateway.setRequestTimeout(requestTimeout);
}
if (replyTimeout == null) {
gateway.setReplyTimeout(-1);
}
else {
gateway.setReplyTimeout(replyTimeout);
}
if (this.getBeanFactory() != null) {
gateway.setBeanFactory(this.getBeanFactory());
}
if (this.shouldTrack) {
gateway.setShouldTrack(this.shouldTrack);
}
gateway.afterPropertiesSet();
return gateway;

The hierarchy of Messaging Gateway implementation

And here is the implementation of send() method:

protected void send(Object object) {

this.initializeIfNecessary();

Assert.notNull(object, "request must not be null");

Assert.state(this.requestChannel != null,

"send is not supported, because no request channel has been configured");

try {

this.messagingTemplate.convertAndSend(this.requestChannel, object, this.historyWritingPostProcessor);

}

catch (Exception e) {

if (this.errorChannel != null) {

this.messagingTemplate.send(this.errorChannel, new ErrorMessage(e));

}

else {

this.rethrow(e, "failed to send message");

}

This matches the idea that the requestChannel is delegated directly to the following messaging component.

The doSendAndReceive() method is more important. But it is too complex to list here.

7.5.13

A very simple example to use JDOM along with TagSoup

Just Read and Build the Document

SAXBuilder builder = new org.jdom.input.SAXBuilder("org.ccil.cowan.tagsoup.Parser");
Reader in = new StringReader(pageContent);
org.jdom.Document doc = builder.build(in);
System.out.println(new XMLOutputter().outputString(doc));

Applying with XPath

SAXBuilder builder = new org.jdom.input.SAXBuilder("org.ccil.cowan.tagsoup.Parser");

Reader in = new StringReader(pageContent);

org.jdom.Document doc = builder.build(in);

//System.out.println(new XMLOutputter().outputString(doc));

XPath xpath = XPath.newInstance("//xhtml:div[@class='views-row views-row-3 views-row-odd']");

xpath.addNamespace("xhtml", "http://www.w3.org/1999/xhtml");

List<Element> nodes = xpath.selectNodes(doc);

for(Element el : nodes) {

System.out.println(new XMLOutputter().outputString(el));

}

Applying XPath with Sub-Element

I am not sure why, but it seems that JDom has problems to handle Sub-elements. You have to build a new document to apply with the XPath.

SAXBuilder builder = new org.jdom.input.SAXBuilder("org.ccil.cowan.tagsoup.Parser");

Reader in = new StringReader(pageContent);

org.jdom.Document doc = builder.build(in);

//System.out.println(new XMLOutputter().outputString(doc));

XPath xpath = XPath.newInstance("//xhtml:div[@class='node node-teaser node-article']");

xpath.addNamespace("xhtml", "http://www.w3.org/1999/xhtml");

List<Element> nodes = xpath.selectNodes(doc);

XPath xpathArticle = XPath.newInstance("//xhtml:a[@class='node-title']");

xpathArticle.addNamespace("xhtml", "http://www.w3.org/1999/xhtml");

XMLOutputter xmlOutputter = new XMLOutputter();

for(Element el : nodes) {

String elXml = xmlOutputter.outputString(el);

builder = new org.jdom.input.SAXBuilder();

in = new StringReader(elXml);

org.jdom.Document doc2 = builder.build(in);

Element result = (Element)xpathArticle.selectSingleNode(doc2);

System.out.println(xmlOutputter.outputString(result));

}

How to find out the implementation of Spring Integration's XML Namespace?

When I was reading the documents or books about Spring Integration, I always wanted to find out the underlying implementation, which is the easiest way for me to understand how it works and how it doesn't.

For example,

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

<int:poller fixed-rate="5000"/>

What is the underlying support Java class?

First of all, I need to know how to define namespace in Spring.

http://mozhenghua.iteye.com/blog/1830842 (Chinese link: spring自定义标签实现)

From this article, I learnt that there are two properties files that are very important to me. And one of them is the entry for me to locate the class I want to know:

* spring.handlers

For Spring Integration (core), this is the content of the file:

http\://www.springframework.org/schema/integration=org.springframework.integration.config.xml.IntegrationNamespaceHandler

In turn, I found out this:

registerBeanDefinitionParser("channel", new PointToPointChannelParser());

registerBeanDefinitionParser("poller", new PollerParser());

A Simple Sample Program for Spring's TaskExecutor

references:

@Test
public void test() throws Exception {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(5);
executor.setMaxPoolSize(10);
executor.setQueueCapacity(25);

executor.initialize();

Runnable task = new Runnable() {

@Override
public void run() {
System.out.printf("Start thread : %d\n", Thread.currentThread().getId());
try {
TimeUnit.SECONDS.sleep(new Random().nextInt(10));
} catch (InterruptedException ex) {
//..
}

System.out.printf("End thread id = %d\n", Thread.currentThread().getId());
}
};

for(int k=0; k<100; ++k)
{
try {
executor.execute(task);
} catch (TaskRejectedException ex) {
System.out.println("Thread rejected. Wait...");
TimeUnit.SECONDS.sleep(1);
}
}
}

1.5.13

How do you find what process is holding a file open in Windows?

ntfs - How do you find what process is holding a file open in Windows? - Server Fault

I've had success with Sysinternals Process Explorer. With this, you can search to find what process(es) have a file open, and you can use it to close the handle(s) if you want. Of course, it is safer to close the whole process. Exercise caution and judgement.

To find a specific file, use the menu option "Find->Find Handle or DLL..." Type in part of the path to the file. The list of processes will appear below.

edited Dec 31 '12 at 17:46