Changeset 100 for branches/supervisor

Timestamp:

10/09/13 17:35:41 (11 years ago)

Author:

stoda

Message:

Idea of thread supervisor done

TODO test the idea, change the remote properties, etc

Location:

branches/supervisor

Files:

• .project (modified) (1 diff)
• src/main/java/omq/common/broker/HasObject.java (modified) (2 diffs)
• src/main/java/omq/server/InvocationThread.java (modified) (5 diffs)
• src/main/java/omq/server/RemoteObject.java (modified) (1 diff)
• src/main/java/omq/server/RemoteThreadPool.java (added)
• src/main/java/omq/supervisor/SupervisorThread.java (modified) (3 diffs)

branches/supervisor/.project

@@ -1 +1 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <projectDescription>
-        <name>objectmq</name>
+        <name>objectmq_supervisor</name>
         <comment></comment>
         <projects>

branches/supervisor/src/main/java/omq/common/broker/HasObject.java

@@ -12 +12 @@
         private String reference;
         private boolean hasObject;
+        private int numThreads;
 
-        public HasObject(String brokerName, String reference, boolean hasObject) {
+        public HasObject(String brokerName, String reference, boolean hasObject, int numThreads) {
                 this.brokerName = brokerName;
                 this.reference = reference;
                 this.hasObject = hasObject;
+                this.numThreads = numThreads;
         }
 
@@ -43 +45 @@
         }
 
+        public int getNumThreads() {
+                return numThreads;
+        }
+
+        public void setNumThreads(int numThreads) {
+                this.numThreads = numThreads;
+        }
+
 }

branches/supervisor/src/main/java/omq/server/InvocationThread.java

@@ -37 +37 @@
         private String UID;
         private Properties env;
+        private boolean idle;
+        private long lastExec;
+
+        private RemoteThreadPool pool; // TODO posar això bé
 
         // Broker
@@ -54 +58 @@
                 this.broker = broker;
                 this.serializer = broker.getSerializer();
+                this.lastExec = 0;
+                this.idle = true;
         }
 
@@ -73 +79 @@
                                 // Get the delivery
                                 Delivery delivery = consumer.nextDelivery();
+
+                                // This thread gets busy
+                                pool.getBusy().incrementAndGet();
+                                idle = false;
 
                                 String serializerType = delivery.getProperties().getType();
@@ -112 +122 @@
 
                                 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
+
+                                // The thread is now idle
+                                lastExec = System.currentTimeMillis();
+                                idle = true;
+                                pool.getBusy().decrementAndGet();
+
                         } catch (InterruptedException i) {
                                 logger.error(i);
@@ -231 +247 @@
         }
 
+        public long getLastExecution() {
+                return lastExec;
+        }
+
+        public boolean isIdle() {
+                return idle;
+        }
+
 }

branches/supervisor/src/main/java/omq/server/RemoteObject.java

@@ -33 +33 @@
         private Properties env;
         private transient Broker broker;
+        private transient RemoteThreadPool pool;
         private transient Map<String, List<Class<?>>> params;
         private transient List<InvocationThread> invocationList;

branches/supervisor/src/main/java/omq/supervisor/SupervisorThread.java

@@ -4 +4 @@
 import java.util.Set;
 
+import org.apache.log4j.Logger;
+
 import com.rabbitmq.client.AMQP.Queue.DeclareOk;
 import com.rabbitmq.client.Channel;
 
 public class SupervisorThread extends Thread {
+        private static final Logger logger = Logger.getLogger(SupervisorThread.class.getName());
 
         private long sleep;
@@ -46 +49 @@
                 int minMessages = settings.getMinNumQueued();
 
-                // TODO treure merda...
+        
                 Channel channel = supervisor.getBroker().getChannel();
                 DeclareOk dok = channel.queueDeclarePassive(reference);
@@ -55 +58 @@
                 System.out.println("Num Consumers: " + numConsumers + ", num Messages: " + numMessages);
 
-                if (maxMessages < numMessages || numConsumers < minObjects ) {
-                        System.out.println("SPAWN TIME!!");
+                if (maxMessages < numMessages || numConsumers < minObjects) {
+                        logger.info("SPAWN TIME!!");
                         supervisor.spawnObject(settings);
-                        // spawn:
-                        // pregunta a tots i qui no t�� l'objecte li poses
                 } else if (numMessages < minMessages && minObjects < numConsumers) {
-                        System.out.println("Unbinding object!!!");
+                        logger.info("Unbinding object!!!");
                         supervisor.unbindObject(settings);
-                        // delete:
-                        // pregunta a tots i qui t�� l'objecte li treus
                 }
         }