package omq.client.listener;

import java.io.IOException;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Map;
import java.util.Properties;

import omq.client.proxy.Proxymq;
import omq.common.broker.Broker;
import omq.common.util.ParameterQueue;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownSignalException;

/**
 * Class that inherits from RemoteListener. It's used in the server side. This
 * class gets the deliveries from the server and stores them into the proxies
 * 
 * @author Sergi Toda <sergi.toda@estudiants.urv.cat>
 * 
 */
public class ResponseListener extends Thread {
	private static ResponseListener rListener;

	private Channel channel;
	private QueueingConsumer consumer;
	private boolean killed = false;
	private Map<String, Map<String, byte[]>> results;
	private Properties env;

	/**
	 * Protected constructor used by the singleton pattern
	 * 
	 * @param env
	 * @throws Exception
	 */
	protected ResponseListener(Properties env) throws Exception {
		this.env = env;

		// Init the hashtable (it's concurrent)
		this.results = new Hashtable<String, Map<String, byte[]>>();

		startRPCQueue();
	}

	@Override
	public void run() {
		Delivery delivery;
		String uid_request;

		while (!killed) {
			try {
				// Get the delivery

				delivery = consumer.nextDelivery();

				BasicProperties props = delivery.getProperties();

				// Get the response with its uid
				uid_request = delivery.getProperties().getCorrelationId();
				System.out.println("Response received -> " + uid_request);

				// Stores the new response
				Map<String, byte[]> proxyResults = results.get(props.getAppId());

				// Put the result into the proxy results and notify him
				synchronized (proxyResults) {
					// If we haven't received this response before, we store it
					if (!proxyResults.containsKey(uid_request)) {
						proxyResults.put(uid_request, delivery.getBody());
						proxyResults.notifyAll();
					}
				}
			} catch (InterruptedException i) {
				i.printStackTrace();
			} catch (ShutdownSignalException e) {
				e.printStackTrace();
				try {
					if (channel.isOpen()) {
						channel.close();
					}
					startRPCQueue();
				} catch (Exception e1) {
					e1.printStackTrace();
					try {
						long milis = Long.parseLong(env.getProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"));
						Thread.sleep(milis);
					} catch (InterruptedException e2) {
						e2.printStackTrace();
					}
				}
			} catch (ConsumerCancelledException e) {
				e.printStackTrace();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	private void startRPCQueue() throws Exception {
		channel = Broker.getNewChannel();

		Map<String, Object> args = null;

		String reply_queue = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
		boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false"));

		int ttl = Integer.parseInt(env.getProperty(ParameterQueue.MESSAGE_TTL_IN_QUEUES, "-1"));
		if (ttl > 0) {
			args = new HashMap<String, Object>();
			args.put("x-message-ttl", ttl);
		}

		channel.queueDeclare(reply_queue, durable, false, false, args);

		// Declare a new consumer
		consumer = new QueueingConsumer(channel);
		channel.basicConsume(reply_queue, true, consumer);
	}

	/**
	 * Static function which initializes the ResponseListener
	 * 
	 * @param env
	 * @throws Exception
	 */
	public static void init(Properties env) throws Exception {
		if (rListener == null) {
			rListener = new ResponseListener(env);
			rListener.start();
		} else {
			throw new Exception("Cannot init because it already exists");
		}
	}

	/**
	 * Method to retrieve the unique ResponseListener, this function can also
	 * initialize a ResponseListener using and environment
	 * 
	 * @param env
	 * @return unique ResponseListener
	 * @throws Exception
	 */
	public static ResponseListener getRequestListener(Properties env) throws Exception {
		if (rListener == null) {
			rListener = new ResponseListener(env);
			rListener.start();
		} else {
			// TODO: create a new exception to indicate that a response listener
			// cannot be init
			throw new Exception("Cannot init because it already exists");
		}
		return rListener;
	}

	public static boolean isVoid() {
		return rListener == null;
	}

	/**
	 * Method to retrieve the unique ResponseListener
	 * 
	 * @return
	 * @throws Exception
	 */
	public static ResponseListener getRequestListener() throws Exception {
		if (rListener == null) {
			throw new Exception("Request listener not initialized");
		}
		return rListener;
	}

	/**
	 * 
	 * @param key
	 * @return whether the map has the param key
	 */
	public boolean containsKey(String key) {
		return results.containsKey(key);
	}

	/**
	 * This method is used to kill the unique responseListener in the system
	 * 
	 * @throws Exception
	 */
	public static void stopResponseListner() throws Exception {
		rListener.kill();
		rListener = null;
	}

	/**
	 * Interrupt and kill the Thread
	 * 
	 * @throws IOException
	 */
	public void kill() throws IOException {
		interrupt();
		killed = true;
		channel.close();
	}

	// Revisar això
	public void registerProxy(Proxymq proxy) {
		if (!results.containsKey(proxy.getRef())) {
			results.put(proxy.getRef(), proxy.getResults());
		}
	}
}
