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

/*
 * @(#)BrokerMonitor.java	1.31 06/28/07
 */ 

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

import com.sun.messaging.jmq.util.DestType;
import com.sun.messaging.jmq.util.timer.*;
import com.sun.messaging.jmq.util.selector.*;
import com.sun.messaging.jmq.jmsserver.util.*;
import com.sun.messaging.jmq.util.log.*;
import com.sun.messaging.jmq.io.*;
import com.sun.messaging.jmq.jmsserver.Globals;
import com.sun.messaging.jmq.util.admin.MessageType;
import com.sun.messaging.jmq.jmsserver.config.ConfigListener;
import com.sun.messaging.jmq.jmsserver.resources.BrokerResources;
import com.sun.messaging.jmq.jmsserver.config.PropertyUpdateException;
import com.sun.messaging.jmq.jmsserver.config.BrokerConfig;
import java.util.*;
import java.io.*;
import java.net.*;



public class BrokerMonitor
{
    private static final long		DEFAULT_INTERVAL	= 60;
    private static final boolean	DEFAULT_PERSIST		= false;
    private static final long		DEFAULT_TTL		= 5 * DEFAULT_INTERVAL;
    private static final boolean	DEFAULT_ENABLED		= true;

    /*
     * Broker monitoring property names
     */
    private static String METRICS_PROP_PREFIX = 
        Globals.IMQ + ".metrics.topic.";

    private static String METRICS_TIME_PROP = 
	METRICS_PROP_PREFIX + "interval";

    private static String PERSIST_PROP = 
	METRICS_PROP_PREFIX + "persist";

    private static String TTL_PROP = 
	METRICS_PROP_PREFIX + "timetolive";

    private static String ENABLED_PROP = 
	METRICS_PROP_PREFIX + "enabled";

    /*
     * Broker monitoring property values
     */
    public static long METRICS_TIME =
         Globals.getConfig().getLongProperty(
             METRICS_TIME_PROP,
             DEFAULT_INTERVAL) * 1000;

    public static boolean PERSIST =
         Globals.getConfig().getBooleanProperty(
             PERSIST_PROP, DEFAULT_PERSIST);

    public static long TTL =
         Globals.getConfig().getLongProperty(
             TTL_PROP, DEFAULT_TTL) * 1000;

    public static boolean ENABLED =
         Globals.getConfig().getBooleanProperty(
             ENABLED_PROP, DEFAULT_ENABLED);

    private static MQTimer timer = Globals.getTimer();

    Logger logger = Globals.getLogger();

    Monitor monitor = null;

    private static HashSet active = new HashSet();

    private static TimerTask task = null;

    boolean valid = true;
    boolean started = false;

    public static void shutdownMonitor() {
        if (task != null)
            task.cancel();
        active.clear();
	BrokerConfig cfg = Globals.getConfig();
	cfg.removeListener(METRICS_TIME_PROP, cl);
	cfg.removeListener(PERSIST_PROP, cl);
	cfg.removeListener(TTL_PROP, cl);
        cl = null;
    }

    private static ConfigListener cl = new ConfigListener()  {
        public void validate(String name, String value)
    			throws PropertyUpdateException {
        }
            
        public boolean update(String name, String value) {
            BrokerConfig cfg = Globals.getConfig();

            if (name.equals(METRICS_TIME_PROP))  {
	        METRICS_TIME = Globals.getConfig().getLongProperty(
	                METRICS_TIME_PROP, DEFAULT_INTERVAL) * 1000;

                synchronized (active) {
                    if (task != null) {
		       task.cancel();
                       task = new NotificationTask();
                       try {
                           timer.schedule(task, METRICS_TIME, METRICS_TIME);
                       } catch (IllegalStateException ex) {
                           Globals.getLogger().log(Logger.INFO,"Internal Error: Shutting down metrics, timer has been canceled", ex);
                       }
                    }
                }
			
            } else if (name.equals(PERSIST_PROP))  {
                PERSIST = Globals.getConfig().getBooleanProperty(
             		PERSIST_PROP, DEFAULT_PERSIST);
            } else if (name.equals(TTL_PROP))  {
                TTL = Globals.getConfig().getLongProperty(
             		TTL_PROP, DEFAULT_TTL) * 1000;
            } else if (name.equals(ENABLED_PROP))  {
                ENABLED = Globals.getConfig().getBooleanProperty(
             		ENABLED_PROP, DEFAULT_ENABLED);
	    }

            return true;
        }
    };
   
    static class NotificationTask extends TimerTask
    {
        public void run() {
            Iterator itr = null; 
            synchronized(active) {
                itr = (new HashSet(active)).iterator();
            }
            while (itr.hasNext()) {
                Monitor m = (Monitor)itr.next();
                m.run();
            }
        }
    }

    public static void init()  {
	/*
	 * The static listener 'cl' updates the 
	 * static variables METRICS_TIME, PERSIST,
	 * and TTL when their corresponding
	 * properties are updated.
	 */
	BrokerConfig cfg = Globals.getConfig();
	cfg.addListener(METRICS_TIME_PROP, cl);
	cfg.addListener(PERSIST_PROP, cl);
	cfg.addListener(TTL_PROP, cl);
	cfg.addListener(ENABLED_PROP, cl);
    }

    public BrokerMonitor(Destination d)
        throws IllegalArgumentException, BrokerException
    {
        monitor = createMonitor(d);
    }



    public static boolean isInternal(String dest)
    {
        return DestType.destNameIsInternal(dest);
    }


    public void start() {
        synchronized (this) {
            if (!valid)  {
                return;
            }
            if (!started) {
                started = true;
            } else {
                return;
            }
        }
        synchronized (active) {
            active.add(monitor);
            if (task == null) {
               task = new NotificationTask();
               try {
                   timer.schedule(task, METRICS_TIME, METRICS_TIME);
               } catch (IllegalStateException ex) {
                   logger.log(Logger.INFO,"InternalError: Shutting down metrics, timer has been canceled", ex);
               }
            }
        }
    }

    public void stop() {
        synchronized (this) {
            if (!valid) return;
            if (started) {
                started = false;
            } else {
                return;
            }
        }
        synchronized (active) {
            active.remove(monitor);
            if (active.size() == 0) {
               task.cancel();
               task = null;
            }
        }
    }

    public void destroy() {
        stop();
        synchronized(this) {
            valid = false;
            started = false;
            monitor = null;
        }
    }

    private Monitor createMonitor(Destination d)
        throws IllegalArgumentException, BrokerException
    {
        String destination = d.getDestinationName();

        //parse it
        if (!DestType.destNameIsInternal(destination)) {
            throw new IllegalArgumentException("Illegal Internal Name"
                 + destination);
        }
        String substring = destination.substring(
                  DestType.INTERNAL_DEST_PREFIX.length());

        StringTokenizer tk = new StringTokenizer(
                 substring, ".");

        if (!tk.hasMoreElements()) {
            throw new IllegalArgumentException("Missing type "
                + " for monitoring " + destination);
        }

        String type=(String)tk.nextElement();

        if (!type.equals("metrics")) {
            throw new IllegalArgumentException("Illegal type " + type
               + " for monitoring. Only Metrics is valid ["
                  + destination + "]");
        }
        if (!tk.hasMoreElements()) {
            throw new IllegalArgumentException("Missing area "
                + " for monitoring " + destination);
        }
        String area=(String)tk.nextElement();

        // parse
        if (area.equals("broker")) {
            if (tk.hasMoreElements()) {
                throw new IllegalArgumentException("Bad name "
                    + " for broker monitoring " + destination
                    + " should be " + DestType.INTERNAL_DEST_PREFIX
                    + "broker");
            }

            monitor = new BrokerMetricsMonitor(d);
            
        } else if (area.equals("jvm")) {
            if (tk.hasMoreElements()) {
                throw new IllegalArgumentException("Bad name "
                    + " for broker monitoring " + destination
                    + " should be " + DestType.INTERNAL_DEST_PREFIX 
                    + "jvm");
            }
            monitor = new JVMMonitor(d);

        } else if (area.equals("destination")) {
            if (!tk.hasMoreElements()) {
                throw new IllegalArgumentException(
		    "Missing destination type or list for broker destination monitoring "
			+ destination);
            }

            String destArea=(String)tk.nextElement();

	    if (destArea.equals("queue")) {
                if (!tk.hasMoreElements()) {
                    throw new IllegalArgumentException("Missing name "
                    + " for broker queue monitoring " + destination);
                }
                String prefix = "metrics.destination.queue.";
                String name = substring.substring(prefix.length());
                DestinationUID duid =DestinationUID.getUID(name, true);
                if (!Destination.canAutoCreate(true) &&
                     (Destination.getDestination(duid) == null)) {
                    throw new BrokerException(
                       Globals.getBrokerResources().getKString(
                           BrokerResources.E_MONITOR_DEST_DISALLOWED,
                           duid.getName(),
                           duid.getDestType()),Status.FORBIDDEN);
                }
               

                monitor = new DestMonitor(d, duid);

            } else if (destArea.equals("topic")) {
                if (!tk.hasMoreElements()) {
                    throw new IllegalArgumentException("Missing name "
                    + " for broker topic monitoring " + destination);
                }
                String prefix = "metrics.destination.topic.";
                String name = substring.substring(prefix.length());
                DestinationUID duid =DestinationUID.getUID(name, false);
                if (!Destination.canAutoCreate(false) &&
                     (Destination.getDestination(duid) == null)) {
                    throw new BrokerException(
                       Globals.getBrokerResources().getKString(
                           BrokerResources.E_MONITOR_DEST_DISALLOWED,
                           duid.getName(),
                           duid.getDestType()),Status.FORBIDDEN);
                }
                monitor = new DestMonitor(d, duid);
            }
        } else if (area.equals("destination_list")) {
                monitor = new DestListMonitor(d);
        } else {
            throw new IllegalArgumentException("Illegal area "
                + area + " for monitoring " + destination);
        }
        return monitor;

    }


    public void updateNewConsumer(Consumer c) {
        monitor.writeToSpecificMonitorConsumer(c);
    }
     

}



abstract class Monitor 
{
    Destination d = null;

    public Monitor(Destination d)
    {
        this.d = d;
    }

    protected abstract Hashtable getMonitorData();


    public void run() {
        Packet p = new Packet(false);
	Hashtable entries = getMonitorData();

	if (entries == null)  {
	    return;
	}

        if (writeMap(p, entries, d.getDestinationName())) {
           
            try {
                PacketReference ref = PacketReference.createReference(
                     p, null);
                d.queueMessage(ref, false);
                Set s = d.routeNewMessage(ref);
                d.forwardMessage(s, ref);
            } catch (BrokerException ex) {
                Globals.getLogger().log(Logger.DEBUG,"Unable to writeMap for "
               + " metrics" + d, ex);
            } catch (SelectorFormatException ex) {
                Globals.getLogger().log(Logger.DEBUG,"Internal Error ", ex);
            }
        } else {
            Globals.getLogger().log(Logger.DEBUG,"Unable to writeMap for "
               + " metrics" + d);
        }
    }

    public void writeToSpecificMonitorConsumer(Consumer c)
    {
        Packet p = new Packet(true);
	Hashtable entries = getMonitorData();

	if (entries == null)  {
	    return;
	}
        if (c == null) {
            return;
        }

        if (writeMap(p, entries, d.getDestinationName())) {
            try {
                PacketReference ref = PacketReference.createReference(
                     p, null);
                d.queueMessage(ref, false);
                ArrayList arl = new ArrayList(1);
                arl.add(c);
                ref.store(arl);
                c.routeMessage(ref, false);
            } catch (BrokerException ex) {
                Globals.getLogger().log(Logger.DEBUG,"Unable to writeMap for "
               + " metrics " + d + " : targeted for " + c.getConsumerUID(), ex);
            }
        } else {
            Globals.getLogger().log(Logger.DEBUG,"Unable to writeMap for "
               + " metrics" + d+ " : targeted for " + c.getConsumerUID());
        }
    }
 

    private static boolean writeMap(Packet pkt, Hashtable entries,
             String destination)
    {
        try {

            // write header information
            Hashtable props = new Hashtable();
	    Long curTime = new Long(System.currentTimeMillis());

            props.put("type",destination);
            props.put("timestamp", curTime);

	    MQAddress addr = Globals.getMQAddress();
            props.put("brokerAddress", addr.toString());
            props.put("brokerHost", addr.getHostName());
            props.put("brokerPort", new Integer(addr.getPort()));

            pkt.setProperties(props);
            pkt.setPacketType(PacketType.MAP_MESSAGE);
            pkt.setDestination(destination);
            pkt.setPriority(5);
            pkt.setIP(InetAddress.getLocalHost().getAddress());
            pkt.updateSequenceNumber();
            pkt.updateTimestamp();
            pkt.generateSequenceNumber(false);
            pkt.generateTimestamp(false);

            // should also set port, but I'm not sure how
            pkt.setIsQueue(false);
            pkt.setTransactionID(0);
            pkt.setSendAcknowledge(false);
            pkt.setPersistent(BrokerMonitor.PERSIST);
            pkt.setExpiration(BrokerMonitor.TTL == 0 ? (long)0
                   : (curTime.longValue()
                      + BrokerMonitor.TTL));
                  

            ByteArrayOutputStream byteArrayOutputStream = 
                    new ByteArrayOutputStream();
            ObjectOutputStream objectOutputStream = 
                    new ObjectOutputStream(byteArrayOutputStream);

            objectOutputStream.writeObject(entries);
            objectOutputStream.flush();
            byteArrayOutputStream.flush();

            byte[] messageBody = byteArrayOutputStream.toByteArray();

            objectOutputStream.close();
            byteArrayOutputStream.close();
            pkt.setMessageBody(messageBody);

            return true;
        } catch (Exception e) {
            Globals.getLogger().log(Logger.ERROR,
               "Error sending metrics data",e);
            return false;
        }
    }

}

class JVMMonitor extends Monitor
{
    public JVMMonitor(Destination d) {
        super(d);
    }

    protected Hashtable getMonitorData() {

        Hashtable mapMessage = new Hashtable();

        MetricManager mm = Globals.getMetricManager();
        MetricData md = mm.getMetrics();
        mapMessage.put("freeMemory", 
               new Long(md.freeMemory));
        mapMessage.put("maxMemory", 
               new Long(Runtime.getRuntime().maxMemory()));
        mapMessage.put("totalMemory", 
               new Long(md.totalMemory));

        return mapMessage;
    }
}

class DestListMonitor extends Monitor
{
    public DestListMonitor(Destination d) {
         super(d);
    }

    protected Hashtable getMonitorData() {

        Hashtable mapMessages = new Hashtable();

	Iterator itr = Destination.getAllDestinations();
	while (itr.hasNext()) {
	    Destination oneDest = (Destination)itr.next();
	    Hashtable values;
	    String key;

	    if (oneDest.isInternal() || 
		oneDest.isAdmin() ||
	        (oneDest.getDestinationName().equals(MessageType.JMQ_ADMIN_DEST)) ||
            (oneDest.getDestinationName().equals(MessageType.JMQ_BRIDGE_ADMIN_DEST)))  {
		continue;
	    }

	    values = new Hashtable();

	    if (oneDest.isQueue())  {
	        key = "mq.metrics.destination.queue." + oneDest.getDestinationName();
	        values.put("type", "queue");
	    } else  {
	        key = "mq.metrics.destination.topic." + oneDest.getDestinationName();
	        values.put("type", "topic");
	    }

	    values.put("name", oneDest.getDestinationName());
	    values.put("isTemporary", Boolean.valueOf(oneDest.isTemporary()));
	    
	    mapMessages.put(key, values);
	}

        return mapMessages;
    }
}

class DestMonitor extends Monitor
{
    DestinationUID target = null;

    public DestMonitor(Destination d,
         DestinationUID target) {
         super(d);
         this.target = target;
    }

    protected Hashtable getMonitorData() {

	if (target == null)  {
	    return (null);
	}

        Destination td = Destination.getDestination(target);

	if (td == null)  {
	    return (null);
	}

        Hashtable values = new Hashtable(td.getMetrics());
        return values;
    }
        

}

class BrokerMetricsMonitor extends Monitor
{

    public BrokerMetricsMonitor(Destination d) {
        super(d);
    }

    protected Hashtable getMonitorData() {

        Hashtable mapMessage = new Hashtable();

        MetricManager mm = Globals.getMetricManager();
        MetricData md = mm.getMetrics();
        mapMessage.put("numConnections", 
               new Long((long)md.nConnections));
        mapMessage.put("numMsgsIn", 
               new Long((long)md.totals.messagesIn));
        mapMessage.put("numMsgsOut", 
               new Long((long)md.totals.messagesOut));
        mapMessage.put("numMsgs", 
               new Long((long)Destination.totalCount()));

        mapMessage.put("msgBytesIn", 
               new Long((long)md.totals.messageBytesIn));
        mapMessage.put("msgBytesOut", 
               new Long((long)md.totals.messageBytesOut));
        mapMessage.put("numPktsIn", 
               new Long((long)md.totals.packetsIn));
        mapMessage.put("numPktsOut", 
               new Long((long)md.totals.packetsOut));
        mapMessage.put("pktBytesIn", 
               new Long((long)md.totals.packetBytesIn));
        mapMessage.put("pktBytesOut", 
               new Long((long)md.totals.packetBytesOut));

        mapMessage.put("totalMsgBytes", 
               new Long(Destination.totalBytes()));

	/*
	 * Calculate number of destinations.
	 * We cannot use Destination.destinationsSize() here because
	 * that includes all destinations. We need to filter out
	 * internal/admin destinations - basically what is returned
	 * by DestListMonitor.
	 */
	Iterator itr = Destination.getAllDestinations();
	long numDests = 0;
	while (itr.hasNext()) {
	    Destination oneDest = (Destination)itr.next();

	    if (oneDest.isInternal() || 
		oneDest.isAdmin() ||
	        (oneDest.getDestinationName().equals(MessageType.JMQ_BRIDGE_ADMIN_DEST)) ||
	        (oneDest.getDestinationName().equals(MessageType.JMQ_ADMIN_DEST)))  {
		continue;
	    }
	    numDests++;
	}

        mapMessage.put("numDestinations", new Long(numDests));

        return mapMessage;
    }
}
