Changeset 37

Timestamp:

06/12/13 17:02:52 (11 years ago)

Author:

stoda

Message:

Stop broker working and unbind too.

Location:

trunk/objectmq

Files:

• src/omq/common/broker/Broker.java (modified) (9 diffs)
• src/omq/server/InvocationThread.java (modified) (1 diff)
• src/omq/server/RemoteObject.java (modified) (1 diff)
• test/farmTest (deleted)
• test/faultToleranceTest (moved) (moved from trunk/objectmq/test/clientToleranceTest)
• test/faultToleranceTest/ClientTest.java (modified) (1 diff)
• test/faultToleranceTest/ServerTest.java (modified) (1 diff)
• test/multiThreadTest (moved) (moved from trunk/objectmq/test/test2)
• test/multiThreadTest/Car.java (modified) (1 diff)
• test/multiThreadTest/CarImpl.java (copied) (copied from trunk/objectmq/test/test2/CarImpl.java) (1 diff)
• test/multiThreadTest/CarThread.java (modified) (1 diff)
• test/multiThreadTest/ClientTest.java (modified) (1 diff)
• test/multiThreadTest/Mobile.java (modified) (1 diff)
• test/multiThreadTest/MobileImpl.java (copied) (copied from trunk/objectmq/test/test2/MobileImpl.java) (1 diff)
• test/multiThreadTest/MobileThread.java (modified) (1 diff)
• test/multiThreadTest/Rim.java (modified) (1 diff)
• test/multiThreadTest/ServerTest.java (modified) (1 diff)
• test/stopBroker (added)
• test/stopBroker/BrokerKiller.java (added)
• test/stopBroker/BrokerKillerImpl.java (added)
• test/stopBroker/ClientTest.java (added)
• test/stopBroker/ServerTest.java (added)
• test/stopBroker/UnbindTest.java (added)
• test/test/Client.java (modified) (1 diff)
• test/test/ClientImpl.java (modified) (1 diff)
• test/test/ClientTest.java (added)
• test/test/ServerTest.java (added)

trunk/objectmq/src/omq/common/broker/Broker.java

@@ -2 +2 @@
 
 import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Properties;
 
@@ -30 +32 @@
         private static Channel channel;
         private static boolean clientStarted = false;
-
-        public static void initBroker(Properties env) throws Exception {
+        private static boolean connectionClosed = false;
+        // TODO ask Pedro if it can be only one object in the map (an object can
+        // have multiple threads in the same broker -see environment-)
+        private static Map<String, RemoteObject> remoteObjs;
+
+        /**
+         * Initializes a new Broker with the environment called by reference
+         * 
+         * @param env
+         * @throws Exception
+         */
+        public static synchronized void initBroker(Properties env) throws Exception {
                 if (Environment.isVoid()) {
+                        remoteObjs = new HashMap<String, RemoteObject>();
                         Environment.setEnvironment(env);
                         connection = OmqConnectionFactory.getNewConnection(env);
@@ -44 +57 @@
                                 throw new InitBrokerException("The connection didn't work");
                         }
-                }
-        }
-
-        // TODO: what happens if the connection is not set
+                } else {
+                        throw new InitBrokerException("Broker already started");
+                }
+        }
+
+        public static void stopBroker() throws Exception {
+                // Stop the client
+                if (clientStarted) {
+                        ResponseListener.stopResponseListner();
+                        EventDispatcher.stopEventDispatcher();
+                }
+                // Stop all the remote objects working
+                for (String reference : remoteObjs.keySet()) {
+                        unbind(reference);
+                }
+                // Close the connection once all the listeners are died
+                closeConnection();
+        }
+
+        /**
+         * @return Broker's connection
+         * @throws Exception
+         */
         public static Connection getConnection() throws Exception {
                 return connection;
         }
 
+        public static void closeConnection() throws IOException {
+                connectionClosed = true;
+                connection.close();
+        }
+
+        /**
+         * 
+         * @return Broker's channel
+         * @throws Exception
+         */
         public static Channel getChannel() throws Exception {
                 return channel;
         }
 
+        /**
+         * Creates a new channel using the Broker's connection
+         * 
+         * @return newChannel
+         * @throws IOException
+         */
         public static Channel getNewChannel() throws IOException {
                 return connection.createChannel();
@@ -85 +133 @@
                         Properties environment = Environment.getEnvironment();
                         remote.startRemoteObject(reference, environment);
+                        remoteObjs.put(reference, remote);
                 } catch (Exception e) {
                         throw new RemoteException(e);
@@ -90 +139 @@
         }
 
-        public static void unbind(String reference) throws RemoteException {
+        public static void unbind(String reference) throws RemoteException, IOException {
+                if (remoteObjs.containsKey(reference)) {
+                        RemoteObject remote = remoteObjs.get(reference);
+                        remote.kill();
+                } else {
+                        throw new RemoteException("The object referenced by 'reference' does not exist in the Broker");
+                }
 
         }
@@ -98 +153 @@
         }
 
+        /**
+         * This method ensures the client will have only one ResponseListener and
+         * only one EventDispatcher. Both with the same environment.
+         * 
+         * @param environment
+         * @throws Exception
+         */
         private static synchronized void initClient(Properties environment) throws Exception {
                 if (ResponseListener.isVoid()) {
@@ -107 +169 @@
         }
 
+        /**
+         * This method sends an event with its information
+         * 
+         * @param event
+         * @throws IOException
+         * @throws SerializerException
+         */
         public static void trigger(Event event) throws IOException, SerializerException {
                 String UID = event.getTopic();
@@ -119 +188 @@
         }
 
+        /**
+         * This function is used to send a ping message to see if the connection
+         * works
+         * 
+         * @param env
+         * @throws Exception
+         */
         public static void tryConnection(Properties env) throws Exception {
+                System.out.println("hola");
                 Channel channel = connection.createChannel();
                 String message = "ping";
@@ -148 +225 @@
         }
 
+        /**
+         * This method adds a ShutdownListener to the Broker's connection. When this
+         * connection falls, a new connection will be created and this will also
+         * have the listener.
+         */
         private static void addFaultTolerance() {
                 connection.addShutdownListener(new ShutdownListener() {
                         @Override
                         public void shutdownCompleted(ShutdownSignalException cause) {
-
-                                if (cause.isHardError()) {
-                                        if (connection.isOpen()) {
+                                if (!connectionClosed)
+                                        if (cause.isHardError()) {
+                                                if (connection.isOpen()) {
+                                                        try {
+                                                                connection.close();
+                                                        } catch (IOException e) {
+                                                                e.printStackTrace();
+                                                        }
+                                                }
                                                 try {
-                                                        connection.close();
-                                                } catch (IOException e) {
+                                                        Properties env = Environment.getEnvironment();
+                                                        connection = OmqConnectionFactory.getNewWorkingConnection(env);
+                                                        channel = connection.createChannel();
+                                                        addFaultTolerance();
+                                                } catch (Exception e) {
                                                         e.printStackTrace();
                                                 }
-                                        }
-                                        try {
-                                                Properties env = Environment.getEnvironment();
-                                                connection = OmqConnectionFactory.getNewWorkingConnection(env);
-                                                channel = connection.createChannel();
-                                                addFaultTolerance();
-                                        } catch (Exception e) {
-                                                e.printStackTrace();
-                                        }
-                                } else {
-                                        Channel channel = (Channel) cause.getReference();
-                                        if (channel.isOpen()) {
-                                                try {
-                                                        channel.close();
-                                                } catch (IOException e) {
-                                                        e.printStackTrace();
+                                        } else {
+                                                Channel channel = (Channel) cause.getReference();
+                                                if (channel.isOpen()) {
+                                                        try {
+                                                                channel.close();
+                                                        } catch (IOException e) {
+                                                                e.printStackTrace();
+                                                        }
                                                 }
                                         }
-                                }
                         }
                 });

trunk/objectmq/src/omq/server/InvocationThread.java

@@ -69 +69 @@
                         }
 
-                }
+                }System.out.println("Invocation Thread dies!!");
         }
 

trunk/objectmq/src/omq/server/RemoteObject.java

@@ -129 +129 @@
 
         public void kill() throws IOException {
+                killed = true;
                 interrupt();
-                killed = true;
                 channel.close();
                 remoteWrapper.stopRemoteWrapper();

trunk/objectmq/test/faultToleranceTest/ClientTest.java

@@ -1 @@
-package clientToleranceTest;
+package faultToleranceTest;
 
 import static org.junit.Assert.assertEquals;

trunk/objectmq/test/faultToleranceTest/ServerTest.java

@@ -1 @@
-package clientToleranceTest;
+package faultToleranceTest;
 
 import java.util.Properties;

trunk/objectmq/test/multiThreadTest/Car.java

@@ -1 @@
-package test2;
+package multiThreadTest;
 
 import java.util.List;

trunk/objectmq/test/multiThreadTest/CarImpl.java

@@ -1 @@
-package test2;
+package multiThreadTest;
 
 import java.util.List;

trunk/objectmq/test/multiThreadTest/CarThread.java

@@ -1 @@
-package test2;
+package multiThreadTest;
 
 import java.util.ArrayList;

trunk/objectmq/test/multiThreadTest/ClientTest.java

@@ -1 @@
-package test2;
+package multiThreadTest;
 
 import java.util.Properties;

trunk/objectmq/test/multiThreadTest/Mobile.java

@@ -1 @@
-package test2;
+package multiThreadTest;
 
 import java.util.List;

trunk/objectmq/test/multiThreadTest/MobileImpl.java

@@ -1 @@
-package test2;
+package multiThreadTest;
 
 import java.util.ArrayList;

trunk/objectmq/test/multiThreadTest/MobileThread.java

@@ -1 @@
-package test2;
+package multiThreadTest;
 
 import omq.common.broker.Broker;

trunk/objectmq/test/multiThreadTest/Rim.java

@@ -1 @@
-package test2;
+package multiThreadTest;
 
 import java.io.Serializable;

trunk/objectmq/test/multiThreadTest/ServerTest.java

@@ -1 @@
-package test2;
+package multiThreadTest;
 
 import java.util.Properties;

trunk/objectmq/test/test/Client.java

@@ -25 +25 @@
 
         @AsyncMethod
-        public void sendContact(String contact) throws RemoteException;
+        public void addContact(String contact) throws RemoteException;
 }

trunk/objectmq/test/test/ClientImpl.java

@@ -53 +53 @@
         @Override
         @AsyncMethod
-        public void sendContact(String contact) throws RemoteException {
+        public void addContact(String contact) throws RemoteException {
                 if (!id.equalsIgnoreCase(contact) && !friendList.containsKey(contact)) {
                         Client client = (Client) Broker.lookup(contact, Client.class);