sjero.net Git - linphone/blob - p2pproxy/dependencies-src/jxse-src-2.5/impl/src/net/jxta/impl/meter/MonitorManager.java

   1 /*
   2  * Copyright (c) 2001-2007 Sun Microsystems, Inc.  All rights reserved.
   3  *
   4  *  The Sun Project JXTA(TM) Software License
   5  *
   6  *  Redistribution and use in source and binary forms, with or without
   7  *  modification, are permitted provided that the following conditions are met:
   8  *
   9  *  1. Redistributions of source code must retain the above copyright notice,
  10  *     this list of conditions and the following disclaimer.
  11  *
  12  *  2. Redistributions in binary form must reproduce the above copyright notice,
  13  *     this list of conditions and the following disclaimer in the documentation
  14  *     and/or other materials provided with the distribution.
  15  *
  16  *  3. The end-user documentation included with the redistribution, if any, must
  17  *     include the following acknowledgment: "This product includes software
  18  *     developed by Sun Microsystems, Inc. for JXTA(TM) technology."
  19  *     Alternately, this acknowledgment may appear in the software itself, if
  20  *     and wherever such third-party acknowledgments normally appear.
  21  *
  22  *  4. The names "Sun", "Sun Microsystems, Inc.", "JXTA" and "Project JXTA" must
  23  *     not be used to endorse or promote products derived from this software
  24  *     without prior written permission. For written permission, please contact
  25  *     Project JXTA at http://www.jxta.org.
  26  *
  27  *  5. Products derived from this software may not be called "JXTA", nor may
  28  *     "JXTA" appear in their name, without prior written permission of Sun.
  29  *
  30  *  THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES,
  31  *  INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
  32  *  FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL SUN
  33  *  MICROSYSTEMS OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
  34  *  INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  35  *  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA,
  36  *  OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
  37  *  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
  38  *  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
  39  *  EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  40  *
  41  *  JXTA is a registered trademark of Sun Microsystems, Inc. in the United
  42  *  States and other countries.
  43  *
  44  *  Please see the license information page at :
  45  *  <http://www.jxta.org/project/www/license.html> for instructions on use of
  46  *  the license in source files.
  47  *
  48  *  ====================================================================
  49  *
  50  *  This software consists of voluntary contributions made by many individuals
  51  *  on behalf of Project JXTA. For more information on Project JXTA, please see
  52  *  http://www.jxta.org.
  53  *
  54  *  This license is based on the BSD license adopted by the Apache Foundation.
  55  */
  56
  57 package net.jxta.impl.meter;
  58
  59
  60 import net.jxta.document.Advertisement;
  61 import net.jxta.exception.JxtaException;
  62 import net.jxta.id.ID;
  63 import net.jxta.impl.util.TimeUtils;
  64 import net.jxta.meter.MonitorEvent;
  65 import net.jxta.meter.MonitorException;
  66 import net.jxta.meter.MonitorFilter;
  67 import net.jxta.meter.MonitorFilterException;
  68 import net.jxta.meter.MonitorListener;
  69 import net.jxta.meter.MonitorReport;
  70 import net.jxta.meter.MonitorResources;
  71 import net.jxta.meter.PeerMonitorInfo;
  72 import net.jxta.meter.ServiceMetric;
  73 import net.jxta.meter.ServiceMonitor;
  74 import net.jxta.meter.ServiceMonitorFilter;
  75 import net.jxta.peergroup.PeerGroup;
  76 import net.jxta.peergroup.PeerGroupID;
  77 import net.jxta.platform.ModuleClassID;
  78 import net.jxta.protocol.ModuleImplAdvertisement;
  79 import net.jxta.service.Service;
  80 import net.jxta.util.documentSerializable.DocumentSerializableUtilities;
  81 import net.jxta.util.documentSerializable.DocumentSerializationException;
  82
  83 import java.util.Enumeration;
  84 import java.util.Hashtable;
  85 import java.util.Iterator;
  86 import java.util.LinkedList;
  87
  88
  89 public class MonitorManager implements Service {
  90     private final static long timeZero = System.currentTimeMillis();
  91     public static final int NOT_PULSING = -1;
  92
  93     private static final int NO_PRIOR_REPORT = 0;
  94
  95     private static long supportedReportRates[] = new long[] {
  96         500, TimeUtils.ASECOND, 5 * TimeUtils.ASECOND, 10 * TimeUtils.ASECOND, 15 * TimeUtils.ASECOND, 30 * TimeUtils.ASECOND
  97                 ,
  98         TimeUtils.AMINUTE, 5 * TimeUtils.AMINUTE, 10 * TimeUtils.AMINUTE, 15 * TimeUtils.AMINUTE, 30 * TimeUtils.AMINUTE
  99                 ,
 100         TimeUtils.ANHOUR, 3 * TimeUtils.ANHOUR, 6 * TimeUtils.ANHOUR, 12 * TimeUtils.ANHOUR, TimeUtils.ADAY, TimeUtils.AWEEK};
 101
 102     private int pulsesPerRate[] = new int[supportedReportRates.length];
 103     private long startTime = System.currentTimeMillis();
 104
 105     private LinkedList<MonitorListenerInfo> monitorListenerInfos = new LinkedList<MonitorListenerInfo>();
 106     private Hashtable<ModuleClassID, ServiceMonitorPulseInfo> serviceMonitorPulseInfos = new Hashtable<ModuleClassID, ServiceMonitorPulseInfo>();
 107     private int filtersPerRate[] = new int[supportedReportRates.length];
 108     private long previousReportTimes[] = new long[supportedReportRates.length];
 109
 110     private PeerGroup peerGroup;
 111     private Thread reportThread;
 112
 113     private long pulseRate = NOT_PULSING;
 114     private int pulseRateIndex = NOT_PULSING;
 115     private int pulseNumber = 0;
 116     private long nextPulseTime = NO_PRIOR_REPORT;
 117     private boolean isRunning = true; // true until monitor is destroyed, triggers termination of report thread
 118
 119     private ModuleClassID[] supportedModuleClassIDs;
 120     private ModuleImplAdvertisement implAdvertisement;
 121     private long lastResetTime = System.currentTimeMillis();
 122
 123     public Advertisement getImplAdvertisement() {
 124         return implAdvertisement;
 125     }
 126
 127     public Service getInterface() {
 128         // This is good enough. No need to get fancy.
 129         return this;
 130     }
 131
 132     // public MonitorManager(PeerGroup peerGroup) {
 133
 134     public void init(PeerGroup peerGroup, ID assignedID, Advertisement implAdvertisement) {
 135         this.implAdvertisement = (ModuleImplAdvertisement) implAdvertisement;
 136         this.peerGroup = peerGroup;
 137         createReportThread();
 138
 139         for (int i = 0; i < previousReportTimes.length; i++) {
 140             pulsesPerRate[i] = (int) (supportedReportRates[i] / supportedReportRates[0]);
 141         }
 142     }
 143
 144     public int startApp(java.lang.String[] args) {
 145         return 0; // fix-me
 146     }
 147
 148     public void stopApp() {
 149         destroy();
 150     }
 151
 152     private class MonitorListenerInfo {
 153         MonitorListener monitorListener;
 154         MonitorFilter monitorFilter;
 155         long reportRate;
 156         int reportRateIndex;
 157         boolean sendCumulativeFirst = false;
 158         boolean wasCumulativeSent = false;
 159
 160         MonitorListenerInfo(MonitorListener monitorListener, long reportRate, MonitorFilter monitorFilter, boolean cumulativeFirst) {
 161             this.monitorListener = monitorListener;
 162             this.monitorFilter = monitorFilter;
 163             this.reportRate = reportRate;
 164             this.sendCumulativeFirst = cumulativeFirst;
 165             this.reportRateIndex = getReportRateIndex(reportRate);
 166         }
 167     }
 168
 169     public static long[] getReportRates() { // return copy so that users can't modify.
 170         long copy[] = new long[supportedReportRates.length];
 171
 172         System.arraycopy(supportedReportRates, 0, copy, 0, supportedReportRates.length);
 173         return copy;
 174     }
 175
 176     public boolean isLocalMonitoringAvailable(ModuleClassID moduleClassID) {
 177         ServiceMonitor serviceMonitor = getServiceMonitor(moduleClassID);
 178
 179         return (serviceMonitor != null);
 180     }
 181
 182     public PeerGroup getPeerGroup() {
 183         return peerGroup;
 184     }
 185
 186     // Cooperate with the code that loaded this module to replace the strong
 187     // group interface given by init() with a non-counted one.
 188     private void setPeerGroup(PeerGroup pg) {
 189         PeerGroup tmp = peerGroup;
 190
 191         peerGroup = pg;
 192         tmp.unref();
 193         tmp = null;
 194     }
 195
 196     public PeerMonitorInfo getPeerMonitorInfo() {
 197         long[] reportRates = getReportRates(); // makes a copy
 198         ModuleClassID[] moduleClassIDs = getMonitorableServiceTypes(); // ensures that array is initialized.
 199         long runningTime = System.currentTimeMillis() - lastResetTime;
 200
 201         return new PeerMonitorInfo(MeterBuildSettings.METERING, moduleClassIDs, reportRates, lastResetTime, runningTime);
 202     }
 203
 204     public int getReportRatesCount() {
 205         return supportedReportRates.length;
 206     }
 207
 208     public int getReportRateIndex(long reportRate) {
 209         for (int i = 0; i < supportedReportRates.length; i++) {
 210             if (supportedReportRates[i] == reportRate) {
 211                 return i;
 212             }
 213         }
 214
 215         return -1;
 216     }
 217
 218     public boolean isSupportedReportRate(long reportRate) {
 219         return getReportRateIndex(reportRate) >= 0;
 220     }
 221
 222     public long getReportRate(int index) {
 223         return supportedReportRates[index];
 224     }
 225
 226     public long getBestReportRate(long desiredReportRate) {
 227         for (long supportedReportRate : supportedReportRates) {
 228             if (desiredReportRate <= supportedReportRate) {
 229                 return supportedReportRate;
 230             }
 231         }
 232
 233         return supportedReportRates[supportedReportRates.length - 1];
 234     }
 235
 236     public ServiceMonitor getServiceMonitor(ModuleClassID moduleClassID) {
 237         ServiceMonitorPulseInfo serviceMonitorPulseInfo = serviceMonitorPulseInfos.get(moduleClassID);
 238
 239         if (serviceMonitorPulseInfo != null) {
 240             return serviceMonitorPulseInfo.serviceMonitor;
 241         } else {
 242
 243             try {
 244                 ModuleImplAdvertisement moduleImplAdvertisement = MonitorResources.getServiceMonitorImplAdvertisement(
 245                         moduleClassID, implAdvertisement);
 246                 ServiceMonitor serviceMonitor = (ServiceMonitor) peerGroup.loadModule(moduleClassID, moduleImplAdvertisement);
 247
 248                 MonitorResources.registerServiceMonitorModuleImplAdvertisement(moduleImplAdvertisement);
 249
 250                 if (serviceMonitor instanceof ServiceMonitorImpl) {
 251                     ((ServiceMonitorImpl) serviceMonitor).init(this);
 252
 253                 }
 254
 255                 serviceMonitorPulseInfo = new ServiceMonitorPulseInfo(this, serviceMonitor);
 256                 serviceMonitorPulseInfos.put(moduleClassID, serviceMonitorPulseInfo);
 257                 return serviceMonitor;
 258             } catch (JxtaException e) {
 259                 throw new RuntimeException("Unable to load Service Monitor: " + moduleClassID + "\n\tException: " + e);
 260             }
 261         }
 262     }
 263
 264     private void resetPulseRate() {
 265         int oldPulseRateIndex = pulseRateIndex;
 266
 267         pulseRateIndex = NOT_PULSING;
 268         pulseRate = NOT_PULSING;
 269
 270         for (int i = 0; i < filtersPerRate.length; i++) {
 271             if (filtersPerRate[i] != 0) {
 272                 pulseRateIndex = i;
 273                 pulseRate = getReportRate(pulseRateIndex);
 274                 break;
 275             }
 276         }
 277
 278         if (oldPulseRateIndex == pulseRateIndex) {
 279             return;
 280         }    // nothing changed
 281
 282         long now = System.currentTimeMillis();
 283
 284         if (oldPulseRateIndex == NOT_PULSING) { // case 1: No pulse to pulse
 285             for (int i = 0; i < filtersPerRate.length; i++) {
 286                 if (filtersPerRate[i] != 0) {
 287                     previousReportTimes[i] = now;
 288                 } else {
 289                     previousReportTimes[i] = NO_PRIOR_REPORT;
 290                 }
 291             }
 292
 293             pulseNumber = 0;
 294             nextPulseTime = now + pulseRate;
 295         } else if (pulseRateIndex == NOT_PULSING) {// case 2: pulse to No pulse
 296             // Do nothing
 297         } else if (pulseRateIndex < oldPulseRateIndex) { // case 3: pulse going to a faster pulse
 298             for (int i = pulseRateIndex; i < (oldPulseRateIndex - 1); i++) {
 299                 if (filtersPerRate[i] != 0) {
 300                     previousReportTimes[i] = now;
 301                 } else {
 302                     previousReportTimes[i] = NO_PRIOR_REPORT;
 303                 }
 304             }
 305
 306             long timeToNextPulse = nextPulseTime - now;
 307
 308             if (pulseRate < timeToNextPulse) {
 309                 int numPulsesToNow = (int) (timeToNextPulse / pulseRate);
 310                 int numNewToOldPulses = (int) (supportedReportRates[oldPulseRateIndex] / supportedReportRates[pulseRateIndex]);
 311
 312                 pulseNumber += (numNewToOldPulses - numPulsesToNow) * pulsesPerRate[pulseRateIndex];
 313                 timeToNextPulse = now - (numPulsesToNow * pulseRate);
 314             } else {
 315                 pulseNumber += (pulsesPerRate[oldPulseRateIndex] - pulsesPerRate[pulseRateIndex]);
 316             }
 317
 318         } else if (pulseRateIndex > oldPulseRateIndex) { // case 3: pulse going to a  slower pulse
 319             int nextPulseNumber = pulseNumber + pulsesPerRate[oldPulseRateIndex];
 320
 321             pulseNumber = ((nextPulseNumber - 1) / pulsesPerRate[pulseRateIndex]) * pulsesPerRate[pulseRateIndex];
 322             nextPulseTime += (nextPulseNumber - pulseNumber) * supportedReportRates[0];
 323
 324             for (int i = 0; i < pulseRateIndex; i++) {
 325                 previousReportTimes[i] = NO_PRIOR_REPORT;
 326             }
 327         }
 328
 329         reportThread.interrupt();
 330     }
 331
 332     private MonitorReport getMonitorReport(MonitorFilter monitorFilter, long reportRate, long previousDeltaTime, long beginReportTime) {
 333         MonitorReport monitorReport = new MonitorReport(previousDeltaTime, beginReportTime, false);
 334
 335         for (Iterator i = monitorFilter.getModuleClassIDs(); i.hasNext();) {
 336             ModuleClassID moduleClassID = (ModuleClassID) i.next();
 337
 338             ServiceMonitorFilter serviceMonitorFilter = monitorFilter.getServiceMonitorFilter(moduleClassID);
 339
 340             ServiceMonitor serviceMonitor = getServiceMonitor(moduleClassID);
 341
 342             if (serviceMonitorFilter != null) {
 343                 ServiceMetric serviceMetric = serviceMonitor.getServiceMetric(serviceMonitorFilter, previousDeltaTime
 344                         ,
 345                         beginReportTime, getReportRateIndex(reportRate), reportRate);
 346
 347                 if (serviceMetric != null) {
 348                     monitorReport.addServiceMetric(serviceMetric);
 349                 }
 350             }
 351         }
 352         return monitorReport;
 353     }
 354
 355     public void validateCumulativeMonitorFilter(MonitorFilter monitorFilter) throws MonitorFilterException {
 356         boolean isAnyServiceFilters = false;
 357
 358         for (Iterator i = monitorFilter.getServiceMonitorFilters(); i.hasNext();) {
 359             ServiceMonitorFilter serviceMonitorFilter = (ServiceMonitorFilter) i.next();
 360
 361             ModuleClassID moduleClassID = serviceMonitorFilter.getModuleClassID();
 362             ServiceMonitor serviceMonitor = getServiceMonitor(moduleClassID);
 363
 364             if (serviceMonitor == null) {
 365                 throw new MonitorFilterException(MonitorFilterException.SERVICE_NOT_SUPPORTED, moduleClassID);
 366             }
 367
 368             serviceMonitor.validateCumulativeServiceMonitorFilter(serviceMonitorFilter);
 369             isAnyServiceFilters = true;
 370         }
 371
 372         if (!isAnyServiceFilters) {
 373             throw new MonitorFilterException("Empty Monitor Filter");
 374         }
 375
 376     }
 377
 378     public void validateMonitorFilter(MonitorFilter monitorFilter, long reportRate) throws MonitorFilterException {
 379
 380         if (!isSupportedReportRate(reportRate)) {
 381             throw new MonitorFilterException(MonitorFilterException.REPORT_RATE_NOT_SUPPORTED, reportRate);
 382         }
 383
 384         boolean isAnyServiceFilters = false;
 385
 386         for (Iterator i = monitorFilter.getServiceMonitorFilters(); i.hasNext();) {
 387             ServiceMonitorFilter serviceMonitorFilter = (ServiceMonitorFilter) i.next();
 388
 389             ModuleClassID moduleClassID = serviceMonitorFilter.getModuleClassID();
 390             ServiceMonitor serviceMonitor = getServiceMonitor(moduleClassID);
 391
 392             if (serviceMonitor == null) {
 393                 throw new MonitorFilterException(MonitorFilterException.SERVICE_NOT_SUPPORTED, moduleClassID);
 394             }
 395
 396             serviceMonitor.validateServiceMonitorFilter(serviceMonitorFilter, reportRate);
 397             isAnyServiceFilters = true;
 398         }
 399
 400         if (!isAnyServiceFilters) {
 401             throw new MonitorFilterException("Empty Monitor Filter");
 402         }
 403     }
 404
 405     public MonitorFilter createSupportedCumulativeMonitorFilter(MonitorFilter monitorFilter) throws MonitorFilterException {
 406         MonitorFilter newMonitorFilter = new MonitorFilter(monitorFilter.getDescription());
 407         boolean anythingAdded = false;
 408
 409         for (Iterator i = monitorFilter.getServiceMonitorFilters(); i.hasNext();) {
 410             ServiceMonitorFilter serviceMonitorFilter = (ServiceMonitorFilter) i.next();
 411
 412             ModuleClassID moduleClassID = serviceMonitorFilter.getModuleClassID();
 413             ServiceMonitor serviceMonitor = getServiceMonitor(moduleClassID);
 414
 415             if (serviceMonitor == null) {
 416                 continue;
 417             }
 418
 419             ServiceMonitorFilter newServiceMonitorFilter = serviceMonitor.createSupportedCumulativeServiceMonitorFilter(
 420                     serviceMonitorFilter);
 421
 422             if (newServiceMonitorFilter != null) {
 423                 newMonitorFilter.addServiceMonitorFilter(newServiceMonitorFilter);
 424                 anythingAdded = true;
 425             }
 426         }
 427
 428         if (anythingAdded) {
 429             return newMonitorFilter;
 430         } else {
 431             return null;
 432         }
 433     }
 434
 435     public MonitorFilter createSupportedMonitorFilter(MonitorFilter monitorFilter, long reportRate) throws MonitorFilterException {
 436         MonitorFilter newMonitorFilter = new MonitorFilter(monitorFilter.getDescription());
 437         boolean anythingAdded = false;
 438
 439         for (Iterator i = monitorFilter.getServiceMonitorFilters(); i.hasNext();) {
 440             ServiceMonitorFilter serviceMonitorFilter = (ServiceMonitorFilter) i.next();
 441
 442             ModuleClassID moduleClassID = serviceMonitorFilter.getModuleClassID();
 443             ServiceMonitor serviceMonitor = getServiceMonitor(moduleClassID);
 444
 445             if (serviceMonitor == null) {
 446                 continue;
 447             }
 448
 449             ServiceMonitorFilter newServiceMonitorFilter = serviceMonitor.createSupportedServiceMonitorFilter(serviceMonitorFilter
 450                     ,
 451                     reportRate);
 452
 453             if (newServiceMonitorFilter != null) {
 454                 newMonitorFilter.addServiceMonitorFilter(newServiceMonitorFilter);
 455                 anythingAdded = true;
 456             }
 457         }
 458
 459         if (anythingAdded) {
 460             return newMonitorFilter;
 461         } else {
 462             return null;
 463         }
 464     }
 465
 466     public synchronized long addMonitorListener(MonitorFilter monitorFilter, long reportRate, boolean includeCumulative, MonitorListener monitorListener) throws MonitorException {
 467
 468         validateMonitorFilter(monitorFilter, reportRate); // if validation fails, it will throw an exception
 469
 470         if (includeCumulative) {
 471             validateCumulativeMonitorFilter(monitorFilter);
 472         }    // if validation fails, it will throw an exception
 473
 474         int reportRateIndex = getReportRateIndex(reportRate);
 475
 476         try {
 477             monitorFilter = (MonitorFilter) DocumentSerializableUtilities.copyDocumentSerializable(monitorFilter); // make a copy of the filter
 478         } catch (DocumentSerializationException e) {
 479             throw new MonitorException(MonitorException.SERIALIZATION, "Error trying to copy MonitorFilter");
 480         }
 481
 482         MonitorListenerInfo monitorListenerInfo = new MonitorListenerInfo(monitorListener, reportRate, monitorFilter
 483                 ,
 484                 includeCumulative);
 485
 486         monitorListenerInfos.add(monitorListenerInfo);
 487         filtersPerRate[reportRateIndex]++;
 488
 489         if ((filtersPerRate[reportRateIndex] == 1) && (pulseRateIndex != NOT_PULSING) && (reportRateIndex > pulseRateIndex)) {
 490             previousReportTimes[reportRateIndex] = previousReportTimes[pulseRateIndex];
 491         }
 492
 493         for (Iterator i = monitorFilter.getModuleClassIDs(); i.hasNext();) {
 494             ModuleClassID moduleClassID = (ModuleClassID) i.next();
 495
 496             ServiceMonitorFilter serviceMonitorFilter = monitorFilter.getServiceMonitorFilter(moduleClassID);
 497             ServiceMonitorPulseInfo serviceMonitorPulseInfo = serviceMonitorPulseInfos.get(moduleClassID);
 498
 499             serviceMonitorPulseInfo.registerServiceMonitorFilter(serviceMonitorFilter, reportRateIndex, reportRate);
 500         }
 501
 502         resetPulseRate();
 503
 504         return reportRate;
 505     }
 506
 507     private MonitorListenerInfo getMonitorListenerInfo(MonitorListener monitorListener) {
 508         for (Object monitorListenerInfo1 : monitorListenerInfos) {
 509             MonitorListenerInfo monitorListenerInfo = (MonitorListenerInfo) monitorListenerInfo1;
 510
 511             if (monitorListenerInfo.monitorListener == monitorListener) {
 512                 return monitorListenerInfo;
 513             }
 514         }
 515         return null;
 516     }
 517
 518     public synchronized int removeMonitorListener(MonitorListener monitorListener) {
 519         int numRemoved = 0;
 520
 521         for (;;) { // remove all instances of this listener
 522             MonitorListenerInfo monitorListenerInfo = getMonitorListenerInfo(monitorListener);
 523
 524             if (monitorListenerInfo == null) {
 525                 break;
 526             } else {
 527                 MonitorFilter monitorFilter = monitorListenerInfo.monitorFilter;
 528                 long reportRate = monitorListenerInfo.reportRate;
 529                 int reportRateIndex = monitorListenerInfo.reportRateIndex;
 530
 531                 monitorListenerInfos.remove(monitorListenerInfo);
 532                 numRemoved++;
 533                 filtersPerRate[reportRateIndex]--;
 534
 535                 for (Iterator i = monitorFilter.getModuleClassIDs(); i.hasNext();) {
 536                     ModuleClassID moduleClassID = (ModuleClassID) i.next();
 537
 538                     ServiceMonitorFilter serviceMonitorFilter = monitorFilter.getServiceMonitorFilter(moduleClassID);
 539                     ServiceMonitorPulseInfo serviceMonitorPulseInfo = serviceMonitorPulseInfos.get(moduleClassID);
 540
 541                     serviceMonitorPulseInfo.deregisterServiceMonitorFilter(serviceMonitorFilter, reportRateIndex, reportRate);
 542                 }
 543             }
 544         }
 545
 546         resetPulseRate();
 547
 548         return numRemoved;
 549     }
 550
 551     public synchronized MonitorReport getCumulativeMonitorReport(MonitorFilter monitorFilter) throws MonitorException {
 552         validateCumulativeMonitorFilter(monitorFilter);
 553
 554         long beginReportTime = System.currentTimeMillis();
 555
 556         MonitorReport monitorReport = new MonitorReport(startTime, beginReportTime, true);
 557
 558         for (Iterator i = monitorFilter.getModuleClassIDs(); i.hasNext();) {
 559             ModuleClassID moduleClassID = (ModuleClassID) i.next();
 560
 561             ServiceMonitorFilter serviceMonitorFilter = monitorFilter.getServiceMonitorFilter(moduleClassID);
 562             ServiceMonitor serviceMonitor = getServiceMonitor(moduleClassID);
 563
 564             ServiceMetric serviceMetric = serviceMonitor.getCumulativeServiceMetric(serviceMonitorFilter, timeZero
 565                     ,
 566                     beginReportTime);
 567
 568             monitorReport.addServiceMetric(moduleClassID, serviceMetric);
 569         }
 570
 571         return monitorReport;
 572     }
 573
 574     public ModuleClassID[] getMonitorableServiceTypes() {
 575         if (supportedModuleClassIDs == null) {
 576             ModuleClassID[] registeredModuleClassIDs = MonitorResources.getRegisteredModuleClassIDs();
 577             LinkedList<ModuleClassID> supportedModuleClassIDsList = new LinkedList<ModuleClassID>();
 578
 579             for (ModuleClassID registeredModuleClassID : registeredModuleClassIDs) {
 580                 if (isLocalMonitoringAvailable(registeredModuleClassID)) {
 581                     supportedModuleClassIDsList.add(registeredModuleClassID);
 582                 }
 583             }
 584
 585             supportedModuleClassIDs = supportedModuleClassIDsList.toArray(new ModuleClassID[0]);
 586         }
 587         return supportedModuleClassIDs;
 588     }
 589
 590     // fastest pulse rate registered anywhere
 591     public long getPulseRate() {
 592         return getReportRate(pulseRateIndex);
 593     }
 594
 595     // index of fastest pulse anywhere
 596     public int getPulseRateIndex() {
 597         return pulseRateIndex;
 598     }
 599
 600     // pulse rate for this monitor
 601     public long getPulseRate(ServiceMonitor serviceMonitor) {
 602         ServiceMonitorPulseInfo serviceMonitorPulseInfo = serviceMonitorPulseInfos.get(serviceMonitor.getModuleClassID());
 603
 604         if (serviceMonitorPulseInfo != null) {
 605             return serviceMonitorPulseInfo.getPulseRate();
 606         } else {
 607             return ServiceMonitorPulseInfo.NOT_PULSING;
 608         }
 609     }
 610
 611     // index of pulse rate for this monitor
 612     public long getPulseRateIndex(ServiceMonitor serviceMonitor) {
 613         ServiceMonitorPulseInfo serviceMonitorPulseInfo = serviceMonitorPulseInfos.get(serviceMonitor.getModuleClassID());
 614
 615         if (serviceMonitorPulseInfo != null) {
 616             return serviceMonitorPulseInfo.getPulseRateIndex();
 617         } else {
 618             return ServiceMonitorPulseInfo.NOT_PULSING;
 619         }
 620     }
 621
 622     private void generateReports() {
 623         long beginReportTime = System.currentTimeMillis();
 624
 625         for (Enumeration<ServiceMonitorPulseInfo> e = serviceMonitorPulseInfos.elements(); e.hasMoreElements();) {
 626             ServiceMonitorPulseInfo serviceMonitorPulseInfo = e.nextElement();
 627             int servicePulseRateIndex = serviceMonitorPulseInfo.getPulseRateIndex();
 628
 629             if ((serviceMonitorPulseInfo.serviceMonitor instanceof ServiceMonitorImpl)
 630                     && isEvenPulseForRateIndex(servicePulseRateIndex)) {
 631                 ((ServiceMonitorImpl) serviceMonitorPulseInfo.serviceMonitor).beginPulse(serviceMonitorPulseInfo);
 632             }
 633         }
 634
 635         for (Object monitorListenerInfo1 : monitorListenerInfos) {
 636             MonitorListenerInfo monitorListenerInfo = (MonitorListenerInfo) monitorListenerInfo1;
 637             MonitorFilter monitorFilter = monitorListenerInfo.monitorFilter;
 638             MonitorListener monitorListener = monitorListenerInfo.monitorListener;
 639
 640             int reportRateIndex = monitorListenerInfo.reportRateIndex;
 641             long reportRate = monitorListenerInfo.reportRate;
 642
 643             if (isEvenPulseForRateIndex(reportRateIndex)) {
 644                 MonitorReport monitorReport = null;
 645
 646                 try {
 647                     if (monitorListenerInfo.sendCumulativeFirst && !monitorListenerInfo.wasCumulativeSent) {
 648                         monitorReport = getCumulativeMonitorReport(monitorFilter);
 649
 650                         MonitorEvent monitorEvent = new MonitorEvent(peerGroup.getPeerGroupID(), monitorReport);
 651
 652                         monitorListener.processMonitorReport(monitorEvent);
 653                         monitorListenerInfo.wasCumulativeSent = true;
 654                     } else {
 655                         monitorReport = getMonitorReport(monitorFilter, reportRate, previousReportTimes[reportRateIndex]
 656                                 ,
 657                                 beginReportTime);
 658                         MonitorEvent monitorEvent = new MonitorEvent(peerGroup.getPeerGroupID(), monitorReport);
 659
 660                         monitorListener.processMonitorReport(monitorEvent);
 661                     }
 662                 } catch (Throwable e) {
 663                     e.printStackTrace();
 664                     // Fix-Me: Where should we report an uncaught exception in one of our listeners?
 665                 }
 666             }
 667         }
 668
 669         for (int rateIndex = 0; rateIndex < supportedReportRates.length; rateIndex++) {
 670             if (isEvenPulseForRateIndex(rateIndex)) {
 671                 if (filtersPerRate[rateIndex] != 0) {
 672                     previousReportTimes[rateIndex] = beginReportTime;
 673                 } else {
 674                     previousReportTimes[rateIndex] = NO_PRIOR_REPORT;
 675                 }
 676             }
 677         }
 678
 679         for (Enumeration<ServiceMonitorPulseInfo> e = serviceMonitorPulseInfos.elements(); e.hasMoreElements();) {
 680             ServiceMonitorPulseInfo serviceMonitorPulseInfo = e.nextElement();
 681             int servicePulseRateIndex = serviceMonitorPulseInfo.getPulseRateIndex();
 682
 683             if ((serviceMonitorPulseInfo.serviceMonitor instanceof ServiceMonitorImpl)
 684                     && isEvenPulseForRateIndex(servicePulseRateIndex)) {
 685                 ((ServiceMonitorImpl) serviceMonitorPulseInfo.serviceMonitor).endPulse(serviceMonitorPulseInfo);
 686             }
 687         }
 688     }
 689
 690     boolean isEvenPulseForRateIndex(int pulseRateIndex) {
 691         if (pulseRateIndex < 0 || pulseRateIndex > pulsesPerRate.length) {
 692             return false;
 693         }
 694         return ((pulseNumber % pulsesPerRate[pulseRateIndex]) == 0);
 695     }
 696
 697     private void createReportThread() {
 698         reportThread = new Thread(new Runnable() {
 699             public void run() {
 700                 mainLoop:
 701                 while (isRunning) {
 702                     synchronized (MonitorManager.this) { // no new listeners while reporting
 703                         while (pulseRate == NOT_PULSING) {
 704                             try {
 705                                 MonitorManager.this.wait();
 706                             } catch (InterruptedException e) {
 707                                 continue mainLoop;
 708                             }
 709                         }
 710
 711                         while (pulseRate != NOT_PULSING) {
 712                             if (Thread.interrupted()) {
 713                                 continue mainLoop;
 714                             }
 715
 716                             long now = System.currentTimeMillis();
 717
 718                             try {
 719                                 long waitTime = nextPulseTime - now;
 720
 721                                 if (waitTime > 0) {
 722                                     MonitorManager.this.wait(nextPulseTime - now);
 723                                 }
 724
 725                                 pulseNumber += pulsesPerRate[pulseRateIndex];
 726                                 generateReports();
 727                                 nextPulseTime += pulseRate;
 728                             } catch (InterruptedException e) {
 729                                 if (pulseRateIndex == NOT_PULSING) {
 730                                     continue mainLoop;
 731                                 }
 732                             } catch (Exception ex) {
 733                                 // don't die forever on exceptions!!
 734                                 ex.printStackTrace(); // fix-me: report this
 735                             }
 736                         }
 737                     }
 738                 }
 739             }
 740         }, "Meter-Monitor-Report");
 741
 742         reportThread.setDaemon(true);
 743         reportThread.start();
 744     }
 745
 746     public synchronized void destroy() {
 747         isRunning = false;
 748         reportThread.interrupt();
 749
 750         for (Enumeration<ServiceMonitorPulseInfo> e = serviceMonitorPulseInfos.elements(); e.hasMoreElements();) {
 751             ServiceMonitorPulseInfo serviceMonitorPulseInfo = e.nextElement();
 752             ServiceMonitor serviceMonitor = serviceMonitorPulseInfo.serviceMonitor;
 753
 754             serviceMonitor.destroy();
 755         }
 756     }
 757
 758     /**
 759      * DO NOT USE THIS FIELD: It will be deprecated when MonitorManager becomes a
 760      * FULL FLEDGED SERVICE
 761      */
 762
 763     private static Hashtable<PeerGroupID, MonitorManager> monitorManagers = new Hashtable<PeerGroupID, MonitorManager>();
 764
 765     /**
 766      * DO NOT USE THIS METHOD: It will be deprecated when MonitorManager becomes a
 767      * FULL FLEDGED SERVICE
 768      */
 769
 770     public static MonitorManager registerMonitorManager(PeerGroup peerGroup) throws JxtaException {
 771         PeerGroupID peerGroupID = peerGroup.getPeerGroupID();
 772         MonitorManager monitorManager = monitorManagers.get(peerGroupID);
 773
 774         if (monitorManager == null) {
 775             boolean includeTransports = true;
 776             ModuleImplAdvertisement moduleImplAdvertisement = MonitorResources.getReferenceAllPurposeMonitorServiceImplAdvertisement(
 777                     includeTransports);
 778
 779             monitorManager = (MonitorManager) peerGroup.loadModule(MonitorResources.refMonitorServiceSpecID
 780                     ,
 781                     moduleImplAdvertisement);
 782             monitorManagers.put(peerGroupID, monitorManager);
 783
 784             // FIXME jice@jxta.org - 20021103 : this
 785             // monitorManager is not a real group service:
 786             // it is being loadModule()'d by another as a
 787             // result, it holds a counted reference to the
 788             // group.  Idealy, we'd need the groupAPI to
 789             // offer a means to loadModule() without
 790             // making a counted reference, so that group
 791             // services can loadModule() things without
 792             // preventing group termination. This could be
 793             // achieved elegantly by making this only
 794             // behaviour available through a weak
 795             // GroupInterface. So it would be enough to
 796             // obtain a weak interface from one's group
 797             // and then use its loadmodule method rather
 798             // than that of the strong group interface.
 799             // However that's a bit too big a change to be
 800             // decided without more carefull
 801             // consideration.  Instead, we just simulate
 802             // it for now: we give to the monitor manager
 803             // the real group reference after loadModule
 804             // is done, and it discards the strong
 805             // interface object that was passed to its
 806             // init routine.
 807
 808             monitorManager.setPeerGroup(peerGroup);
 809         }
 810         return monitorManager;
 811     }
 812
 813     /**
 814      * DO NOT USE THIS METHOD: It will be deprecated when MonitorManager becomes a
 815      * FULL FLEDGED SERVICE
 816      */
 817
 818     public static void unregisterMonitorManager(PeerGroup peerGroup) {
 819         PeerGroupID peerGroupID = peerGroup.getPeerGroupID();
 820
 821         monitorManagers.remove(peerGroupID);
 822     }
 823
 824     public static ServiceMonitor getServiceMonitor(PeerGroup peerGroup, ModuleClassID serviceClassID) {
 825         try {
 826             PeerGroupID peerGroupID = peerGroup.getPeerGroupID();
 827             MonitorManager monitorManager = monitorManagers.get(peerGroupID);
 828
 829             return monitorManager.getServiceMonitor(serviceClassID);
 830         } catch (Exception e) { // Fix-Me: This is a bit sloppy
 831             throw new RuntimeException("Unable to find MonitorManager or MonitorService");
 832         }
 833     }
 834 }