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

/*
 * @(#)ClusterDiscoveryProtocol.java	1.4 06/27/07
 */ 

package com.sun.messaging.jmq.io;

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

/**
 * Encapsulates the cluster discovery wire protocol implementation.
 * This class is used by both the client and the broker.
 */
public class ClusterDiscoveryProtocol {
    /** Cluster discovery protocol version. */
    public static final int CLUSTER_DISCOVERY_VERSION = 210;

    public static final String SERVICE_NAME = "cluster_discovery";
    public static final String END_TOKEN = "END\r\n";

    // Cluster discovery protocol attributes (property names).
    public static final String VERSION         = "version";
    public static final String OPERATION_TYPE  = "operation_type";
    public static final String BROKER_NAME     = "broker_name";
    public static final String BROKER_VERSION  = "broker_version";
    public static final String ACTIVE_BROKER   = "active_broker";
    public static final String REMOTE_SERVICES = "remote_services";
    public static final String LOCAL_SERVICES  = "local_services";

    /**
     * Send a cluster discovery request packet.
     *
     * This method is used by the cluster discovery client.
     */
    public static void sendRequest(OutputStream os)
        throws IOException {
        Properties p = new Properties();

        p.setProperty(VERSION,
            Integer.toString(CLUSTER_DISCOVERY_VERSION));
        p.setProperty(OPERATION_TYPE, "REQUEST");

        p.save(os, null);
        os.write(END_TOKEN.getBytes("ASCII"));
        os.flush();
    }

    /**
     * Receive a request packet.
     *
     * This method is used by the cluster discovery service.
     */
    public static Properties receiveRequest(InputStream is)
        throws IOException {
        return receivePacket(is);
    }

    /**
     * Send cluster discovery response packet.
     *
     * This method is used by the cluster discovery service.
     */
    public static void sendResponse(ServiceTable st, OutputStream os)
        throws IOException {
        // First compile the set of properties from the given
        // ServiceTable object.
        Properties p = new Properties();

        p.setProperty(VERSION,
            Integer.toString(CLUSTER_DISCOVERY_VERSION));
        p.setProperty(OPERATION_TYPE, "RESPONSE");
        p.setProperty(BROKER_NAME, st.getBrokerInstanceName());
        p.setProperty(BROKER_VERSION, st.getBrokerVersion());

        String active = st.getActiveBroker();
        if (active != null) {
            p.setProperty(ACTIVE_BROKER, active);
        }

        Hashtable r = st.getRemoteServices();
        if (r.size() > 0) {
            Enumeration e = r.keys();
            String rval = (String) e.nextElement();

            while (e.hasMoreElements())
                rval = rval + "," + e.nextElement();

            p.setProperty(REMOTE_SERVICES, rval);
        }

        Enumeration local = st.getServices().elements();
        String lval = null;

        while (local.hasMoreElements()) {
            ServiceEntry se = (ServiceEntry) local.nextElement();
            String prefix = "local." + se.getName();

            p.setProperty(prefix + ".type", se.getType());
            p.setProperty(prefix + ".protocol", se.getProtocol());
            p.setProperty(prefix + ".address", se.getAddress());

            if (lval == null)
                lval = se.getName();
            else
                lval = lval + "," + se.getName();
        }

        if (lval != null)
            p.setProperty(LOCAL_SERVICES, lval);

        // Now serialize the properties on the given output stream.
        p.save(os, null);
        os.write(END_TOKEN.getBytes("ASCII"));
        os.flush();
    }

    /**
     * Receive a response packet.
     *
     * This method is used by the cluster discovery client.
     *
     * @return A ServiceTable object containing the cluster discovery
     * information.
     * @exception IOException If an I/O error occurs or if there is a
     * version mismatch.
     */
    public static ServiceTable receiveResponse(InputStream is)
        throws IOException {
        Properties p = receivePacket(is);

        ServiceTable st = new ServiceTable();
        st.setBrokerInstanceName(p.getProperty(BROKER_NAME));
        st.setBrokerVersion(p.getProperty(BROKER_VERSION));

        String s = p.getProperty(ACTIVE_BROKER);
        if (s != null)
            st.setActiveBroker(s);

        s = p.getProperty(REMOTE_SERVICES);
        if (s != null) {
            StringTokenizer t = new StringTokenizer(s, " ,");
            while (t.hasMoreTokens()) {
                String r = t.nextToken();
                st.addRemoteService(r);
            }
        }

        s = p.getProperty(LOCAL_SERVICES);
        if (s != null) {
            StringTokenizer t = new StringTokenizer(s, " ,");
            while (t.hasMoreTokens()) {
                String l = t.nextToken();
                String prefix = "local." + l;
                String type = p.getProperty(prefix + ".type");
                String protocol = p.getProperty(prefix + ".protocol");
                String address = p.getProperty(prefix + ".address");

                ServiceEntry se = new ServiceEntry();
                se.setName(l);
                se.setType(type);
                se.setProtocol(protocol);
                se.setAddress(address);

                st.add(se);
            }
        }

        return st;
    }

    /**
     * Reads all the bytes from the wire and generates a Properties
     * object.
     */
    private static Properties receivePacket(InputStream is)
        throws IOException {
        byte[] endbuf = new byte[5];
        BufferedInputStream bis = new BufferedInputStream(is);
        ByteArrayOutputStream baos = new ByteArrayOutputStream();

        while (true) {
            int c = bis.read();

            if (c == -1) {
                throw new IOException(
                    "Premature end of input stream");
            }

            baos.write(c);

            if (c == '\n') {
                int n = bis.read(endbuf);

                if (n < endbuf.length) {
                    throw new IOException(
                        "Premature end of input stream");
                }

                if (endbuf[0] == 'E' &&
                    endbuf[1] == 'N' &&
                    endbuf[2] == 'D' &&
                    endbuf[3] == '\r' &&
                    endbuf[4] == '\n') {
                    break;
                }

                baos.write(endbuf);
            }
        }

        byte[] buffer = baos.toByteArray();

        Properties p = new Properties();
        ByteArrayInputStream bais = new ByteArrayInputStream(buffer);
        p.load(bais);

        int version = 0;
        try {
            version = Integer.parseInt(p.getProperty(VERSION));
        }
        catch (Throwable t) {}
        if (version == 0 || version > CLUSTER_DISCOVERY_VERSION) {
            throw new IOException(
                "Bad cluster discovery version number: " +
                version + " Expecting: " + CLUSTER_DISCOVERY_VERSION);

        }

        return p;
    }
}

/*
 * EOF
 */
