source: trunk/objectmq/src/omq/common/broker/Broker.java @ 18

package omq.common.broker;

import java.io.IOException;
import java.util.Properties;

import omq.Remote;
import omq.client.proxy.Proxymq;
import omq.client.remote.response.ResponseListener;
import omq.common.event.Event;
import omq.common.event.EventDispatcher;
import omq.common.event.EventWrapper;
import omq.common.util.Environment;
import omq.common.util.OmqConnectionFactory;
import omq.common.util.Serializer;
import omq.exception.EnvironmentException;
import omq.exception.RemoteException;
import omq.exception.SerializerException;
import omq.server.remote.request.RemoteObject;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Broker {
        private static Connection connection;
        private static Channel channel;
        private static boolean clientStarted = false;

        public static void initBroker(Properties env) throws Exception {
                try {
                        Environment.getEnvironment();
                } catch (EnvironmentException ex) { // environment not set.
                        Environment.setEnvironment(env);
                        connection = OmqConnectionFactory.getNewConnection(env);
                        channel = connection.createChannel();
                }
        }

        // TODO: what happens if the connection is not set
        public static Connection getConnection() throws Exception {
                return connection;
        }

        public static Channel getChannel() throws Exception {
                return channel;
        }

        public static Channel getNewChannel() throws IOException {
                return connection.createChannel();
        }

        public static Remote lookup(String reference, Class<?> contract) throws RemoteException {
                try {
                        Properties environment = Environment.getEnvironment();

                        if (!clientStarted) {
                                initClient(environment);
                                clientStarted = true;
                        }

                        if (!Proxymq.containsProxy(reference)) {
                                Proxymq proxy = new Proxymq(reference, contract, environment);
                                Class<?>[] array = { contract };
                                return (Remote) Proxymq.newProxyInstance(contract.getClassLoader(), array, proxy);
                        }
                        return (Remote) Proxymq.getInstance(reference);

                } catch (Exception e) {
                        throw new RemoteException(e);
                }
        }

        public static void bind(String reference, RemoteObject remote) throws RemoteException {
                try {
                        Properties environment = Environment.getEnvironment();
                        remote.start(reference, environment);
                } catch (Exception e) {
                        throw new RemoteException(e);
                }
        }

        public static void unbind(String reference) throws RemoteException {

        }

        public void rebind(String name, Remote obj) throws RemoteException {

        }

        private static void initClient(Properties environment) throws Exception {
                if (ResponseListener.isVoid()) {
                        ResponseListener.init(environment);
                }
                if (EventDispatcher.isVoid()) {
                        EventDispatcher.init(environment);
                }
        }
        
        
        public static void trigger(Event event) throws IOException, SerializerException{
                String UID = event.getTopic();
                EventWrapper wrapper = new EventWrapper(event);
                channel.exchangeDeclare(UID, "fanout");
                channel.basicPublish(UID, "", null, Serializer.serialize(wrapper));
        }

}