/*
 * 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. 
 */

/*
 * @(#)ClusterDiscovery.java	1.7 06/27/07
 */ 

package com.sun.messaging.jmq.io;

import java.io.*;
import java.net.*;
import java.util.*;

/**
 * This class implements the cluster discovery service client.
 *
 * The iMQ client runtime should use one instance of this class for
 * every connection factory object.
 */
public class ClusterDiscovery {
    public static final long RETRY_SLEEP_INTERVAL = 5000;

    private long retrySleepInterval = RETRY_SLEEP_INTERVAL;
    private ServiceTable st = null;
    private Vector addrList = null;
    private Hashtable failsafeAddrList = null;
    private int nextBroker = 0; // index into the addrList.

    /**
     * Create a new cluster discovery factory object.
     */
    public ClusterDiscovery() {
        addrList = null;
        st = null;
    }

    /**
     * Get the current retry sleep interval.
     *
     * This attribute is used by the findService search algorithm.
     * The findService method repeatedly scans all the known cluster
     * discovery addresses (i.e. brokers) until it finds a suitable
     * service. The retrySleepInterval attribute specifies the sleep
     * period after every unsuccessful iteration (so that we don't
     * keep spinning when all the brokers are down).
     */
    public long getRetrySleepInterval() {
        return retrySleepInterval;
    }

    /**
     * Set the retry sleep interval.
     */
    public void setRetrySleepInterval(long retrySleepInterval) {
        this.retrySleepInterval = retrySleepInterval;
    }

    /**
     * Broker address list initialization.
     *
     * This method is called by the client to specify the initial
     * broker address list (based on the connection factory
     * information). This must be done before the client can call
     * ClusterDiscovery.findService() method.
     *
     * Client can specify more than one broker addresses by calling
     * this method multiple times.
     *
     * @param addr Target address of the portmapper or cluster
     * discovery service. The address string must use the following
     * syntax -
     *    "portmapper@host:port" or
     *    "cluster_discovery@host:port"
     */
    public void addBrokerAddress(String addr) throws IOException {
        addBrokerAddress(addr, false);
    }

    private void addBrokerAddress(String addr, boolean isDiscovered)
        throws IOException {
        if (! validateAddress(addr))
            throw new IOException("Invalid broker address : " + addr);

        if (addrList == null) {
            addrList = new Vector();
            nextBroker = 0;
        }

        if (! isDiscovered) {
            // Discovered addresses can come and go, but the
            // initial address list must be preserved!
            if (failsafeAddrList == null)
                failsafeAddrList = new Hashtable();

            failsafeAddrList.put(addr, addr);
        }

        if (! addrList.contains(addr)) {
            addrList.add(addr);
        }
    }

    /**
     * Find a service.
     *
     * This is the service discovery client workhorse. Clients can use
     * this method to find a service anywhere in the HACluster. This
     * method keeps traversing through the known addresses with the
     * help of cluster discovery service until it finds the requested
     * service. The search is aborted if the service is not found
     * within the specified timeout period.
     *
     * @param serviceType can be either "NORMAL" or "ADMIN".
     * @param serviceProtocol can be "tcp", "http" or "ssl".
     * @param timeout Search timeout in milliseconds.
     * @return String representation of the service address.
     *
     * Service address syntax examples :
     * <pre>
     *     jms@host:port
     *     ssljms@host:port
     *     httpjms@http://www.foo.com/jmqservlet?ServerName=jpgserv
     * </pre>
     */
    public String findService(String serviceType, String serviceProtocol,
        long timeout) throws IOException {
        if (addrList == null || addrList.size() == 0)
            throw new IOException("Broker address list empty");

        long startTime = System.currentTimeMillis();
        boolean done = false;
        boolean timeToSleep = false;

        while (true) {
            nextBroker = nextBroker % addrList.size();
            String addr = (String) addrList.elementAt(nextBroker);

            try {
                discover(addr); // Try the next broker...

                // At this point 'st' cannot be null.
                String ret = st.getServiceAddress(serviceType,
                    serviceProtocol);
                if (ret != null)
                    return ret;

                // Try the current active broker...
                String active = st.getActiveBroker();

                if (active != null) {
                    discover(st.getActiveBroker());
                    ret = st.getServiceAddress(serviceType,
                        serviceProtocol);
                    if (ret != null)
                        return ret;

                    // Active broker turned out to be a dud! The
                    // cluster must be going through state transition.
                    // Sleep for some time and try again.

                    timeToSleep = true;
                }
            }
            catch (IOException e) {}

            nextBroker++;

            if (timeToSleep || nextBroker >= addrList.size()) {
                try {
                    Thread.sleep(retrySleepInterval);
                }
                catch (Exception e) {
                    // nothing to do
                }
                timeToSleep = false;
            }

            if (System.currentTimeMillis() - startTime > timeout)
                return null;
        }
    }

    /**
     * Query the cluster discovery service at the given address.
     */
    private void discover(String addr) throws IOException {
        // Contact the broker.
        String protocol = getProtocol(addr);
        String host = getHost(addr);
        int port = getPort(addr);

        if (protocol.equals("portmapper")) {
            int p = doPortmapper(host, port);

            if (p > 0) {
                port = p;
                protocol = "cluster_discovery";
            }
            else {
                throw new IOException(
    "Could not find cluster discovery service at " + addr);
            }
        }

        if (protocol.equals("cluster_discovery")) {
            st = doClusterDiscovery(host, port);
            return;
        }
    }

    /**
     * Query the iMQ 2.0 portmapper service.
     */
    private int doPortmapper(String host, int port)
        throws IOException {
        String version =
            String.valueOf(PortMapperTable.PORTMAPPER_VERSION) + "\n";
        PortMapperTable pmt = new PortMapperTable();
        Socket socket = new Socket(host, port);
        InputStream  is = socket.getInputStream();
        OutputStream os = socket.getOutputStream();

        try {
            os.write(version.getBytes());
            os.flush();
        } catch (IOException e) {
            // This can sometimes fail if the server already wrote
            // the port table and closed the connection
        }

        pmt.read(is);
        is.close();
        os.close();
        socket.close();

        return pmt.getPortForService(
            ClusterDiscoveryProtocol.SERVICE_NAME);
    }

    /**
     * Query the iMQ Falcon cluster discovery service.
     */
    private ServiceTable doClusterDiscovery(String host, int port)
        throws IOException {
        Socket socket = new Socket(host, port);
        InputStream is = socket.getInputStream();
        OutputStream os = socket.getOutputStream();

        ClusterDiscoveryProtocol.sendRequest(os);
        ServiceTable st = ClusterDiscoveryProtocol.receiveResponse(is);

        // Reconcile the broker address list with the discovered
        // remote service list.
        Hashtable r = st.getRemoteServices();
        Vector removeList = new Vector();

        // First remove the dynamically discovered addresses that are
        // no longer available.
        for (int i = 0; i < addrList.size(); i++) {
            String old = (String) addrList.elementAt(i);
            if (! r.contains(old) && ! failsafeAddrList.contains(old)) {
                removeList.add(old);
            }
        }

        for (int i = 0; i < removeList.size(); i++) {
            addrList.remove(removeList.elementAt(i));
        }

        // Now add all the new addresses.
        Enumeration e = r.keys();
        while (e.hasMoreElements()) {
            String newaddr = (String) e.nextElement();
            addBrokerAddress(newaddr, true);
        }

        // finally return the service table.
        return st;
    }

    /**
     * Validate an address and return a full address string.
     */
    private boolean validateAddress(String addr) {
        if (getProtocol(addr) == null)
            return false;

        if (getHost(addr) == null)
            return false;

        if (getPort(addr) == -1)
            return false;

        return true;
    }

    /**
     * Extract the protocol name from the address.
     */
    public static String getProtocol(String s) {
        int i = s.indexOf('@');
        if (i < 0)
            return null;

        return s.substring(0, i);
    }

    /**
     * Extract the transport address part from the address.
     */
    public static String getAddress(String s) {
        int i = s.indexOf('@');
        if (i < 0)
            return null;

        return s.substring(i+1);
    }

    /**
     * Extract the host name from the address.
     */
    public static String getHost(String s) {
        int i = s.indexOf('@');
        if (i < 0)
            return null;

        int j = s.indexOf(':');
        if (j < 0)
            return null;

        return s.substring(i+1, j);
    }

    /**
     * Extract the port number from the address.
     */
    public static int getPort(String s) {
        int i = s.indexOf(':');
        if (i < 0)
            return -1;

        String t = s.substring(i+1);
        try {
            return Integer.parseInt(t);
        }
        catch (Throwable e) {} // Catches NullPointerException too.

        return -1;
    }

    /**
     * Test application for the service discovery client.
     *
     * <pre>
     * Usage :
     *    java com.sun.messaging.jmq.io.ClusterDiscovery property_file
     * </pre>
     *
     * The property_file looks something like this -
     *
     * <pre>
     * brokers=portmapper@host1:port1, portmapper@host2:port2,...
     * type=NORMAL
     * protocol=tcp
     * timeout=10000
     * repeat_interval=10000
     * repeat_count=10
     * </pre>
     *
     * If search attributes (type, protocol, timeout and
     * repeat_interval) are not specified, then the program will just
     * dump the contents ServiceTable.
     *
     * If the search attributes are specified, the program will search
     * for the specified service and print the address. The search can
     * be performed repeatedly by specifying the repeat_interval and
     * repeat_count attributes. Default repeat_count is 1.
     */
    public static void main(String args[]) throws Exception {
        FileInputStream fis = new FileInputStream(args[0]);
        Properties p = new Properties();
        p.load(fis);

        String brokers = p.getProperty("brokers");

        String type = p.getProperty("type");
        String protocol = p.getProperty("protocol");
        long timeout = 30000;
        long repeat_interval = 10000;
        int repeat_count = 1;

        try {
            timeout = Long.parseLong(p.getProperty("timeout"));
        } catch (Throwable t) {}

        try {
            repeat_interval =
                Long.parseLong(p.getProperty("repeat_interval"));
        } catch (Throwable t) {}

        try {
            repeat_count =
                Integer.parseInt(p.getProperty("repeat_count"));
        } catch (Throwable t) {}

        ClusterDiscovery cd = new ClusterDiscovery();
        Vector v = new Vector();

        StringTokenizer t = new StringTokenizer(brokers, " ,");
        while (t.hasMoreTokens()) {
            String s = (String) t.nextToken();
            cd.addBrokerAddress(s);
            v.add(s);
        }

        for (int i = 0; repeat_count < 0 || i < repeat_count; i++) {
            if (type == null) {
                for (int j = 0; j < v.size(); j++) {
                    try {
                        cd.discover((String) v.elementAt(j));
                    }
                    catch (ConnectException ce) {
                        System.out.println("Broker [" +
                            (String) v.elementAt(j) + "] is down.");
                        continue;
                    }
                    System.out.println(
                        "\n---------------------------------------------");
                    System.out.println((String) v.elementAt(j) + " :");
                    cd.st.dumpServiceTable();
                }
                System.out.println();
            }
            else {
                String addr = cd.findService(type, protocol, timeout);
                System.out.println("Address = " + addr);
            }

            try {
                Thread.sleep(repeat_interval);
            }
            catch (Exception e) {}
        }
    }
}

/*
 * EOF
 */
