source: branches/supervisor/src/main/java/omq/server/RemoteThreadPool.java

package omq.server;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.ListIterator;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.Logger;

import omq.common.broker.Broker;

import com.rabbitmq.client.AMQP.Queue.DeclareOk;
import com.rabbitmq.client.Channel;

/**
 * 
 * @author Sergi Toda <sergi.toda@estudiants.urv.cat>
 * 
 */

public class RemoteThreadPool extends Thread {
        private static final Logger logger = Logger.getLogger(RemoteThreadPool.class.getName());
        private List<InvocationThread> workers;
        private MultiInvocationThread multiWorker;
        private AtomicInteger busy;
        private int minPoolThreads;
        private int maxPoolThreads;
        private long refresh;
        private long keepAliveTime;
        private int maxMessagesPerThread;

        private RemoteObject obj;
        private Broker broker;
        private boolean killed = false;

        public RemoteThreadPool(int minPoolThreads, int maxPoolThreads, long refresh, long keepAliveTime, int maxMessagesPerThread,
                        RemoteObject obj, Broker broker) {
                this.minPoolThreads = minPoolThreads;
                this.maxPoolThreads = maxPoolThreads;
                this.refresh = refresh;
                this.keepAliveTime = keepAliveTime;
                this.maxMessagesPerThread = maxMessagesPerThread;
                this.obj = obj;
                this.broker = broker;

                workers = new ArrayList<InvocationThread>(minPoolThreads);
                busy = new AtomicInteger();
        }

        @Override
        public void run() {

                /*
                 * Create and start minPoolThreads
                 */
                logger.info("ObjectMQ reference: " + obj.getRef() + ", creating: " + minPoolThreads + ", maxPoolThreads: " + maxPoolThreads
                                + ", refresh time: " + refresh + ", keepAlive: " + keepAliveTime + ", maxMessagesPerThread: " + maxMessagesPerThread);

                try {
                        multiWorker = new MultiInvocationThread(obj);
                        multiWorker.start();
                } catch (Exception e1) {
                        // TODO Auto-generated catch block
                        e1.printStackTrace();
                }

                for (int i = 0; i < minPoolThreads; i++) {
                        try {
                                InvocationThread iThread = new InvocationThread(obj);
                                workers.add(iThread);
                                iThread.start();
                        } catch (Exception e) {
                                logger.error("Error while creating pool threads", e);
                                e.printStackTrace();
                        }
                }

                while (!killed) {

                        try {
                                Channel channel = broker.getChannel();
                                DeclareOk dok = channel.queueDeclarePassive(obj.getRef());

                                int numConsumers = dok.getConsumerCount();
                                int numMessages = dok.getMessageCount();
                                int division = numMessages / numConsumers;
                                int numWorkers = workers.size();

                                if (numWorkers < maxPoolThreads && division >= maxMessagesPerThread) {
                                        // Create a new thread
                                        System.out.println("Add worker");
                                        InvocationThread worker = new InvocationThread(obj);
                                        workers.add(worker);
                                        worker.start();
                                } else if (numWorkers > minPoolThreads && busy.get() < numWorkers) {
                                        // Kill idle threads
                                        System.out.println("Kill lazy workers, numWorkers = " + numWorkers + ", minPool = " + minPoolThreads + ", busy = "
                                                        + busy.get());
                                        stopIdleThreads();
                                }

                                Thread.sleep(refresh);

                        } catch (Exception e) {
                                e.printStackTrace();
                                logger.error(e);
                        }
                }

        }

        private void stopIdleThreads() {
                long now = System.currentTimeMillis();

                ListIterator<InvocationThread> lIter = workers.listIterator();
                while (workers.size() > minPoolThreads && lIter.hasNext()) {
                        InvocationThread worker = lIter.next();
                        long lastExec = worker.getLastExecution();
                        System.out.println("last - now = " + (now - lastExec) + " keep alive = " + keepAliveTime);
                        if (worker.isIdle() && (now - lastExec) > keepAliveTime) {
                                // Kill thread
                                try {
                                        worker.kill();
                                        lIter.remove();
                                } catch (IOException e) {
                                        logger.error(e);
                                }

                        }
                }
        }

        public void kill() throws IOException {
                killed = true;
                for (InvocationThread iThread : workers) {
                        iThread.kill();
                }
                interrupt();
        }

        public List<InvocationThread> getWorkers() {
                return workers;
        }

        public void setWorkers(List<InvocationThread> workers) {
                this.workers = workers;
        }

        public AtomicInteger getBusy() {
                return busy;
        }

        public void setBusy(AtomicInteger busy) {
                this.busy = busy;
        }

        public int getMinPoolThreads() {
                return minPoolThreads;
        }

        public void setMinPoolThreads(int minPoolThreads) {
                this.minPoolThreads = minPoolThreads;
        }

        public int getMaxPoolThreads() {
                return maxPoolThreads;
        }

        public void setMaxPoolThreads(int maxPoolThreads) {
                this.maxPoolThreads = maxPoolThreads;
        }

        public long getRefresh() {
                return refresh;
        }

        public void setRefresh(long refresh) {
                this.refresh = refresh;
        }

        public long getKeepAliveTime() {
                return keepAliveTime;
        }

        public void setKeepAliveTime(long keepAliveTime) {
                this.keepAliveTime = keepAliveTime;
        }

        public int getMaxMessagesPerThread() {
                return maxMessagesPerThread;
        }

        public void setMaxMessagesPerThread(int maxMessagesPerThread) {
                this.maxMessagesPerThread = maxMessagesPerThread;
        }

}