source: branches/supervisor/src/main/java/omq/server/RemoteWrapper.java @ 92

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;
        }
}