If you want to use common reply channel then I think this is what you are looking for.
public class RemoteProxyCalculatorService
{
public int Square(int n)
{
PollableChannel replyChannel = SendRequest(n, "squareRequestChannel");
int squareOfn = ReceiveReply(replyChannel);
return squareOfn;
}
private <T> PollableChannel SendRequest(T payload, String requestChannel)
{
UUID requestID = UUID.randomUUID();
Message<T> inputMessage = MessageBuilder.withPayload(payload)
.setCorrelationId(requestID)
.setReplyChannel(myMessageHandler.getSubscribedChannel())
.build();
// Create a Pollable channel for two things
// 1. Pollable channel is where this thread should look for reply.
QueueChannel replyQueueChannel = new QueueChannel();
// 2. Message Handler will send reply to this Pollable channel once it receives the reply using correlation Id.
myMessageHandler.add(requestID, replyQueueChannel);
MessageChannel channel = context.getBean(requestChannel, MessageChannel.class);
channel.send(inputMessage);
return replyQueueChannel;
}
@SuppressWarnings("unchecked")
private <T> T ReceiveReply(PollableChannel replyChannel)
{
Message<?> groupMessage = replyChannel.receive();
return (T) groupMessage.getPayload();
}
private ClassPathXmlApplicationContext context;
@Autowired
private MyMessageHandler myMessageHandler;
}
/**
* Message Handler
*
*/
public class MyMessageHandler implements MessageHandler
{
private final Map<Object, MessageChannel> idChannelsMap = new TreeMap<>();
private final Object lock = new Object();
private final SubscribableChannel subscribedChannel;
public MyMessageHandler(SubscribableChannel subscribedChannel)
{
this.subscribedChannel = subscribedChannel;
}
@Override
public void handleMessage(Message<?> message) throws MessagingException
{
synchronized (lock)
{
this.idChannelsMap.get(message.getHeaders().getCorrelationId()).send(message);
this.idChannelsMap.remove(message.getHeaders().getCorrelationId());
}
}
public void add(Object correlationId, MessageChannel messageChannel)
{
synchronized (lock)
{
this.idChannelsMap.put(correlationId, messageChannel);
}
}
public SubscribableChannel getSubscribedChannel()
{
return subscribedChannel;
}
}