package omq.common.message.request;

import java.util.Map;
import java.util.Properties;
import java.util.Vector;

import omq.Remote;
import omq.common.event.Event;
import omq.common.message.response.Response;
import omq.common.util.ParameterQueue;
import omq.common.util.Serializer;
import omq.server.remote.request.RequestListener;


import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;

/**
 * This class is used to make Asynchronous requests
 * 
 * @author Sergi Toda <sergi.toda@estudiants.urv.cat>
 * 
 */
public class AsyncRequest extends Request {
	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	private String topic;

	public AsyncRequest() { 
		super();
	}

	/**
	 * This is the constructor of an asynchronous request
	 * 
	 * @param uid
	 *            This represents the real object uid registered in the server
	 * @param corrId
	 *            This is the correlation id of this method request
	 * @param methodName
	 *            The name of the method to send
	 * @param args
	 *            The arguments of this method
	 * @param topic
	 *            Since an asynchronous method cannot return anything, if a
	 *            topic is different to null, the method will response using
	 *            this topic
	 */
	public AsyncRequest(String uid, String corrId, String methodName, Vector<Object> args, String topic) {
		super(uid, corrId, methodName, args);
		if (topic != null && topic.equals("")) {
			topic = null;
		} else {
			this.topic = topic;
		}
	}

	@Override
	public void invokeRequest(Remote obj, BasicProperties props) throws Exception {
		Response resp = obj.invokeMethod(this.getMethodName(), this.getArgs());

		String topic = props.getReplyTo();
		// If this request has a topic, a new event will be generated and
		// published to everybody interested in.
		if (topic != null) {
			Channel channel = RequestListener.getRequestListener().getChannel();
			channel.exchangeDeclare(topic, "fanout");

			Event event = new Event(topic, props.getCorrelationId(), resp.getResp());

			channel.basicPublish(topic, "", null, Serializer.serialize(event));
			channel.close();
		}
	}

	@Override
	public Object publishRequest(Map<String, Response> results, Properties env, Channel channel) throws Exception {
		// Get the environment properties
		String exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE);
		String routingkey = env.getProperty(ParameterQueue.RPC_ROUTING_KEY);

		// Add the correlation ID and create a replyTo property
		BasicProperties props = new BasicProperties.Builder().appId(uid).correlationId(corrId).replyTo(topic).build();

		// Publish the message
		channel.basicPublish(exchange, routingkey, props, Serializer.serialize(this));

		return null;
	}

	public String getTopic() {
		return topic;
	}

	public void setTopic(String topic) {
		this.topic = topic;
	}
}
