source: branches/supervisor/src/main/java/omq/common/broker/RemoteBrokerImpl.java @ 91

package omq.common.broker;

import java.io.IOException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Properties;
import java.util.Set;

import omq.common.util.ParameterQueue;
import omq.exception.RemoteException;
import omq.server.RemoteObject;
import omq.server.RemoteWrapper;

import com.rabbitmq.client.QueueingConsumer;

/**
 * 
 * @author Sergi Toda <sergi.toda@estudiants.urv.cat>
 * 
 */
public class RemoteBrokerImpl extends RemoteObject implements RemoteBroker {

        /**
         * 
         */
        private static final long serialVersionUID = 1L;

        // fanout broker
        private String brokerSet;
        // id broker
        private String brokerName;

        public void startRemoteBroker(String brokerSet, String brokerName, Broker broker, Properties env) throws Exception {
                this.broker = broker;
                this.UID = brokerName;
                this.env = env;
                this.brokerSet = brokerSet;
                this.brokerName = brokerName;

                this.params = new HashMap<String, List<Class<?>>>();
                for (Method m : this.getClass().getMethods()) {
                        List<Class<?>> list = new ArrayList<Class<?>>();
                        for (Class<?> clazz : m.getParameterTypes()) {
                                list.add(clazz);
                        }
                        this.params.put(m.getName(), list);
                }

                // Get num threads to use
                int numThreads = Integer.parseInt(env.getProperty(ParameterQueue.NUM_THREADS, "1"));
                this.remoteWrapper = new RemoteWrapper(this, numThreads, broker.getSerializer());

                startQueues();

                // Start this listener
                this.start();
        }

        private void startQueues() throws Exception {
                /*
                 * Unique queue
                 */
                channel.exchangeDeclare(brokerSet, "direct");
                channel.queueDeclare(brokerName, false, true, true, null);
                channel.queueBind(brokerName, brokerSet, brokerName);

                /*
                 * Multi queue
                 */
                channel.exchangeDeclare("multi#" + brokerSet, "fanout");
                channel.queueDeclare("multi#" + brokerName, false, true, true, null);
                channel.queueBind("multi#" + brokerName, "multi#" + brokerSet, "");

                /*
                 * Consumer
                 */

                consumer = new QueueingConsumer(channel);
                channel.basicConsume(brokerName, true, consumer);
                channel.basicConsume(brokerName + "#multi", true, consumer);
        }

        @Override
        public Set<String> getRemoteObjects() {
                return this.broker.getRemoteObjs().keySet();
        }

        @Override
        public void spawnObject(String reference, String className, Class<?> parameterTypes, Object... args) throws Exception {
                RemoteObject remote = (RemoteObject) Class.forName(className).getConstructor(parameterTypes).newInstance(args);
                this.broker.bind(reference, remote);
        }

        @Override
        public void deleteObject(String reference) throws RemoteException, IOException {
                this.broker.unbind(reference);
        }

        @Override
        public HasObject hasObject(String reference) {
                boolean hasIt = this.broker.getRemoteObjs().containsKey(reference);
                return new HasObject(this.brokerName, reference, hasIt);
        }

}