Home » Java » Enterprise Java » Aggregating async results using Spring Integration

About Idan Fridman

Idan Fridman
Idan is Software engineer with experience in Server side technologies. Idan is responsible for various infrastructure models in the software industry(Telecommunications, Finance).

Aggregating async results using Spring Integration

Hi, I came across an issue which has very nice solution using Spring Integration. Many times we have the need for a scenario of dispatching a message to unknown number of destinations. For this purpose we have the Topic methodology. But some times in addition we also want to receive answers from all destinations which received the message and aggregate them to to a single result answer. For this purpose we can use channels combined with Aggregator and ReleaseStrategy interfaces. In this post I wont concentrate on the “channels implementation”.

So let’s say we have a producer who sends it’s message to a Topic. Now we have a consumer which receives that message. Using gateway and a processor interface we can send that message in any type we want:

public interface Processor
{
    public void sendResponse(String response);
}

Consumer code:

@Override public void onMessage(Message message)
{
  String resultMessage = "";
    try
    {
          processor.sendResponse(resultMessage);
    }
    catch (Exception e)
    {
         log.error("Error while processing message in channel consumer. errorMsg=" + e.getMessage(), e);
     }
}

Now the message will be delivered to a channel(“In Channel”) We can add to this message an extra information(in case we have different message groups). After adding an extra information we dispatch that message to another channel (“Out channel”) Now here is the magic: We create two pojo’s that later will be bind to interfaces using the XML configuration. ReleaseStrategy:

public class ReleaseStrategy
{
	public boolean canRelease(List results)
	{
            // check if all 5  subscribers sent responses
		if (results.size() == 5)
                {

			return true;
		}
		return false;
	}
}

Aggregator:

public class Aggregator
{
	public String aggregate(List results)
	{
               String finalResult= "SUCCESS_RESULT";
		for (String result: results) {

			if (result.equals("ERROR_RESULT")) {
				finalResult= "ERROR_RESULT";
				break;
			}
		}

		return finalResult;
	}
}

Basically what happens here is that after we return a “true” value via the canRelease method of ReleaseStrategy Interface the Aggregator will be able to receive the aggregated message and dispatch single result to the final destination(could be a Queue where another message consumer will get and process the result) Xml configuration:

< ?xml version="1.0" encoding="UTF-8"?>

 

Do you want to know how to develop your skillset to become a Java Rockstar?

Subscribe to our newsletter to start Rocking right now!

To get you started we give you our best selling eBooks for FREE!

1. JPA Mini Book

2. JVM Troubleshooting Guide

3. JUnit Tutorial for Unit Testing

4. Java Annotations Tutorial

5. Java Interview Questions

6. Spring Interview Questions

7. Android UI Design

and many more ....

 

Leave a Reply

Your email address will not be published. Required fields are marked *

*


× 6 = forty eight

You may use these HTML tags and attributes: <a href="" title=""> <abbr title=""> <acronym title=""> <b> <blockquote cite=""> <cite> <code> <del datetime=""> <em> <i> <q cite=""> <s> <strike> <strong>

Do you want to know how to develop your skillset and become a ...

Subscribe to our newsletter to start Rocking right now!

To get you started we give you our best selling eBooks for FREE!
Get ready to Rock!
To download the books, please verify your email address by following the instructions found on the email we just sent you.

THANK YOU!

Close