RequestReplyGateway.java

/*
 * Copyright 2002-2010 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.generic.amqp;

import java.io.UnsupportedEncodingException;
import java.util.UUID;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.support.RabbitGatewaySupport;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;

public class RequestReplyGateway extends RabbitGatewaySupport {

  @Autowired @Qualifier("requestQueue") private Queue requestQueue;
  @Autowired @Qualifier("responseQueue") private Queue responseQueue;

  private String defaultReplyTo;
	
  public void setDefaultReplyTo(String defaultReplyTo) {
    this.defaultReplyTo = defaultReplyTo;
  }

  //public void send(RequestDelegate requestDelegate) {
  public Object send(RequestDelegate requestDelegate) {
    //getRabbitTemplate().convertAndSend(requestDelegate, new MessagePostProcessor() {
    return getRabbitTemplate().convertSendAndReceive(
      requestQueue.getName(), // for default exchange, the queue name is same as routingKey
      requestDelegate,
      new MessagePostProcessor() {
        public Message postProcessMessage(Message message) throws AmqpException {
          message.getMessageProperties().setReplyTo(defaultReplyTo);
          try {
            message.getMessageProperties().setCorrelationId(UUID.randomUUID().toString().getBytes("UTF-8"));
          }
          catch (UnsupportedEncodingException e) {
            throw new AmqpException(e);
          }
          return message;
        }
      });
  }
}