RPCClient.java

package org.wso2.esb.rabbitmq;

import java.util.HashMap;
import java.util.Map;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;

public class RPCClient {

	static String queue = "rpc-queue";
	static String exchange = "rpc-exchange";
	static String host = "localhost";
	static String username = "guest";
	static String password = "guest";
	static int port = 5672;
	static String replyToQueue = "callback-queue";

	public static void main(String[] argv) throws Exception {

		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");
		factory.setUsername("guest");
		factory.setPassword("guest");
		factory.setPort(5672);
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		channel.queueDeclare(queue, false, false, false, null);
		channel.exchangeDeclare(exchange, "direct", true);
		channel.queueBind(queue, exchange, "");

		// The message to be sent
		String message = "<soapenv:Envelope"
				+ " xmlns:soapenv=\"http://schemas.xmlsoap.org/soap/envelope/\">\n"
				+ "<soapenv:Body>\n"
				+ "  <p:greet xmlns:p=\"http://service.rajkumar.org\">\n"
				+ "     <in>" + "IBM" + "</in>\n" + "  </p:greet>\n"
				+ "</soapenv:Body>\n" + "</soapenv:Envelope>";

		// Populate the AMQP message properties
		AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties()
				.builder();
		builder.messageId("007");
		builder.contentType("text/xml");
		builder.correlationId("1111");
		builder.replyTo(replyToQueue);
		builder.contentEncoding("UTF-8");

		// Custom user properties
		Map<String, Object> headers = new HashMap<String, Object>();
		headers.put("SOAP_ACTION", "getQuote");
		builder.headers(headers);

		// Publish the message to exchange
		channel.basicPublish(exchange, queue, builder.build(),
				message.getBytes());
	}
}