Changeset 93

Timestamp:

10/01/13 15:34:32 (11 years ago)

Author:

stoda

Message:

TODO: unbind object

Location:

branches/supervisor/src

Files:

• main/java/omq/supervisor/Supervisor.java (modified) (1 diff)
• main/java/omq/supervisor/SupervisorImpl.java (modified) (2 diffs)
• main/java/omq/supervisor/SupervisorThread.java (modified) (1 diff)
• test/java/omq/test/supervisor/SleepTest.java (modified) (2 diffs)

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

@@ -9 +9 @@
         public void spawnObject(OmqSettings settings) throws Exception;
 
+        public void unbindObject(OmqSettings settings) throws Exception;
+
 }

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

@@ -44 +44 @@
 
                 String reference = settings.getReference();
-                objectSettings.put(reference, settings);
+
+                if (!objectSettings.containsKey(reference)) {
+                        objectSettings.put(reference, settings);
+                }
 
                 int minObjects = settings.getMinNumberObjects();
@@ -69 +72 @@
                 }
 
+        }
+
+        @Override
+        public void unbindObject(OmqSettings settings) throws Exception {
+                String reference = settings.getReference();
+
+                int minObjects = settings.getMinNumberObjects();
+
+                Channel channel = getBroker().getNewChannel();
+
+                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);
+                                }
+                        }
+
+                } catch (Exception io) {
+                        // The queue doesn't exist & the channel has been closed
+                }
+
+                for (RemoteBroker broker : brokers) {
+                        if (!broker.hasObject(reference) && minObjects >= numObjects) {
+                                broker.spawnObject(reference, settings.getClassName(), settings.getProps());
+                                numObjects++;
+                                if (minObjects >= numObjects) {
+                                        break;
+                                }
+                        }
+                }
         }
 

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

@@ -61 +61 @@
                         // pregunta a tots i qui no té l'objecte li poses
                 } else if (numMessages < minMessages && minObjects > numConsumers) {
+                        supervisor.unbindObject(settings);
                         // delete:
                         // pregunta a tots i qui té l'objecte li treus

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

@@ -15 +15 @@
 
         @BeforeClass
-        public static void Server() throws Exception {System.out.println("hola");
+        public static void Server() throws Exception {
+                System.out.println("hola");
                 Properties env1 = new Properties();
                 env1.setProperty(ParameterQueue.USER_NAME, "guest");
@@ -59 +60 @@
                 Supervisor supervisor = broker.lookup("supervisor", Supervisor.class);
 
-                OmqSettings settings = new OmqSettings("sleep", SleepImpl.class.getName(), env, 1, 70, 2);
+                OmqSettings settings = new OmqSettings("sleep", SleepImpl.class.getName(), env, 20, 20, 2);
                 supervisor.spawnObject(settings);
 
                 Sleep sleep = broker.lookup("sleep", Sleep.class);
 
-                for (int i = 0; i < 50; i++) {
+                for (int i = 0; i < 10; i++) {
                         sleep.sleep();
                 }
                 Thread.sleep(5000);
-                for (int i = 0; i < 50; i++) {
+                for (int i = 0; i < 20; i++) {
                         sleep.sleep();
                 }
-                
+
                 Thread.sleep(100000);
         }