package omq.server;

import omq.common.util.ParameterQueue;
import omq.exception.SerializerException;

import org.apache.log4j.Logger;

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

public class MultiInvocationThread extends AInvocationThread {

	private static final Logger logger = Logger.getLogger(MultiInvocationThread.class.getName());
	private static final String multi = "multi#";

	// Consumer
	private String multiQueue;

	public MultiInvocationThread(RemoteObject obj) throws Exception {
		super(obj);
	}

	@Override
	public void run() {
		while (!killed) {
			try {
				// Get the delivery
				Delivery delivery = consumer.nextDelivery();
				// This thread does not need to set busy because it's mandatory
				// to exist
				executeTask(delivery);
			} catch (InterruptedException i) {
				logger.error(i);
			} catch (ShutdownSignalException e) {
				logger.error(e);
				try {
					if (channel.isOpen()) {
						channel.close();
					}
					startQueues();
				} catch (Exception e1) {
					try {
						long milis = Long.parseLong(env.getProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"));
						Thread.sleep(milis);
					} catch (InterruptedException e2) {
						logger.error(e2);
					}
					logger.error(e1);
				}
			} catch (ConsumerCancelledException e) {
				logger.error(e);
			} catch (SerializerException e) {
				logger.error(e);
			} catch (Exception e) {
				e.printStackTrace();
				logger.error(e);
			}

		}
		logger.info("ObjectMQ ('" + obj.getRef() + "') InvocationThread " + Thread.currentThread().getId() + " is killed");
	}

	@Override
	protected void startQueues() throws Exception {
		// Start channel
		channel = broker.getNewChannel();

		/*
		 * Multi queue, exclusive per each instance
		 */

		// Get info about the multiQueue
		String multiExchange = multi + reference;
		// TODO:String multiExchange = multi + exchange + reference;
		multiQueue = env.getProperty(ParameterQueue.MULTI_QUEUE_NAME);

		// Multi queue (exclusive queue per remoteObject)
		boolean multiDurable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_MQUEUE, "false"));
		boolean multiExclusive = Boolean.parseBoolean(env.getProperty(ParameterQueue.EXCLUSIVE_MQUEUE, "true"));
		boolean multiAutoDelete = Boolean.parseBoolean(env.getProperty(ParameterQueue.AUTO_DELETE_MQUEUE, "true"));

		// Declares and bindings
		channel.exchangeDeclare(multiExchange, "fanout");
		if (multiQueue == null) {
			multiQueue = channel.queueDeclare().getQueue();
		} else {
			channel.queueDeclare(multiQueue, multiDurable, multiExclusive, multiAutoDelete, null);
		}
		channel.queueBind(multiQueue, multiExchange, "");
		logger.info("RemoteObject: " + reference + " declared fanout exchange: " + multiExchange + ", Queue: " + multiQueue + ", Durable: "
				+ multiDurable + ", Exclusive: " + multiExclusive + ", AutoDelete: " + multiAutoDelete);

		/*
		 * Consumer
		 */

		// Disable Round Robin behavior
		boolean autoAck = false;

		// Declare a new consumer
		consumer = new QueueingConsumer(channel);
		channel.basicConsume(multiQueue, autoAck, consumer);
	}

}