package omq.server;

import java.util.ArrayList;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

import omq.common.util.Serializer;

import org.apache.log4j.Logger;

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

/**
 * This class is used to encapsulate the invocationThreads under the
 * RemoteObject.
 * 
 * @author Sergi Toda <sergi.toda@estudiants.urv.cat>
 * 
 */
public class RemoteWrapper {
	private static final Logger logger = Logger.getLogger(RemoteWrapper.class.getName());

	private RemoteObject obj;
	private int numThreads;
	// private AtomicInteger busy;
	private Object waitLock;
	private ArrayList<InvocationThread> invocationList;
	private BlockingQueue<Delivery> deliveryQueue;

	public RemoteWrapper(RemoteObject obj, int numThreads, Serializer serializer) {
		this.obj = obj;
		this.numThreads = numThreads;
		// this.busy = new AtomicInteger(0);
		this.waitLock = new Object();
		invocationList = new ArrayList<InvocationThread>();
		deliveryQueue = new LinkedBlockingDeque<QueueingConsumer.Delivery>();

		logger.info("Object reference: " + obj.getRef() + ", numthreads listening = " + numThreads);

		for (int i = 0; i < numThreads; i++) {
			InvocationThread thread = new InvocationThread(obj, this, serializer);
			invocationList.add(thread);
			thread.start();
		}
	}

	/**
	 * This method notifies a delivery to an invocationThread using a
	 * blockingQueue.
	 * 
	 * @param delivery
	 *            - delivery which contains a Request to be invoked
	 * @throws Exception
	 */
	public void notifyDelivery(Delivery delivery) throws Exception {

		// // Ensure there is at least one thread available
		// while (this.busy.get() == this.numThreads) {
		// System.out.println("Waiting for a thread available");
		// logger.debug("Object reference: " + obj.getRef() + " is busy");
		//
		// synchronized (waitLock) {
		// waitLock.wait();
		// }
		// }
		// Notify an available thread
		this.deliveryQueue.put(delivery);

	}

	/**
	 * This method interrups all the invocationThreads under this remoteWrapper
	 */
	public void stopRemoteWrapper() {
		logger.warn("Stopping Invocation threads vinculed to " + obj.getRef());
		for (InvocationThread thread : invocationList) {
			thread.interrupt();
		}
	}

	// public int increaseBusy() {
	// return this.busy.incrementAndGet();
	// }
	//
	// public int decreaseBusy() {
	// int value = this.busy.decrementAndGet();
	// synchronized (waitLock) {
	// waitLock.notifyAll();
	// }
	// return value;
	// }

	public RemoteObject getObj() {
		return obj;
	}

	public void setObj(RemoteObject obj) {
		this.obj = obj;
	}

	public int getNumThreads() {
		return numThreads;
	}

	public void setNumThreads(int numThreads) {
		this.numThreads = numThreads;
	}

	public ArrayList<InvocationThread> getInvocationList() {
		return invocationList;
	}

	public void setInvocationList(ArrayList<InvocationThread> invocationList) {
		this.invocationList = invocationList;
	}

	public BlockingQueue<Delivery> getDeliveryQueue() {
		return deliveryQueue;
	}

	public void setDeliveryQueue(BlockingQueue<Delivery> deliveryQueue) {
		this.deliveryQueue = deliveryQueue;
	}

	public Object getLock() {
		return waitLock;
	}

	public void setLock(Object lock) {
		this.waitLock = lock;
	}
}
