Idan Fridman

About 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


7 + = thirteen



Java Code Geeks and all content copyright © 2010-2015, Exelixis Media Ltd | Terms of Use | Privacy Policy | Contact
All trademarks and registered trademarks appearing on Java Code Geeks are the property of their respective owners.
Java is a trademark or registered trademark of Oracle Corporation in the United States and other countries.
Java Code Geeks is not connected to Oracle Corporation and is not sponsored by Oracle Corporation.
Do you want to know how to develop your skillset and 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!

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