package omq.server.remote.request;

import java.util.Collection;
import java.util.Hashtable;
import java.util.Properties;

import omq.Remote;
import omq.common.remote.RemoteListener;
import omq.common.util.ParameterQueue;
import omq.exception.ObjectAlreadyExistsException;
import omq.exception.ObjectNotFoundException;
import omq.exception.SerializerException;


import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.QueueingConsumer.Delivery;

/**
 * 
 * @author Sergi Toda <sergi.toda@estudiants.urv.cat> 
 *
 */
public class RequestListener extends RemoteListener {
	private static int numThreads = 4;
	private static RequestListener rListener;
	private Hashtable<String, RemoteWrapper> remoteWrappers;

	private RequestListener(Properties env) throws Exception {
		super(env);
		// Init the hashtable
		this.remoteWrappers = new Hashtable<String, RemoteWrapper>();

		// Get info about the queue & the exchange where the RemoteListener will
		// listen to.
		String exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE);
		String queue = env.getProperty(ParameterQueue.RPC_QUEUE);
		String key = env.getProperty(ParameterQueue.RPC_ROUTING_KEY);

		// Declares and bindings
		channel.exchangeDeclare(exchange, ParameterQueue.RPC_TYPE);
		channel.queueDeclare(queue, false, false, false, null);
		channel.queueBind(queue, exchange, key);

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

	@Override
	public void run() {
		System.out.println("waiting for requests");
		while (!killed) {
			try {
				Delivery delivery = consumer.nextDelivery();
				String key = delivery.getProperties().getAppId();
				RemoteWrapper remoteWrapper = remoteWrappers.get(key);
				remoteWrapper.notifyDelivery(delivery);
			} catch (InterruptedException i) {
				i.printStackTrace();
			} catch (ShutdownSignalException e) {
				e.printStackTrace();
			} catch (ConsumerCancelledException e) {
				e.printStackTrace();
			} catch (SerializerException e) {
				e.printStackTrace();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	public static void stopRequestListener() throws Exception {
		rListener.kill();
		Hashtable<String, RemoteWrapper> hashTable = rListener.getRemoteWrappers();
		Collection<RemoteWrapper> collection = hashTable.values();
		for (RemoteWrapper rW : collection) {
			rW.stopRemoteWrapper();
		}
		rListener = null;
	}

	public static void init(Properties env) throws Exception {
		if (rListener == null) {
			rListener = new RequestListener(env);
			rListener.start();
		} else {
			throw new Exception("Cannot init because it already exists");
		}
	}

	public static RequestListener getRequestListener() throws Exception {
		if (rListener == null) {
			throw new Exception("Request listener not initialized");
		}
		return rListener;
	}

	public synchronized void addObj(Remote obj) throws ObjectAlreadyExistsException {
		if (!remoteWrappers.containsKey(obj.getRef())) {
			RemoteWrapper remoteWrapper = new RemoteWrapper(obj, numThreads);
			remoteWrappers.put(obj.getRef(), remoteWrapper);
		} else {
			throw new ObjectAlreadyExistsException(obj.getRef());
		}
	}

	public synchronized void removeObj(Remote obj) throws ObjectNotFoundException {
		if (remoteWrappers.containsKey(obj.getRef())) {
			remoteWrappers.remove(obj.getRef());
		} else {
			throw new ObjectNotFoundException(obj.getRef());
		}
	}

	public synchronized Remote getObj(String key) {
		return remoteWrappers.get(key).getObj();
	}

	public Hashtable<String, RemoteWrapper> getRemoteWrappers() {
		return remoteWrappers;
	}
}
