I created an application that calls a gateway asynchronously using Splitter / Aggregator. In my concfiguration file, I invoke the process via InvestmentMessagingGateway that proceeds on calling the splitter. Every splitted message calls a service activator in parallel and pass it inn aggregator. I placed an error channel in the InvestmentMessagingGateway and transform every failed message to pass to the aggregator as well.
I collect every successful and failed message in aggregator as a compilation for the response. But when I tried to place an exception in one or more of the messages, I get an error in my aggregator,
Reply message received but the receiving thread has already received a reply.
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="......."">
<context:component-scan base-package="com.api.investments"/>
<!--The gateway to be called in parallel-->
<gateway id="InvestmentGateway" service-interface="com.api.investments.gateways.InvestmentGateway"/>
<channel id="investmentDetailChannel"/>
<service-activator input-channel="investmentDetailChannel" ref="investmentService" method="getAccountPortfolio"/>
<!--Inbound gateway to invoke Splitter / Aggregator-->
<gateway id="InvestmentMessageGateway" service-interface="com.api.investments.gateways.InvestmentMessageGateway"
default-reply-channel="investmentAsyncReceiver" error-channel="investmentAsyncException"/>
<channel id="investmentAsyncSender"/>
<channel id="investmentAsyncReceiver"/>
<!-- Splitter for Invesment Details-->
<splitter input-channel="investmentAsyncSender" output-channel="investmentSplitChannel" id="investmentDetailsSplitter" ref="investmentComponentsSplitter" />
<channel id="investmentSplitChannel">
<queue />
</channel>
<!--Calls the Investment Gateway asynchronously using split messages ad send the response in aggregator-->
<service-activator input-channel="investmentSplitChannel" output-channel="investmentAggregateChannel" ref="investmentAsyncActivator" method="retrieveInvestmentDetailsAsync" requires-reply="true">
<poller receive-timeout="5000" task-executor="investmentExecutor" fixed-rate="50"/>
</service-activator>
<channel id="investmentAsyncException"/>
<!--Handles failed messages and pass it in aggregator-->
<transformer input-channel="investmentAsyncException" output-channel="investmentAggregateChannel" ref="invesmentErrorLogger" method="logError"/>
<!--Aggreggates successfull and failed messaged-->
<publish-subscribe-channel id="investmentAggregateChannel"/>
<aggregator input-channel="investmentAggregateChannel" output-channel="investmentAsyncReceiver" id="investmentAggregator"
ref="investmentComponentsAggregator" correlation-strategy="investmentComponentsCorrelationStrategy"
expire-groups-upon-completion="true"
send-partial-result-on-expiry="true" />
<task:executor id="investmentExecutor" pool-size="10-1000"
queue-capacity="5000"/>
</beans:beans>
I tried putting my error channel in the poller of the service activator and but the error is still the same but this time it didn't went to the aggregator. I also tried putting a mid-gateway for the service activator like this but the error became null.
<gateway id="InvestmentAsyncActivatorGateway" service-interface="com.api.investments.gateways.InvestmentAsyncActivatorGateway"
default-reply-channel="investmentAggregateChannel" error-channel="investmentAsyncException"/>
----UPDATE------
This is the transformer that handles every error message
@Component("invesmentErrorLogger")
public class InvesmentErrorLoggerImpl implements InvestmentErrorLogger {
private final Logger logger = LoggerFactory.getLogger(Application.class.getName());
/**
* handles all error messages in InvestmentMessageGateway
* Creates an error message and pass it in the aggregator channel
* @param invesmentMessageError
* @return errorMessage
*/
@Override
public Message<ErrorDetails> logError(Message<?> invesmentMessageError) {
if(invesmentMessageError.getPayload().getClass().equals(MessagingException.class)) {
MessagingException messageException = (MessagingException) invesmentMessageError.getPayload();
AccountPortfolioRequest failedMsgPayload = (AccountPortfolioRequest) messageException.getFailedMessage().getPayload();
String logError = "Exception occured in Account Number: " + failedMsgPayload.getiAccNo();
logger.error(logError);
ErrorDetails productErrorDetail = new ErrorDetails();
productErrorDetail.setCode(InvestmentAPIErrorMessages.SVC_ERR_INQACCNTPORTFOLIO);
productErrorDetail.setMessage(InvestmentAPIErrorMessages.SVC_ERR_INQACCNTPORTFOLIO_DESC + ". Problem occured in Account Number: " + failedMsgPayload.getiAccNo());
Message<ErrorDetails> errorMessage = MessageBuilder.withPayload(productErrorDetail)
.setHeaderIfAbsent(InvestmentInquiryConstants.INV_CORRELATION_STRATEGY, InvestmentInquiryConstants.INV_CORRELATION_STRATEGY_VALUE)
.build();
return errorMessage;
}
else if(invesmentMessageError.getPayload().getClass().equals(MessageDeliveryException.class)) {
MessageDeliveryException messageException = (MessageDeliveryException) invesmentMessageError.getPayload();
AccountPortfolioRequest failedMsgPayload = (AccountPortfolioRequest) messageException.getFailedMessage().getPayload();
String logError = "Exception occured in Account Number: " + failedMsgPayload.getiAccNo();
logger.error(logError);
ErrorDetails productErrorDetail = new ErrorDetails();
productErrorDetail.setCode(InvestmentAPIErrorMessages.SVC_ERR_INQACCNTPORTFOLIO);
productErrorDetail.setMessage(InvestmentAPIErrorMessages.SVC_ERR_INQACCNTPORTFOLIO_DESC + ". Problem occured in Account Number: " + failedMsgPayload.getiAccNo());
Message<ErrorDetails> errorMessage = MessageBuilder.withPayload(productErrorDetail)
.setHeaderIfAbsent(InvestmentInquiryConstants.INV_CORRELATION_STRATEGY, InvestmentInquiryConstants.INV_CORRELATION_STRATEGY_VALUE)
.build();
return errorMessage;
}
else {
Exception messageException = (Exception) invesmentMessageError.getPayload();
String logError = "Exception occured in Investment Gateway ";
logger.error(logError);
logger.equals(messageException.getMessage());
ErrorDetails productErrorDetail = new ErrorDetails();
productErrorDetail.setCode(InvestmentAPIErrorMessages.SVC_ERR_INQACCNTPORTFOLIO);
productErrorDetail.setMessage(InvestmentAPIErrorMessages.SVC_ERR_INQACCNTPORTFOLIO_DESC + " " + messageException.getMessage());
Message<ErrorDetails> errorMessage = MessageBuilder.withPayload(productErrorDetail)
.setHeaderIfAbsent(InvestmentInquiryConstants.INV_CORRELATION_STRATEGY, InvestmentInquiryConstants.INV_CORRELATION_STRATEGY_VALUE)
.build();
return errorMessage;
}
}
}
