/*
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 *
 * Copyright 2000-2009 Sun Microsystems, Inc. All rights reserved. 
 *
 * The contents of this file are subject to the terms of either the GNU
 * General Public License Version 2 only ("GPL") or the Common Development
 * and Distribution License ("CDDL") (collectively, the "License").  You may
 * not use this file except in compliance with the License.  You can obtain
 * a copy of the License at https://glassfish.dev.java.net/public/CDDL+GPL.html
 * or mq/legal/LICENSE.txt.  See the License for the specific language
 * governing permissions and limitations under the License.
 * 
 * When distributing the software, include this License Header Notice in each
 * file and include the License file at mq/legal/LICENSE.txt.  Sun designates
 * this particular file as subject to the "Classpath" exception as provided by
 * Sun in the GPL Version 2 section of the License file that accompanied this
 * code.  If applicable, add the following below the License Header, with the
 * fields enclosed by brackets [] replaced by your own identifying information:
 * "Portions Copyrighted [year] [name of copyright owner]"
 * 
 * Contributor(s):
 * 
 * If you wish your version of this file to be governed by only the CDDL or
 * only the GPL Version 2, indicate your decision by adding "[Contributor]
 * elects to include this software in this distribution under the [CDDL or GPL
 * Version 2] license."  If you don't indicate a single choice of license, a
 * recipient has the option to distribute your version of this file under
 * either the CDDL, the GPL Version 2 or  to extend the choice of license to
 * its licensees as provided above.  However, if you add GPL Version 2 code
 * and therefore, elected the GPL Version 2 license, then the option applies
 * only if the new code is made subject to such option by the copyright holder. 
 */

/*
 * @(#)ClusterDiscoveryService.java	1.11 06/29/07
 */ 

package com.sun.messaging.jmq.jmsserver.service;

import java.io.*;
import java.net.*;
import java.util.Properties;
import java.util.Hashtable;
import java.util.Enumeration;
import javax.net.ServerSocketFactory;

import com.sun.messaging.jmq.io.ServiceTable;
import com.sun.messaging.jmq.io.ServiceEntry;
import com.sun.messaging.jmq.io.ClusterDiscoveryProtocol;
import com.sun.messaging.jmq.util.log.Logger;
import com.sun.messaging.jmq.net.MQServerSocketFactory;
import com.sun.messaging.jmq.jmsserver.config.*;
import com.sun.messaging.jmq.jmsserver.Globals;
import com.sun.messaging.jmq.jmsserver.resources.*;
import com.sun.messaging.jmq.jmsserver.core.BrokerAddress;

public class ClusterDiscoveryService implements Runnable, ConfigListener {
    public static final int CLUSTER_DISCOVERY_PORT = 0;
    private static final String PORT_PROPERTY =
        Globals.IMQ + ".cluster_discovery.port";
    private static final String BACKLOG_PROPERTY =
        Globals.IMQ + ".cluster_discovery.backlog";
    private static final String SERVICE_NAME = "cluster_discovery";
    private static boolean DEBUG = false;

    private static Logger logger = null;
    private static BrokerResources rb = null;
    private static BrokerConfig bc = null;

    private static ServerSocketFactory ssf = MQServerSocketFactory.getDefault();

    private ServiceTable st = null;
    private ServerSocket serverSocket = null;
    private String hostname = "???";
    private int port = 0;
    private int backlog = 100;

    /**
     * Create a cluster discovery service handler.
     */
    public ClusterDiscoveryService(String instance, String hostname) {
        if (DEBUG) {
            logger.log(Logger.DEBUG,
                "Creating cluster discovery service.");
        }

        this.hostname = hostname;
        if (hostname == null) {
            try {
                this.hostname = InetAddress.getLocalHost().getHostName();
            }
            catch (Exception e) {
                // If this happens it is a coding error. No need to I18N
                logger.logStack(Logger.WARNING,
                    "ClusterDiscoveryService: initialization error",
                    e);
            }
        }
        st = new ServiceTable();
        st.setBrokerInstanceName(instance);
        st.setBrokerVersion(Globals.getVersion().getProductVersion());
        logger = Globals.getLogger();
        rb = Globals.getBrokerResources();

        bc = Globals.getConfig();
        bc.addListener(PORT_PROPERTY, this);
        bc.addListener(BACKLOG_PROPERTY, this);
    }

    private String addressString(BrokerAddress b) {
        String addr = null;
        if (b != null)
            addr = "portmapper@" + b.toConfigString();

        return addr;
    }

    /**
     * Remember the current active broker.
     */
    public void setActiveBroker(BrokerAddress b) {
        if (DEBUG) {
            logger.log(Logger.DEBUG,
                "ClusterDiscoveryService: active broker = " + b);
        }

        st.setActiveBroker(addressString(b));
    }

    /**
     * Add this broker's address to the remote service list.
     */
    public void addSelfAddress(BrokerAddress b) {
        addRemoteService(b);
    }

    /**
     * Add a remote service address.
     */
    public void addRemoteService(BrokerAddress b) {
        if (DEBUG) {
            logger.log(Logger.DEBUG,
                "ClusterDiscoveryService: New remote service = " + b);
        }

        st.addRemoteService(addressString(b));
    }

    /**
     * Remove a remote service address.
     */
    public void removeRemoteService(BrokerAddress b) {
        if (DEBUG) {
            logger.log(Logger.DEBUG,
                "ClusterDiscoveryService: Removing remote service = " + b);
        }

        st.removeRemoteService(addressString(b));
    }

    /**
     * Add a service.
     *
     * @param   name    Name of service
     * @param   protocl Transport protocol of service ("tcp", "ssl", etc)
     * @param   type    Service type (NORMAL, ADMIN, etc)
     * @param   address String representation of the service address.
     */
    public synchronized void addService(String name, 
        String protocol, String type, String address) {
        ServiceEntry se = new ServiceEntry();

        se.setName(name);
        se.setProtocol(protocol);
        se.setType(type);
        se.setAddress(address);
        addService(name, se);
    }

    /**
     * update the address for a service.
     * @param   name    Name of service
     * @param   address String representation of the service address.
     */
    public synchronized void updateServiceAddress(String name,
        String address) {
        if (DEBUG) {
            logger.log(Logger.DEBUG,
                "ClusterDiscoveryService: Changing service:" +
                "\n\tname = " + name +
                "\n\taddress = " + address);
        }

        ServiceEntry se = st.get(name);
        if (se != null) {
            se.setAddress(address);
        }
    }

    /**
     * Add a service.
     *
     * @param   se    A ServiceEntry containing all information for
     *                the service.
     */
    public synchronized void addService(String name, ServiceEntry se) {
        if (DEBUG) {
            logger.log(Logger.DEBUG, "Adding service :" +
                "\n\tname = " + se.getName() +
                "\n\ttype = " + se.getType() +
                "\n\tprotocol = " + se.getProtocol() +
                "\n\taddress = " + se.getAddress());
        }
        st.add(se);
    }

    /**
     * Remove a service.
     */
    public synchronized void removeService(String name) {
        if (DEBUG) {
            logger.log(Logger.DEBUG, "Removing service : " + name);
        }
        st.remove(name);
    }

    /**
     * Add a service.
     *
     * @param   name    Name of service
     * @param   protocl Transport protocol of service ("tcp", "ssl", etc)
     * @param   type    Service type (NORMAL, ADMIN, etc)
     * @param   port    Port number.
     */
    public synchronized void addService(String name,
        String protocol, String type, int port, String host) {
        if (host == null)
            host = hostname;
        String address = name + '@' + host + ":" + port;
        addService(name, protocol, type, address);
    }

    /**
     * update the address for a service.
     * @param   name    Name of service
     * @param   port    Port service is runningon
     */
    public synchronized void updateServiceAddress(String name,
        String host, int port) {
        if (host == null)
            host = hostname;
        String address = name + '@' + host + "-" + port;
        updateServiceAddress(name, address);
    }

    /**
     * Get a hashtable containing information for each service.
     * This hashtable is indexed by the service name, and the values
     * are PortMapperEntry's
     */
    public synchronized Hashtable getServices() {
        return st.getServices();
    }

    public synchronized String toString() {
        return st.toString();
    }

    private String selfAddress() {
        if (serverSocket != null)
            return "cluster_discovery@" + hostname + ":" +
                serverSocket.getLocalPort();

        return "cluster_discovery@" + hostname + ":" + port;
    }

    private static void sleep(int nseconds) {
        try {
            Thread.sleep(nseconds * 1000L);
        } catch (InterruptedException e) {
            /* sleep got interrupted exception */
        }
    }

    public void init() throws PropertyUpdateException {
        String value;

        // Configure port number and backlog
        value = (String)bc.getProperty(PORT_PROPERTY);
        if (value != null) {
            validate(PORT_PROPERTY, value);
            update(PORT_PROPERTY, value);
        }

        value = (String)bc.getProperty(BACKLOG_PROPERTY);
        if (value != null) {
            validate(BACKLOG_PROPERTY, value);
            update(BACKLOG_PROPERTY, value);
        }
    }

    public void run() {
        Socket connection = null;

        if (serverSocket == null) {
            createServerSocket(this.port);
        }

        while (true) {
            if (serverSocket == null) {
                logger.log(Logger.ERROR,
                    "Could not start cluster discovery service. Exiting.");
                return;
            }

            try {
                connection = serverSocket.accept();
            }
            catch (SocketException e) {
                if (e instanceof BindException ||
                    e instanceof ConnectException ||
                    e instanceof NoRouteToHostException) {
                    logger.log(Logger.ERROR,
                        "Cluster discovery exception. Continuing.", e);
                    // We sleep in case the exception is not repairable. This
                    // prevents us from a tight loop.
                    sleep(1);
                } else {
                    // Serversocket was closed. Should be because something
                    // like the port number has changed. Try to recreate
                    // the server socket.
                    try {
                        // Make sure it is closed
                        serverSocket.close();
                    } catch (IOException ioe) { }

                    createServerSocket(this.port);
                }
                continue;
            } catch (IOException e) {
                logger.log(Logger.ERROR,
                    "Cluster discovery exception. Continuing.", e);
                sleep(1);
                continue;
            }

            // Got a connection.
            try {
                InputStream is = connection.getInputStream();
                OutputStream os = connection.getOutputStream();

                Properties p = ClusterDiscoveryProtocol.receiveRequest(is);
                String req = p.getProperty(
                    ClusterDiscoveryProtocol.OPERATION_TYPE);

                if (req.equals("REQUEST")) {
                    ClusterDiscoveryProtocol.sendResponse(st, os);
                }
                else {
                    logger.log(Logger.ERROR,
                        "Unknown cluster discovery request. Continuing.");
                }

                os.close();
                is.close();
                connection.close();
            }
            catch (IOException e) {
                logger.log(Logger.ERROR,
                    "Cluster discovery exception. Continuing.", e);
            }
            finally {
                try {
                    connection.close();
                } catch (IOException e) {}
            }
        }
    }

    private void createServerSocket(int port)  {
        try {
	    serverSocket = ssf.createServerSocket(port, backlog);

            Globals.getPortMapper().addService(SERVICE_NAME,
                "tcp", "CLUSTER_DISCOVERY",
                serverSocket.getLocalPort(), null);
            addService(SERVICE_NAME,
                "tcp", "CLUSTER_DISCOVERY", selfAddress());
        } catch (BindException e) {
            logger.log(Logger.ERROR, rb.E_BROKER_PORT_BIND, 
                SERVICE_NAME, new Integer(port));
            serverSocket = null;
            return;
        } catch (IOException e) {
            logger.log(Logger.ERROR, rb.E_BAD_SERVICE_START, 
                SERVICE_NAME, new Integer(port), e);
            serverSocket = null;
            return;
        }

        Object[] args = {SERVICE_NAME,
                         "tcp [ " + port + ", " + backlog + " ]",
                          new Integer(1), new Integer(1)};
        logger.log(Logger.INFO, rb.I_SERVICE_START, args);
    }

    /**
     * Change the service's port.
     */
    public synchronized void setPort(int port) {
        if (port == this.port) {
            return;
        }

        this.port = port;

        if (serverSocket != null) {
            // If there is a server socket close it so we create a new
            // one with the new port
            try {
                serverSocket.close();
            } catch (IOException e) {}
        }
    }

    public int getPort() {
        return port;
    }

    /**
     * Set the backlog parameter on the socket the portmapper is running on
     */
    public synchronized void setBacklog(int backlog) {
        this.backlog = backlog;

        if (serverSocket != null) {
            // If there is a server socket close it so we create a new
            // one with the new backlog
            try {
                serverSocket.close();
            } catch (IOException e) {
            }
        }
    }

    public void validate(String name, String value) 
        throws PropertyUpdateException {

        if (!name.equals(PORT_PROPERTY) && !name.equals(BACKLOG_PROPERTY)) {
            throw new PropertyUpdateException(
                rb.getString(rb.X_BAD_PROPERTY, name));
        }

        // Will throw an exception if integer value is bad
        int n = getIntProperty(name, value);

        if (name.equals(PORT_PROPERTY)) {
            if (n == this.port)  {
                return;
            }

            // Check if we will be able to bind to this port
            try {
                canBind(n);
            } catch (BindException e) {
                throw new PropertyUpdateException(
                    rb.getString(rb.E_BROKER_PORT_BIND, SERVICE_NAME, value));
            } catch (IOException e) {
                throw new PropertyUpdateException(
                    rb.getString(rb.E_BAD_SERVICE_START, SERVICE_NAME, value) +
                    e.toString());
            }
        }
    }

    public boolean update(String name, String value) {
        try {
            if (name.equals(PORT_PROPERTY)) {
                setPort(getIntProperty(name, value));
            } else if (name.equals(BACKLOG_PROPERTY)) {
                setBacklog(getIntProperty(name, value));
            }
        } catch (PropertyUpdateException e) {
            // Should never happen since this will be caught by validate
        }
        return true;
    }

    /**
     * Check if you can open a server socket on the specified port.
     * Throws an IOException if you can't
     */
    private static void canBind(int port) throws IOException {

        ServerSocket ss = null;
	ss = ssf.createServerSocket(port);
        ss.close();
        return;
    }

    private int getIntProperty(String name, String value)
                        throws PropertyUpdateException  {
        try {
            return Integer.parseInt(value);
        } catch (NumberFormatException e) {
            throw new PropertyUpdateException(
                rb.getString(rb.X_BAD_PROPERTY_VALUE, name + "=" + value));
        }
    }
}

/*
 * EOF
 */
