/**
 * JOnAS: Java(TM) Open Application Server
 * Copyright (C) 2007,2008 Bull S.A.S.
 * Contact: jonas-team@ow2.org
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
 *
 * --------------------------------------------------------------------------
 * $Id: JGroupsDiscoveryUtils.java 21631 2011-08-12 13:35:48Z benoitf $
 * --------------------------------------------------------------------------
 */

package org.ow2.jonas.discovery.jgroups.utils;

import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Proxy;
import java.net.URL;
import java.net.UnknownHostException;
import java.util.List;
import java.util.Map;

import javax.management.MBeanServer;

import org.jgroups.Address;
import org.jgroups.Channel;
import org.jgroups.ChannelClosedException;
import org.jgroups.ChannelException;
import org.jgroups.conf.ConfiguratorFactory;
import org.jgroups.conf.ProtocolConfiguration;
import org.jgroups.conf.ProtocolStackConfigurator;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;
import org.ow2.jonas.discovery.DiscoveryState;
import org.ow2.jonas.discovery.base.DiscHelper;
import org.ow2.jonas.discovery.base.comm.ClusterdDiscoveryEvent;
import org.ow2.jonas.discovery.base.comm.DiscEvent;
import org.ow2.jonas.discovery.base.comm.DiscGreeting;
import org.ow2.jonas.lib.bootstrap.JProp;
import org.ow2.jonas.lib.util.Log;
import org.ow2.util.cluster.jgroups.ConnectionManager;
import org.ow2.util.cluster.jgroups.IChannel;
import org.ow2.util.cluster.jgroups.JChannelWrapper;


public class JGroupsDiscoveryUtils extends DiscHelper {

    /**
     * Current logger.
     */
    private static Logger logger = Log.getLogger(Log.JONAS_DISCOVERY_PREFIX);
    /**
     * Used by greeting receiver and communication manager.
     * to handle duplicate server name exception
     */
    public static boolean duplicateExceptionName = false;

    /**
     * Group name property.
     */
    @SuppressWarnings("unused")
    private static final String DEFAULT_GROUP_NAME_KEY = "jonas.discovery.jgroups.group.name";
    /**
     * default discovery group name.
     */
    private static final String DAFAULT_GROUP_NAME = "jgroups-discovery";
    /**
     * default Jgroups discovery configuration file property name.
     */
    @SuppressWarnings("unused")
    public static final String JGROUPS_CONFIG_PROPERTY = "jonas.service.discovery.jgroups.conf";
    /**
     * Default file name for the stack.
     */
    public static final String DEFAULT_STACK_FILENAME = "jgroups-discovery.xml";
    /**
     * The jonas name.
     */
    public static final String DEFAULT_JONAS_NAME = "jonas";
    /**
     * The default domain name.
     */
    public static final String DEFAULT_DOMAIN_NAME = "jonas";
    /**
     * The default server Id.
     */
    public static final String DEFAULT_SERVER_ID = "jonas";
    /**
     * The default greeting port name.
     */
    public static final String DISCOVERY_GREETING_PORT_DEFAULT = "9080";
    /**
     * The default greeting Ip.
     */
    public static final String DISCOVERY_GREETING_IP = "localhost";
    /**
     * The JGroups discovery version.
     */
    public static final String DISCOVERY_PROTOCOL_VERSION = "1.3";
    /**
     * The default JGroups discovery timeout.
     */
    public static final String DISCOVERY_GREETING_TIMEOUT_DEFAULT = "5000";
    /**
     * The default key for discovery master started by JOnAS server.
     */
    public static final String DISCOVERY_IS_MASTER = "MASTER";
    /**
     * The default key for discovery slave started by JOnAS server.
     */
    public static final String DISCOVERY_IS_SLAVE = "SLAVE";
    /**
     * The default key for discovery master started by clusterd.
     */
    public static final String DISCOVERY_IS_CLUSTERD = "CLUSTERD";
    /**
     * The default key for failure detection protocol of JGroups stack.
     */
    private static final String FD_PROTOCOL = "FD";
    /**
     * The default key for failure detection timeout of JGroups stack.
     */
    private static final String FD_TIMEOUT = "timeout";
    /**
     * The default key for Ip ttl of JGroups stack.
     */
    private static final String COMM_TTL = "ip_ttl";
    /**
     * The default key for multicast address of JGroups stack.
     */
    private static final String COMM_MCAST_ADDRESS = "mcast_addr";
    /**
     * The default key for multicast port of JGroups stack.
     */
    private static final String COMM_MCAST_PORT = "mcast_port";
    /**
     * Default group name for discovery channel.
     */
    public static final String DEF_GROUP_NAME = "JGroupsDiscovery";

    /**
     * Channel wrapper.
     */
    private IChannel channel = null;
    /**
     * Unique instance of discovery utils.
     */
    private static DiscoveryUtils  unique = null;
    /**
     * Gets the channel
     * @return the channel.
     */
    public IChannel getChannel() {
        return channel;
    }

    /**
     * Sets the channel
     * @param channel the channel to set
     */
    public void setChannel(final IChannel channel) {
        this.channel = channel;
    }
    /**
     * Private connstructor.
     */
    private JGroupsDiscoveryUtils() {
    }


    /**
     * Initializes properties of the discovery(groupname, stackfile, channel...).
     * @param srvName The host name of discovery.
     * @param domain The domain name of discovery's host.
     * @param discoveryType The discovery type.
     * @param confFileName The file name of JGroups stack.
     * @param discoveryGroup The group of JGroups discovery (channel connection)
     * @param beanServer The Mbean server of discovery's host.
     * @param connUrls Connectors urls of discovery hosts.
     */
    public static void init(final String srvName, final String domain, final String discoveryType, final String confFileName, final String discoveryGroup, final MBeanServer beanServer, final String[] connUrls, final long reconnectionTimeout) throws Throwable {
        try {
            unique = (unique == null) ? new DiscoveryUtils() : unique;
            unique.setGroupName(discoveryGroup);
            unique = new DiscoveryUtils();
            unique.setUrls(connUrls);
            unique.setJonasName(srvName);
            unique.setDomainName(domain);
            /**
             * url is the best Id.
             */
            unique.setServerId(connUrls.toString());
            unique.setDiscType(discoveryType);
            URL confile = null;
            //create the JChannel
            try {
                //get the communication protocol config
                 File conf = new File(JProp.getJonasBase(), "conf");
                 confile = new File(conf, confFileName).toURL();
                if (confile == null) {
                    logger.log(BasicLevel.DEBUG, " Unable to get JGroups stack file named: " + confFileName + " from JONAS_BASE "+ JProp.getJonasBase()+" using JProp ");
                    confile = ClassLoader.getSystemClassLoader().getResource(confFileName);
                    if (confile == null) {
                        logger.log(BasicLevel.DEBUG, "Unable to get JGroups stack file named: " + confFileName + " from class loader: " + ClassLoader.getSystemClassLoader().getClass());
                         confile = Thread.currentThread().getContextClassLoader().getResource(confFileName);
                         if (confile == null) {
                             logger.log(BasicLevel.DEBUG, " Unable to get JGroups stack file named: " + confFileName + " from class loader: " + Thread.currentThread().getContextClassLoader().getClass());
                             logger.log(BasicLevel.DEBUG, " Load default stack file : " + DEFAULT_STACK_FILENAME + " with class loader " + JGroupsDiscoveryUtils.class.getClassLoader().getClass().getName());
                             confile = JGroupsDiscoveryUtils.class.getResource(unique.getConfigurationFile());
                             logger.log(BasicLevel.INFO, "Default JGroups stack file retrieved successfully");
                         }else {
                             logger.log(BasicLevel.DEBUG, " JGroups stack file named: " + confFileName + " successfully retrieved from class loader: "+Thread.currentThread().getContextClassLoader().getClass());
                        }
                    }else {
                        logger.log(BasicLevel.DEBUG, " JGroups stack file named: " + confFileName + " successfully retrieved from class loader: "+ClassLoader.getSystemClassLoader().getClass());
                    }

                }else {
                    logger.log(BasicLevel.DEBUG, "JGroups stack file named: " + confFileName + " successfully retrieved from JONAS_BASE "+ JProp.getJonasBase()+" using JProp ");
                }
                unique.setConfigurationFile(confile.getFile());
            } catch (Throwable e1) {
                e1.printStackTrace();
                logger.log(BasicLevel.ERROR, "Unable to get communication protocol configuration: ", e1);
                throw e1;
            }
            try {
            unique.setStackConfigurator(ConfiguratorFactory.getStackConfigurator(unique.getConfigurationFile()));
            } catch (Exception e) {
                logger.log(BasicLevel.ERROR, " Unable to get JGroups stack configuration: \n", e);
                throw e;
            }

            try {
                IChannel channel = new JChannelWrapper(unique.getStackConfigurator().getProtocolStackString());
                // Avoid our own messages
                channel.setOpt(Channel.LOCAL, false);
                // Set the auto-reconnect option for enabling a node to leave and re-join the cluster
                channel.setOpt(Channel.AUTO_RECONNECT, true);

                InvocationHandler invocationHandler =
                    new ConnectionManager(reconnectionTimeout, channel, IDiscoveryChannel.class);
                channel.addChannelListener((ConnectionManager)invocationHandler);
                IChannel chan = (IChannel)Proxy.newProxyInstance(
                        IChannel.class.getClassLoader(), new Class[] {IChannel.class}, invocationHandler);
                unique.setComChannel(chan);
                logger.log(BasicLevel.DEBUG, " JGroups channel created from file " + confFileName + " successfully connected: \n");
            } catch (Exception e) {
                logger.log(BasicLevel.ERROR, " Unable to connect JGroups channel: \n", e);
                throw e;
            }


            try {
                getCommProtocolConf(unique.getStackConfigurator());
            } catch (Exception e) {
                logger.log(BasicLevel.ERROR, " Unable to get communication protocol \n", e);
                throw e;
            }
            try {
                //gets greeting ack timeout
                unique.setGreetingAckTimeOut(new Integer((String)getParamValueForProtoParam(getProtocolParam(unique.getStackConfigurator(), FD_PROTOCOL), FD_TIMEOUT)));
            } catch (Exception e) {
                logger.log(BasicLevel.DEBUG, " Unable to get greeting ack timeout from JGroups configuration file \n");
                logger.log(BasicLevel.INFO, " Greeting ack timeout is set to default value :" + DISCOVERY_GREETING_TIMEOUT_DEFAULT);
            }
            try {
                unique.setTtl(new Integer((getParamValueForCommParam(COMM_TTL))));
            } catch (Exception e) {
                logger.log(BasicLevel.DEBUG, " Unable to get IP ttl from stack configuration \n");
                logger.log(BasicLevel.INFO, " IP ttl is set to " + unique.getTtl());
            }
        } catch (Exception e) {
            logger.log(BasicLevel.ERROR, " Cannot get file named " + DEFAULT_STACK_FILENAME + " in current environment");
            return;
        }
    }
    /**
     * Connect the channel.
     * @throws ChannelException
     */
    public static void connectChannel() throws ChannelException {
        unique.getComChannel().connect(unique.getGroupName());
    }

    /**
     * dDsconnect the channel.
     * @throws ChannelException
     */
    public static void disconnectChannel() throws ChannelException {
        unique.getComChannel().close();
    }
    /**
     * Create a discovery greeting message.
     * @param startup -
     *            true: means that is a greeting message at the point of start
     *            up of the discovery service. false: means this is a reply to
     *            a previous greeting message.
     * @return a Discovery Greeting message.
     * @throws UnknownHostException If host is unknown
     */
    public static DiscGreeting createDiscGreeting(final boolean startup) throws UnknownHostException {
        return new DiscGreeting(unique.getComChannel().getLocalAddress().toString(), unique.getDiscPort(),
                unique.getJonasName(), unique.getDomainName(), startup, unique.getServerId());
    }

    /**
     * Gets configuration for communication protocol(UDP || TCP).
     * @param stackconfigurator The stack configuration.
     * @throws Exception Any.
     */
    private static void getCommProtocolConf(final ProtocolStackConfigurator stackconfigurator) throws Exception {
        List<ProtocolConfiguration> proto = stackconfigurator.getProtocolStack();
        try {
            boolean stop = false;
            int i = 0;
            while (!stop && i < proto.size()) {
                String protoName = proto.get(i).getProtocolName();
                if (protoName.equals("UDP") || protoName.equals("TCP")) {
                    stop = true;
                    unique.setCommProtoName(protoName);
                    // hashMap containing configuration for communication proto
                    unique.setCommProtoProperties(proto.get(i).getProperties());
                } else {
                    i++;
                }
            }

        } catch (Exception e) {
            throw new Exception("Unable to get value for parameter. Current stack is: " + stackconfigurator
                    + " Following exceptions occurred " + e);
        }
    }
    /**
     * Gets parameters for given stack protocol.
     * @param stackconfigurator The JGroups stack configuration
     * @param protocolName The name of the protocol to get.
     * @return Given protol parameters.
     * @throws Exception Any.
     */
    private static Map<String, String> getProtocolParam(final ProtocolStackConfigurator stackconfigurator, final String protocolName) throws Exception {
        List<ProtocolConfiguration> proto = stackconfigurator.getProtocolStack();
        Map<String, String> protoConf = null;
        try {
            boolean stop = false;
            int i = 0;
            while(!stop && i < proto.size()){
                String protoName = proto.get(i).getProtocolName();
                if (protoName.equals(protocolName)) {
                    stop = true;
                    //hashMap containing configuration for communication proto
                    protoConf = proto.get(i).getProperties();
                }else {
                    i++;
                }
            }
            if (!stop) {
                throw new Exception("The " + protocolName + " protocol is not defined in jgroups stack. Current stack configuration is: "+stackconfigurator);
            }

        } catch (Exception e) {
            throw new Exception("Unable to get configuration for " + protocolName + " protocol. Current stack is " + stackconfigurator + ". Following exception occurred "+e);
        }
        return protoConf;
    }

    /**
     * Gets the value of given communication protocol parameter. Communication protol is more often used than the others.
     * That makes this method very usefull.
     * @param paramName The key of communication parameter.
     * @return The value for given parameter.
     * @throws Exception Any.
     */
    private static String getParamValueForCommParam(final String paramName) throws Exception {
        String ret = unique.getCommProtoProperties().get(paramName);
        if (ret == null) {
            logger.log(BasicLevel.DEBUG, "Cannot get value for communication parameter " + paramName);
            throw new Exception("Cannot get value for communication parameter " + paramName);
        }
        return ret;
    }

    /**
     * Gets Value for a given parameter of JGroups stack protocol.
     * @param protoConf Configuration of the given protocol
     * @param paramName Get the value for this parameter.
     * @throws Exception An exception, if the parameter is not found in given configuration.
     */
    private static Object getParamValueForProtoParam(final Map<String, String> protoConf, final String paramName ) throws Exception {
        String ret = protoConf.get(paramName);
        if ( ret == null) {
            logger.log(BasicLevel.DEBUG, " Can't get value for communication parameter " + paramName);
            throw new Exception(" Can't get value for parameter " + paramName);
        }
        return ret;
    }

    /**
     * Convert network datas into a discovey communication object.
     * @param data
     * @return <code>DiscGreeting || DiscMessage || DiscEvent</code>
     * @throws IOException
     * @throws ClassNotFoundException
     */
    public static Object bytesToObject(final Data data) throws IOException, ClassNotFoundException {
        if (data.getType() == Data.DISC_GREETING) {
            return data.getPayload();
        } else if (data.getType() == Data.DISC_EVENT) {
            return data.getPayload();

        } else if (data.getType() == Data.DISC_MESSAGE) {
            return data.getPayload();
        } else{
            logger.log(BasicLevel.DEBUG, "Undefined type of message received by JGroups discovery \n");
            logger.log(BasicLevel.DEBUG, "Expected type are\n "
                    +Data.DISC_GREETING+": DISC_GREETING\n"
                    +Data.DISC_EVENT+": DISC_EVENT\n"
                    +Data.DISC_MESSAGE+": DISC_MESSAGE\n But received: "+data.getType());
        }
        return null;
    }

    /**
     * Construct a byte[] containing type info about a DiscMessage object to be sent,
     * plus the content of this message.
     * @param obj Object to be send. Only supported DiscMessage objects.
     * @return Null if the object is not an instance of DiscMessage or one of its subclasses.
     * @throws IOException Could not create an ObjectOutputStream to write into the
     * underlying ByteArrayOutputStream.
     */
    public static Data objectToBytes(final Serializable obj) throws IOException {
        if (obj instanceof DiscEvent) {
            return new Data(Data.DISC_EVENT, obj);
        }
        if (obj instanceof DiscGreeting) {
            return new Data(Data.DISC_GREETING, obj);
        }
        return new Data(Data.DISC_MESSAGE, obj);
    }

    /**
     * Create a discovery event to notify about a state change of the
     * event sender
     * @param state
     *         - RUNNING if the sender notifies that it gets running
     *         - STOPPING if the sender notifies that it stops running
     * @return a Discovery event (notification)
     * @throws Exception
     *           is thrown if the jmx service is not reached.
     */
    public static DiscEvent createNotifMessage(final String state ) throws Exception {
        logger.log(BasicLevel.DEBUG, "Creating a discovery notification with state:" + state + " \n");
        if (!state.equals(DiscoveryState.RUNNING)&& !state.equals(DiscoveryState.STARTUP)) {
            unique.setUrls(null);
        }
        // In the case of a notification, the field port is not important since the
        // notifier is not waiting for an acknowledfement.
        DiscEvent resp = new DiscEvent(unique.getLocalAddress().getIpAddress().toString(), unique.getDiscPort(), unique.getJonasName(), unique.getDomainName(), unique.getServerId(), unique.getUrls(), unique.getDiscType().equals(DISCOVERY_IS_MASTER));
        resp.setState(state);
        logger.log(BasicLevel.DEBUG, " Discovery notification successfully created. \n");
        return resp;
    }





    /**
     * Create a discovery event to notify about a state change of the
     * event sender
     * @param state
     *         - RUNNING if the sender notifies that it gets running
     *         - STOPPING if the sender notifies that it stops running
     * @return a Discovery event (notification)
     * @throws Exception
     *           is thrown if the jmx service is not reached.
     */
    public static DiscEvent createNotifMessageForClusterd(final String state ) throws Exception {
        logger.log(BasicLevel.DEBUG, "Cluster daemon is creating a discovery notification on startup \n");
        if (!state.equals(DiscoveryState.RUNNING) && !state.equals(DiscoveryState.STARTUP)) {
            unique.setUrls(null);
        }
        // In the case of a notification, the field port is not important since the
        // notifier is not waiting for an acknowledfement.
        ClusterdDiscoveryEvent resp = new ClusterdDiscoveryEvent(unique.getLocalAddress().getIpAddress().toString(), unique.getDiscPort(), unique.getJonasName(), unique.getDomainName(), unique.getServerId(), unique.getUrls());
        resp.setState(state);
        resp.setDiscoveryMaster(unique.getDiscType().equals(DISCOVERY_IS_MASTER));
        logger.log(BasicLevel.DEBUG, "Cluster daemon has successfully created a discovery notification on startup \n");
        return resp;
    }


    /**
     * Sends a given message to the given dest.
     * @param dest The message destination.
     * @param msg The message to send.
     * @throws Exception Any.
     */

    public synchronized static void send(final Address dest, final Serializable msg) throws Exception {
        String destString = null;
        try {
            if (dest == null) {
                destString = " All members ";
            }else {
                destString = dest.toString();
            }
            //send it on the given address
            //after transforming the object to a datagram
            if (logger.isLoggable(BasicLevel.DEBUG)) {
                logger.log(BasicLevel.DEBUG, " My name is " + unique.getJonasName() + " Sending a message to " + destString);
            }

            //send a discovery event
            unique.getComChannel().send(dest, unique.getLocalAddress(), msg);
        } catch (ChannelClosedException e) {
            logger.log(BasicLevel.DEBUG, " Channel closed report when sending a message to " + destString + "\n", e);
            throw new Exception(" Channel closed report when sending a message to  " + destString + "\n" + e);
        }
    }
    /**
     * Close JGroups channel.
     */
    public static void closeChannel() {
        if(unique.getComChannel() != null && unique.getComChannel().isOpen()) {
            //destroy all channel resources (threads, etc..)
            unique.getComChannel().close();
        }
    }

    /**
     * Gets Multicast address.
     * @return Multicast address
     * @throws Exception
     */
    public static String getMulticastAddress() throws Exception {
        return getParamValueForCommParam(COMM_MCAST_ADDRESS);

    }

    /**
     * Gets Multicast port.
     * @return Multicast address
     * @throws Exception
     */
    public static String getMulticastPort() throws Exception {
        return getParamValueForCommParam(COMM_MCAST_PORT);

    }

    /**
     * Gets Discovery utils instance.
     * @return
     */
    public static DiscoveryUtils getInstance() {
        return unique;
    }
}
