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

package omq.server;

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

import omq.common.broker.Broker;

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

public class RemoteThreadPool extends Thread {
        private List<InvocationThread> workers;
        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() {

                // Crear aquí tots els fils?

                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
                                        InvocationThread worker = new InvocationThread(obj, broker);
                                        workers.add(worker);
                                        worker.start();
                                } else if (numWorkers > minPoolThreads && busy.get() < numWorkers) {
                                        // Kill idle threads
                                        stopIdleThreads();
                                }

                                Thread.sleep(refresh);

                        } catch (Exception e) {

                        }
                }

        }

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

                for (InvocationThread worker : workers) {
                        long lastExec = worker.getLastExecution();
                        if (worker.isIdle() && (lastExec - now) > keepAliveTime) {
                                // Kill thread
                                try {
                                        worker.kill();
                                } catch (IOException e) {
                                        // TODO Auto-generated catch block
                                        e.printStackTrace();
                                }

                        }
                }
        }

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

}