monitormanager.java

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

            ModuleClassID moduleClassID = (ModuleClassID) i.next();

            ServiceMonitorFilter serviceMonitorFilter = monitorFilter.getServiceMonitorFilter(moduleClassID);
            ServiceMonitor serviceMonitor = getServiceMonitor(moduleClassID);

            ServiceMetric serviceMetric = serviceMonitor.getCumulativeServiceMetric(serviceMonitorFilter, timeZero
                    ,
                    beginReportTime);

            monitorReport.addServiceMetric(moduleClassID, serviceMetric);
        }

        return monitorReport;
    }

    public ModuleClassID[] getMonitorableServiceTypes() {
        if (supportedModuleClassIDs == null) {
            ModuleClassID[] registeredModuleClassIDs = MonitorResources.getRegisteredModuleClassIDs();
            LinkedList<ModuleClassID> supportedModuleClassIDsList = new LinkedList<ModuleClassID>();

            for (ModuleClassID registeredModuleClassID : registeredModuleClassIDs) {
                if (isLocalMonitoringAvailable(registeredModuleClassID)) {
                    supportedModuleClassIDsList.add(registeredModuleClassID);
                }
            }

            supportedModuleClassIDs = supportedModuleClassIDsList.toArray(new ModuleClassID[0]);
        }
        return supportedModuleClassIDs;
    }

    // fastest pulse rate registered anywhere
    public long getPulseRate() {
        return getReportRate(pulseRateIndex);
    }

    // index of fastest pulse anywhere
    public int getPulseRateIndex() {
        return pulseRateIndex;
    }

    // pulse rate for this monitor
    public long getPulseRate(ServiceMonitor serviceMonitor) {
        ServiceMonitorPulseInfo serviceMonitorPulseInfo = serviceMonitorPulseInfos.get(serviceMonitor.getModuleClassID());

        if (serviceMonitorPulseInfo != null) {
            return serviceMonitorPulseInfo.getPulseRate();
        } else {
            return ServiceMonitorPulseInfo.NOT_PULSING;
        }
    }

    // index of pulse rate for this monitor
    public long getPulseRateIndex(ServiceMonitor serviceMonitor) {
        ServiceMonitorPulseInfo serviceMonitorPulseInfo = serviceMonitorPulseInfos.get(serviceMonitor.getModuleClassID());

        if (serviceMonitorPulseInfo != null) {
            return serviceMonitorPulseInfo.getPulseRateIndex();
        } else {
            return ServiceMonitorPulseInfo.NOT_PULSING;
        }
    }

    private void generateReports() {
        long beginReportTime = System.currentTimeMillis();

        for (Enumeration<ServiceMonitorPulseInfo> e = serviceMonitorPulseInfos.elements(); e.hasMoreElements();) {
            ServiceMonitorPulseInfo serviceMonitorPulseInfo = e.nextElement();
            int servicePulseRateIndex = serviceMonitorPulseInfo.getPulseRateIndex();

            if ((serviceMonitorPulseInfo.serviceMonitor instanceof ServiceMonitorImpl)
                    && isEvenPulseForRateIndex(servicePulseRateIndex)) {
                ((ServiceMonitorImpl) serviceMonitorPulseInfo.serviceMonitor).beginPulse(serviceMonitorPulseInfo);
            }
        }

        for (Object monitorListenerInfo1 : monitorListenerInfos) {
            MonitorListenerInfo monitorListenerInfo = (MonitorListenerInfo) monitorListenerInfo1;
            MonitorFilter monitorFilter = monitorListenerInfo.monitorFilter;
            MonitorListener monitorListener = monitorListenerInfo.monitorListener;

            int reportRateIndex = monitorListenerInfo.reportRateIndex;
            long reportRate = monitorListenerInfo.reportRate;

            if (isEvenPulseForRateIndex(reportRateIndex)) {
                MonitorReport monitorReport = null;

                try {
                    if (monitorListenerInfo.sendCumulativeFirst && !monitorListenerInfo.wasCumulativeSent) {
                        monitorReport = getCumulativeMonitorReport(monitorFilter);

                        MonitorEvent monitorEvent = new MonitorEvent(peerGroup.getPeerGroupID(), monitorReport);

                        monitorListener.processMonitorReport(monitorEvent);
                        monitorListenerInfo.wasCumulativeSent = true;
                    } else {
                        monitorReport = getMonitorReport(monitorFilter, reportRate, previousReportTimes[reportRateIndex]
                                ,
                                beginReportTime);
                        MonitorEvent monitorEvent = new MonitorEvent(peerGroup.getPeerGroupID(), monitorReport);

                        monitorListener.processMonitorReport(monitorEvent);
                    }
                } catch (Throwable e) {
                    e.printStackTrace();
                    // Fix-Me: Where should we report an uncaught exception in one of our listeners?
                }
            }
        }

        for (int rateIndex = 0; rateIndex < supportedReportRates.length; rateIndex++) {
            if (isEvenPulseForRateIndex(rateIndex)) {
                if (filtersPerRate[rateIndex] != 0) {
                    previousReportTimes[rateIndex] = beginReportTime;
                } else {
                    previousReportTimes[rateIndex] = NO_PRIOR_REPORT;
                }
            }
        }

        for (Enumeration<ServiceMonitorPulseInfo> e = serviceMonitorPulseInfos.elements(); e.hasMoreElements();) {
            ServiceMonitorPulseInfo serviceMonitorPulseInfo = e.nextElement();
            int servicePulseRateIndex = serviceMonitorPulseInfo.getPulseRateIndex();

            if ((serviceMonitorPulseInfo.serviceMonitor instanceof ServiceMonitorImpl)
                    && isEvenPulseForRateIndex(servicePulseRateIndex)) {
                ((ServiceMonitorImpl) serviceMonitorPulseInfo.serviceMonitor).endPulse(serviceMonitorPulseInfo);
            }
        }
    }

    boolean isEvenPulseForRateIndex(int pulseRateIndex) {
        if (pulseRateIndex < 0 || pulseRateIndex > pulsesPerRate.length) {
            return false;
        }
        return ((pulseNumber % pulsesPerRate[pulseRateIndex]) == 0);
    }

    private void createReportThread() {
        reportThread = new Thread(new Runnable() {
            public void run() {
                mainLoop:
                while (isRunning) {
                    synchronized (MonitorManager.this) { // no new listeners while reporting
                        while (pulseRate == NOT_PULSING) {
                            try {
                                MonitorManager.this.wait();
                            } catch (InterruptedException e) {
                                continue mainLoop;
                            }
                        }

                        while (pulseRate != NOT_PULSING) {
                            if (Thread.interrupted()) {
                                continue mainLoop;
                            }

                            long now = System.currentTimeMillis();

                            try {
                                long waitTime = nextPulseTime - now;

                                if (waitTime > 0) {
                                    MonitorManager.this.wait(nextPulseTime - now);
                                }

                                pulseNumber += pulsesPerRate[pulseRateIndex];
                                generateReports();
                                nextPulseTime += pulseRate;
                            } catch (InterruptedException e) {
                                if (pulseRateIndex == NOT_PULSING) {
                                    continue mainLoop;
                                }
                            } catch (Exception ex) {
                                // don't die forever on exceptions!!
                                ex.printStackTrace(); // fix-me: report this
                            }
                        }
                    }
                }
            }
        }, "Meter-Monitor-Report");

        reportThread.setDaemon(true);
        reportThread.start();
    }

    public synchronized void destroy() {
        isRunning = false;
        reportThread.interrupt();

        for (Enumeration<ServiceMonitorPulseInfo> e = serviceMonitorPulseInfos.elements(); e.hasMoreElements();) {
            ServiceMonitorPulseInfo serviceMonitorPulseInfo = e.nextElement();
            ServiceMonitor serviceMonitor = serviceMonitorPulseInfo.serviceMonitor;

            serviceMonitor.destroy();
        }
    }

    /**
     * DO NOT USE THIS FIELD: It will be deprecated when MonitorManager becomes a
     * FULL FLEDGED SERVICE
     */

    private static Hashtable<PeerGroupID, MonitorManager> monitorManagers = new Hashtable<PeerGroupID, MonitorManager>();

    /**
     * DO NOT USE THIS METHOD: It will be deprecated when MonitorManager becomes a
     * FULL FLEDGED SERVICE
     */

    public static MonitorManager registerMonitorManager(PeerGroup peerGroup) throws JxtaException {
        PeerGroupID peerGroupID = peerGroup.getPeerGroupID();
        MonitorManager monitorManager = monitorManagers.get(peerGroupID);

        if (monitorManager == null) {
            boolean includeTransports = true;
            ModuleImplAdvertisement moduleImplAdvertisement = MonitorResources.getReferenceAllPurposeMonitorServiceImplAdvertisement(
                    includeTransports);

            monitorManager = (MonitorManager) peerGroup.loadModule(MonitorResources.refMonitorServiceSpecID
                    ,
                    moduleImplAdvertisement);
            monitorManagers.put(peerGroupID, monitorManager);

            // FIXME [e-mail] - 20021103 : this
            // monitorManager is not a real group service:
            // it is being loadModule()'d by another as a
            // result, it holds a counted reference to the
            // group.  Idealy, we'd need the groupAPI to
            // offer a means to loadModule() without
            // making a counted reference, so that group
            // services can loadModule() things without
            // preventing group termination. This could be
            // achieved elegantly by making this only
            // behaviour available through a weak
            // GroupInterface. So it would be enough to
            // obtain a weak interface from one's group
            // and then use its loadmodule method rather
            // than that of the strong group interface.
            // However that's a bit too big a change to be
            // decided without more carefull
            // consideration.  Instead, we just simulate
            // it for now: we give to the monitor manager
            // the real group reference after loadModule
            // is done, and it discards the strong
            // interface object that was passed to its
            // init routine.

            monitorManager.setPeerGroup(peerGroup);
        }
        return monitorManager;
    }

    /**
     * DO NOT USE THIS METHOD: It will be deprecated when MonitorManager becomes a
     * FULL FLEDGED SERVICE
     */

    public static void unregisterMonitorManager(PeerGroup peerGroup) {
        PeerGroupID peerGroupID = peerGroup.getPeerGroupID();

        monitorManagers.remove(peerGroupID);
    }

    public static ServiceMonitor getServiceMonitor(PeerGroup peerGroup, ModuleClassID serviceClassID) {
        try {
            PeerGroupID peerGroupID = peerGroup.getPeerGroupID();
            MonitorManager monitorManager = monitorManagers.get(peerGroupID);

            return monitorManager.getServiceMonitor(serviceClassID);
        } catch (Exception e) { // Fix-Me: This is a bit sloppy
            throw new RuntimeException("Unable to find MonitorManager or MonitorService");
        }
    }
}