Changeset 103 for branches/supervisor/src

Timestamp:

10/15/13 13:21:33 (11 years ago)

Author:

stoda

Message:

Supervisor done

TODO: revise supervisor

Location:

branches/supervisor/src

Files:

• main/java/omq/common/broker/HasObject.java (modified) (1 diff)
• main/java/omq/common/broker/RemoteBroker.java (modified) (2 diffs)
• main/java/omq/common/broker/RemoteBrokerImpl.java (modified) (2 diffs)
• main/java/omq/supervisor/Supervisor.java (modified) (1 diff)
• main/java/omq/supervisor/SupervisorImpl.java (modified) (4 diffs)
• main/java/omq/supervisor/SupervisorThread.java (modified) (2 diffs)
• test/java/omq/test/supervisor/SleepTest.java (modified) (1 diff)

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

@@ -37 +37 @@
         }
 
-        public boolean isHasObject() {
+        public boolean hasObject() {
                 return hasObject;
         }

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

@@ -8 +8 @@
 import omq.client.annotation.SyncMethod;
 import omq.exception.RemoteException;
+import omq.exception.RetryException;
 
 public interface RemoteBroker extends Remote {
@@ -19 +20 @@
 
         @SyncMethod(retry = 1, timeout = 1000)
-        public boolean hasObject(String reference);
+        public boolean hasObject(String reference) throws RetryException;
+
+        @SyncMethod(retry = 1, timeout = 1000)
+        public HasObject hasObjectInfo(String reference) throws RetryException;
 
 }

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

@@ -6 +6 @@
 
 import omq.exception.RemoteException;
+import omq.exception.RetryException;
 import omq.server.RemoteObject;
 
@@ -43 +44 @@
 
         @Override
-        public boolean hasObject(String reference) {
+        public boolean hasObject(String reference) throws RetryException {
                 return getBroker().getRemoteObjs().containsKey(reference);
         }
 
+        @Override
+        public HasObject hasObjectInfo(String reference) throws RetryException {
+                if (getBroker().getRemoteObjs().containsKey(reference)) {
+                        RemoteObject r = getBroker().getRemoteObjs().get(reference);
+                        int numThreads = r.getPool().getWorkers().size();
+                        return new HasObject(this.getRef(), reference, true, numThreads);
+                }
+                return new HasObject(this.getRef(), reference, false, 0);
+        }
+
 }

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

@@ -7 +7 @@
         public void subscribe(String brokerName) throws Exception;
 
-        public void spawnObject(OmqSettings settings) throws Exception;
+        public void spawnObject(OmqSettings settings, int numObjects) throws Exception;
 
-        public void unbindObject(OmqSettings settings) throws Exception;
+        public void unbindObject(OmqSettings settings, int numObjects) throws Exception;
 
 }

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

@@ -10 +10 @@
 
 import org.apache.log4j.Logger;
-
-import com.rabbitmq.client.AMQP.Queue.DeclareOk;
-import com.rabbitmq.client.Channel;
 
 public class SupervisorImpl extends RemoteObject implements Supervisor {
@@ -41 +38 @@
 
         @Override
-        public void spawnObject(OmqSettings settings) throws Exception {
+        public void spawnObject(OmqSettings settings, int numObjects) throws Exception {
 
                 String reference = settings.getReference();
@@ -50 +47 @@
 
                 int minObjects = settings.getMinNumberObjects();
-
-                Channel channel = getBroker().getNewChannel();
-
-                int numObjects = 0;
-                try {
-                        DeclareOk dok = channel.queueDeclarePassive(reference);
-                        numObjects = dok.getConsumerCount();
-                        channel.close();
-                } catch (Exception io) {
-                        // The queue doesn't exist & the channel has been closed
-                }
 
                 for (RemoteBroker broker : brokers) {
@@ -75 +61 @@
 
         @Override
-        public void unbindObject(OmqSettings settings) throws Exception {
+        public void unbindObject(OmqSettings settings, int numObjects) throws Exception {
                 String reference = settings.getReference();
 
                 int minObjects = settings.getMinNumberObjects();
 
-                Channel channel = getBroker().getNewChannel();
+                for (RemoteBroker broker : brokers) {
+                        if (broker.hasObject(reference) && (numObjects - 1) >= minObjects) {
+                                broker.deleteObject(reference);
+                                break;
+                        }
+                }
 
-                int numObjects = 0;
-                try {
-                        DeclareOk dok = channel.queueDeclarePassive(reference);
-                        numObjects = dok.getConsumerCount();
-                        channel.close();
-
-                        for (RemoteBroker broker : brokers) {
-                                if (broker.hasObject(reference) && (numObjects - 1) >= minObjects) {
-                                        broker.deleteObject(reference);
-                                        break;
-                                }
-                        }
-
-                } catch (Exception io) {
-                        // The queue doesn't exist & the channel has been closed
-                }
         }
 

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

@@ -3 +3 @@
 import java.util.Map;
 import java.util.Set;
+
+import omq.common.broker.RemoteBroker;
+import omq.exception.RetryException;
 
 import org.apache.log4j.Logger;
@@ -49 +52 @@
                 int minMessages = settings.getMinNumQueued();
 
-        
                 Channel channel = supervisor.getBroker().getChannel();
                 DeclareOk dok = channel.queueDeclarePassive(reference);
 
-                int numConsumers = dok.getConsumerCount();
+                int numObjects = getNumObjects(reference);
                 int numMessages = dok.getMessageCount();
 
-                System.out.println("Num Consumers: " + numConsumers + ", num Messages: " + numMessages);
+                System.out.println("Num Consumers: " + numObjects + ", num Messages: " + numMessages);
 
-                if (maxMessages < numMessages || numConsumers < minObjects) {
+                if (maxMessages < numMessages || numObjects < minObjects) {
                         logger.info("SPAWN TIME!!");
-                        supervisor.spawnObject(settings);
-                } else if (numMessages < minMessages && minObjects < numConsumers) {
+                        supervisor.spawnObject(settings, numObjects);
+                } else if (numMessages < minMessages && minObjects < numObjects) {
                         logger.info("Unbinding object!!!");
-                        supervisor.unbindObject(settings);
+                        supervisor.unbindObject(settings, numObjects);
                 }
         }
+
+        private int getNumObjects(String reference) {
+                int num = 0;
+                for (RemoteBroker broker : supervisor.getBrokers()) {
+                        try {
+                                if (broker.hasObject(reference)) {
+                                        num++;
+                                }
+                        } catch (RetryException e) {
+                                e.printStackTrace();
+                        }
+                }
+                return num;
+        }
 }

branches/supervisor/src/test/java/omq/test/supervisor/SleepTest.java

@@ -63 +63 @@
 
                 OmqSettings settings = new OmqSettings("sleep", SleepImpl.class.getName(), env, 1, 20, 20);
-                supervisor.spawnObject(settings);
+                supervisor.spawnObject(settings, 0);
 
                 Sleep sleep = broker.lookup("sleep", Sleep.class);