package omq.common.broker;

import java.io.IOException;
import java.net.URL;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Map;
import java.util.Properties;

import omq.Remote;
import omq.client.listener.ResponseListener;
import omq.client.proxy.Proxymq;
import omq.common.event.Event;
import omq.common.event.EventDispatcher;
import omq.common.event.EventWrapper;
import omq.common.util.OmqConnectionFactory;
import omq.common.util.ParameterQueue;
import omq.common.util.Serializer;
import omq.exception.InitBrokerException;
import omq.exception.RemoteException;
import omq.exception.SerializerException;
import omq.server.RemoteObject;

import org.apache.log4j.Logger;
import org.apache.log4j.xml.DOMConfigurator;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownListener;
import com.rabbitmq.client.ShutdownSignalException;

public class Broker {

	private static final Logger logger = Logger.getLogger(Broker.class.getName());

	private Connection connection;
	private Channel channel;
	private ResponseListener responseListener;
	private EventDispatcher eventDispatcher;
	private Serializer serializer;
	private boolean clientStarted = false;
	private boolean connectionClosed = false;
	private Properties environment = null;
	private Map<String, RemoteObject> remoteObjs;
	private Map<String, Object> proxies = new Hashtable<String, Object>();

	public Broker(Properties env) throws Exception {
		// Load log4j configuration
		URL log4jResource = Broker.class.getResource("/log4j.xml");
		DOMConfigurator.configure(log4jResource);

		remoteObjs = new HashMap<String, RemoteObject>();
		serializer = new Serializer(env);
		environment = env;
		connection = OmqConnectionFactory.getNewConnection(env);
		channel = connection.createChannel();
		addFaultTolerance();
		try {
			tryConnection(env);
		} catch (Exception e) {
			channel.close();
			connection.close();
			throw new InitBrokerException("The connection didn't work");
		}
	}

	public void stopBroker() throws Exception {
		logger.warn("Stopping broker");
		// Stop the client
		if (clientStarted) {
			responseListener.kill();
			eventDispatcher.kill();
			//TODO proxies = null; ??
		}
		// Stop all the remote objects working
		for (String reference : remoteObjs.keySet()) {
			unbind(reference);
		}

		// Close the connection once all the listeners are died
		closeConnection();

		clientStarted = false;
		connectionClosed = false;
		environment = null;
		remoteObjs = null;
		// Serializer.removeSerializers();
	}

	/**
	 * @return Broker's connection
	 * @throws Exception
	 */
	public Connection getConnection() throws Exception {
		return connection;
	}

	public void closeConnection() throws IOException {
		logger.warn("Clossing connection");
		connectionClosed = true;
		connection.close();
		connectionClosed = false;
	}

	/**
	 * 
	 * @return Broker's channel
	 * @throws Exception
	 */
	public Channel getChannel() throws Exception {
		return channel;
	}

	/**
	 * Creates a new channel using the Broker's connection
	 * 
	 * @return newChannel
	 * @throws IOException
	 */
	public Channel getNewChannel() throws IOException {
		return connection.createChannel();
	}

	@SuppressWarnings("unchecked")
	public synchronized <T extends Remote> T lookup(String reference, Class<T> contract) throws RemoteException {
		try {

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

			if (!proxies.containsKey(reference)) {
				Proxymq proxy = new Proxymq(reference, contract, this);
				Class<?>[] array = { contract };
				Object newProxy = Proxymq.newProxyInstance(contract.getClassLoader(), array, proxy);
				proxies.put(reference, newProxy);
				return (T) newProxy;
			}
			return (T) proxies.get(reference);

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

	public void bind(String reference, RemoteObject remote) throws RemoteException {
		try {
			remote.startRemoteObject(reference, this);
			remoteObjs.put(reference, remote);
		} catch (Exception e) {
			throw new RemoteException(e);
		}
	}

	public 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");
		}

	}

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

	}

	/**
	 * This method ensures the client will have only one ResponseListener and
	 * only one EventDispatcher. Both with the same environment.
	 * 
	 * @param environment
	 * @throws Exception
	 */
	private synchronized void initClient(Properties environment) throws Exception {
		if (responseListener == null) {
			responseListener = new ResponseListener(this);
			responseListener.start();
		}
		if (eventDispatcher == null) {
			eventDispatcher = new EventDispatcher(this);
			eventDispatcher.start();
		}
	}

	/**
	 * This method sends an event with its information
	 * 
	 * @param event
	 * @throws IOException
	 * @throws SerializerException
	 */
	public void trigger(Event event) throws IOException, SerializerException {
		String UID = event.getTopic();
		EventWrapper wrapper = new EventWrapper(event);
		logger.debug("EventTrigger fanout exchange: " + UID + " Event topic: " + UID + " Event corrID: " + event.getCorrId());
		channel.exchangeDeclare(UID, "fanout");

		byte[] bytesResponse = serializer.serialize(wrapper);
		channel.basicPublish(UID, "", null, bytesResponse);
	}

	/**
	 * This function is used to send a ping message to see if the connection
	 * works
	 * 
	 * @param env
	 * @throws Exception
	 */
	public void tryConnection(Properties env) throws Exception {
		Channel channel = connection.createChannel();
		String message = "ping";

		String exchange = env.getProperty(ParameterQueue.USER_NAME) + "ping";
		String queueName = exchange;
		String routingKey = "routingKey";

		channel.exchangeDeclare(exchange, "direct");
		channel.queueDeclare(queueName, false, false, false, null);
		channel.queueBind(queueName, exchange, routingKey);

		channel.basicPublish(exchange, routingKey, null, message.getBytes());

		QueueingConsumer consumer = new QueueingConsumer(channel);

		channel.basicConsume(queueName, true, consumer);
		Delivery delivery = consumer.nextDelivery(1000);

		channel.exchangeDelete(exchange);
		channel.queueDelete(queueName);

		channel.close();

		if (!message.equalsIgnoreCase(new String(delivery.getBody()))) {
			throw new IOException("Ping initialitzation has failed");
		}
	}

	/**
	 * 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 void addFaultTolerance() {
		connection.addShutdownListener(new ShutdownListener() {
			@Override
			public void shutdownCompleted(ShutdownSignalException cause) {
				logger.warn("Shutdown message received. Cause: " + cause.getMessage());
				if (!connectionClosed)
					if (cause.isHardError()) {
						if (connection.isOpen()) {
							try {
								connection.close();
							} catch (IOException e) {
								e.printStackTrace();
							}
						}
						try {
							connection = OmqConnectionFactory.getNewWorkingConnection(environment);
							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();
							}
						}
					}
			}
		});
	}

	public Properties getEnvironment() {
		return environment;
	}

	public ResponseListener getResponseListener() {
		return responseListener;
	}

	public EventDispatcher getEventDispatcher() {
		return eventDispatcher;
	}

	public Serializer getSerializer() {
		return serializer;
	}
}
