monitormanager.java

JXTA&#8482 is a set of open, generalized peer-to-peer (P2P) protocols that allow any networked devi

/*
 * Copyright (c) 2001-2007 Sun Microsystems, Inc.  All rights reserved.
 *  
 *  The Sun Project JXTA(TM) Software License
 *  
 *  Redistribution and use in source and binary forms, with or without 
 *  modification, are permitted provided that the following conditions are met:
 *  
 *  1. Redistributions of source code must retain the above copyright notice,
 *     this list of conditions and the following disclaimer.
 *  
 *  2. Redistributions in binary form must reproduce the above copyright notice, 
 *     this list of conditions and the following disclaimer in the documentation 
 *     and/or other materials provided with the distribution.
 *  
 *  3. The end-user documentation included with the redistribution, if any, must 
 *     include the following acknowledgment: "This product includes software 
 *     developed by Sun Microsystems, Inc. for JXTA(TM) technology." 
 *     Alternately, this acknowledgment may appear in the software itself, if 
 *     and wherever such third-party acknowledgments normally appear.
 *  
 *  4. The names "Sun", "Sun Microsystems, Inc.", "JXTA" and "Project JXTA" must 
 *     not be used to endorse or promote products derived from this software 
 *     without prior written permission. For written permission, please contact 
 *     Project JXTA at http://www.jxta.org.
 *  
 *  5. Products derived from this software may not be called "JXTA", nor may 
 *     "JXTA" appear in their name, without prior written permission of Sun.
 *  
 *  THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES,
 *  INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND 
 *  FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL SUN 
 *  MICROSYSTEMS OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, 
 *  INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT 
 *  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, 
 *  OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
 *  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
 *  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, 
 *  EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *  
 *  JXTA is a registered trademark of Sun Microsystems, Inc. in the United 
 *  States and other countries.
 *  
 *  Please see the license information page at :
 *  <http://www.jxta.org/project/www/license.html> for instructions on use of 
 *  the license in source files.
 *  
 *  ====================================================================
 *  
 *  This software consists of voluntary contributions made by many individuals 
 *  on behalf of Project JXTA. For more information on Project JXTA, please see 
 *  http://www.jxta.org.
 *  
 *  This license is based on the BSD license adopted by the Apache Foundation. 
 */

package net.jxta.impl.meter;


import net.jxta.document.Advertisement;
import net.jxta.exception.JxtaException;
import net.jxta.id.ID;
import net.jxta.impl.util.TimeUtils;
import net.jxta.meter.MonitorEvent;
import net.jxta.meter.MonitorException;
import net.jxta.meter.MonitorFilter;
import net.jxta.meter.MonitorFilterException;
import net.jxta.meter.MonitorListener;
import net.jxta.meter.MonitorReport;
import net.jxta.meter.MonitorResources;
import net.jxta.meter.PeerMonitorInfo;
import net.jxta.meter.ServiceMetric;
import net.jxta.meter.ServiceMonitor;
import net.jxta.meter.ServiceMonitorFilter;
import net.jxta.peergroup.PeerGroup;
import net.jxta.peergroup.PeerGroupID;
import net.jxta.platform.ModuleClassID;
import net.jxta.protocol.ModuleImplAdvertisement;
import net.jxta.service.Service;
import net.jxta.util.documentSerializable.DocumentSerializableUtilities;
import net.jxta.util.documentSerializable.DocumentSerializationException;

import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.LinkedList;


public class MonitorManager implements Service {
    private final static long timeZero = System.currentTimeMillis();
    public static final int NOT_PULSING = -1;

    private static final int NO_PRIOR_REPORT = 0;

    private static long supportedReportRates[] = new long[] {
        500, TimeUtils.ASECOND, 5 * TimeUtils.ASECOND, 10 * TimeUtils.ASECOND, 15 * TimeUtils.ASECOND, 30 * TimeUtils.ASECOND
                ,
        TimeUtils.AMINUTE, 5 * TimeUtils.AMINUTE, 10 * TimeUtils.AMINUTE, 15 * TimeUtils.AMINUTE, 30 * TimeUtils.AMINUTE
                ,
        TimeUtils.ANHOUR, 3 * TimeUtils.ANHOUR, 6 * TimeUtils.ANHOUR, 12 * TimeUtils.ANHOUR, TimeUtils.ADAY, TimeUtils.AWEEK};

    private int pulsesPerRate[] = new int[supportedReportRates.length];
    private long startTime = System.currentTimeMillis();

    private LinkedList<MonitorListenerInfo> monitorListenerInfos = new LinkedList<MonitorListenerInfo>();
    private Hashtable<ModuleClassID, ServiceMonitorPulseInfo> serviceMonitorPulseInfos = new Hashtable<ModuleClassID, ServiceMonitorPulseInfo>();
    private int filtersPerRate[] = new int[supportedReportRates.length];
    private long previousReportTimes[] = new long[supportedReportRates.length];

    private PeerGroup peerGroup;
    private Thread reportThread;

    private long pulseRate = NOT_PULSING;
    private int pulseRateIndex = NOT_PULSING;
    private int pulseNumber = 0;
    private long nextPulseTime = NO_PRIOR_REPORT;
    private boolean isRunning = true; // true until monitor is destroyed, triggers termination of report thread

    private ModuleClassID[] supportedModuleClassIDs;
    private ModuleImplAdvertisement implAdvertisement;
    private long lastResetTime = System.currentTimeMillis();

    public Advertisement getImplAdvertisement() {
        return implAdvertisement;
    }

    public Service getInterface() {
        // This is good enough. No need to get fancy.
        return this;
    }

    // public MonitorManager(PeerGroup peerGroup) {

    public void init(PeerGroup peerGroup, ID assignedID, Advertisement implAdvertisement) {
        this.implAdvertisement = (ModuleImplAdvertisement) implAdvertisement;
        this.peerGroup = peerGroup;
        createReportThread();

        for (int i = 0; i < previousReportTimes.length; i++) {
            pulsesPerRate[i] = (int) (supportedReportRates[i] / supportedReportRates[0]);
        }
    }

    public int startApp(java.lang.String[] args) {
        return 0; // fix-me
    }

    public void stopApp() {
        destroy();
    }

    private class MonitorListenerInfo {
        MonitorListener monitorListener;
        MonitorFilter monitorFilter;
        long reportRate;
        int reportRateIndex;
        boolean sendCumulativeFirst = false;
        boolean wasCumulativeSent = false;

        MonitorListenerInfo(MonitorListener monitorListener, long reportRate, MonitorFilter monitorFilter, boolean cumulativeFirst) {
            this.monitorListener = monitorListener;
            this.monitorFilter = monitorFilter;
            this.reportRate = reportRate;
            this.sendCumulativeFirst = cumulativeFirst;
            this.reportRateIndex = getReportRateIndex(reportRate);
        }
    }

    public static long[] getReportRates() { // return copy so that users can't modify.
        long copy[] = new long[supportedReportRates.length];

        System.arraycopy(supportedReportRates, 0, copy, 0, supportedReportRates.length);
        return copy;
    }

    public boolean isLocalMonitoringAvailable(ModuleClassID moduleClassID) {
        ServiceMonitor serviceMonitor = getServiceMonitor(moduleClassID);

        return (serviceMonitor != null);
    }

    public PeerGroup getPeerGroup() {
        return peerGroup;
    }

    // Cooperate with the code that loaded this module to replace the strong
    // group interface given by init() with a non-counted one.
    private void setPeerGroup(PeerGroup pg) {
        PeerGroup tmp = peerGroup;

        peerGroup = pg;
        tmp.unref();
        tmp = null;
    }

    public PeerMonitorInfo getPeerMonitorInfo() {
        long[] reportRates = getReportRates(); // makes a copy
        ModuleClassID[] moduleClassIDs = getMonitorableServiceTypes(); // ensures that array is initialized.
        long runningTime = System.currentTimeMillis() - lastResetTime;

        return new PeerMonitorInfo(MeterBuildSettings.METERING, moduleClassIDs, reportRates, lastResetTime, runningTime);
    }

    public int getReportRatesCount() {
        return supportedReportRates.length;
    }

    public int getReportRateIndex(long reportRate) {
        for (int i = 0; i < supportedReportRates.length; i++) {
            if (supportedReportRates[i] == reportRate) {
                return i;
            }
        }

        return -1;
    }

    public boolean isSupportedReportRate(long reportRate) {
        return getReportRateIndex(reportRate) >= 0;
    }

    public long getReportRate(int index) {
        return supportedReportRates[index];
    }

    public long getBestReportRate(long desiredReportRate) {
        for (long supportedReportRate : supportedReportRates) {
            if (desiredReportRate <= supportedReportRate) {
                return supportedReportRate;
            }
        }

        return supportedReportRates[supportedReportRates.length - 1];
    }

    public ServiceMonitor getServiceMonitor(ModuleClassID moduleClassID) {
        ServiceMonitorPulseInfo serviceMonitorPulseInfo = serviceMonitorPulseInfos.get(moduleClassID);

        if (serviceMonitorPulseInfo != null) {
            return serviceMonitorPulseInfo.serviceMonitor;
        } else {

            try {
                ModuleImplAdvertisement moduleImplAdvertisement = MonitorResources.getServiceMonitorImplAdvertisement(
                        moduleClassID, implAdvertisement);
                ServiceMonitor serviceMonitor = (ServiceMonitor) peerGroup.loadModule(moduleClassID, moduleImplAdvertisement);

                MonitorResources.registerServiceMonitorModuleImplAdvertisement(moduleImplAdvertisement);

                if (serviceMonitor instanceof ServiceMonitorImpl) {
                    ((ServiceMonitorImpl) serviceMonitor).init(this);

                }

                serviceMonitorPulseInfo = new ServiceMonitorPulseInfo(this, serviceMonitor);
                serviceMonitorPulseInfos.put(moduleClassID, serviceMonitorPulseInfo);
                return serviceMonitor;
            } catch (JxtaException e) {
                throw new RuntimeException("Unable to load Service Monitor: " + moduleClassID + "\n\tException: " + e);
            }
        }
    }

    private void resetPulseRate() {
        int oldPulseRateIndex = pulseRateIndex;

        pulseRateIndex = NOT_PULSING;
        pulseRate = NOT_PULSING;

        for (int i = 0; i < filtersPerRate.length; i++) {
            if (filtersPerRate[i] != 0) {
                pulseRateIndex = i;
                pulseRate = getReportRate(pulseRateIndex);
                break;
            }
        }

        if (oldPulseRateIndex == pulseRateIndex) {
            return;