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"?>
