monitormanager.java

jxta_src_2.41b jxta 2.41b 最新版源码 from www.jxta.org

			if (newServiceMonitorFilter != null) {
				newMonitorFilter.addServiceMonitorFilter(newServiceMonitorFilter);
				anythingAdded = true;
			}
		}

		if (anythingAdded)
			return newMonitorFilter;	
		else
			return null;
	}

	public MonitorFilter createSupportedMonitorFilter(MonitorFilter monitorFilter, long reportRate) throws MonitorFilterException{
		MonitorFilter newMonitorFilter = new MonitorFilter(monitorFilter.getDescription());
		boolean anythingAdded = false;

		for (Iterator i = monitorFilter.getServiceMonitorFilters(); i.hasNext(); ) {
			ServiceMonitorFilter serviceMonitorFilter = (ServiceMonitorFilter)i.next();
			
			ModuleClassID moduleClassID = serviceMonitorFilter.getModuleClassID();
			ServiceMonitor serviceMonitor = getServiceMonitor(moduleClassID);			

			if (serviceMonitor == null)
				continue;
			
			ServiceMonitorFilter newServiceMonitorFilter = serviceMonitor.createSupportedServiceMonitorFilter(serviceMonitorFilter, reportRate);

			if (newServiceMonitorFilter != null) {
				newMonitorFilter.addServiceMonitorFilter(newServiceMonitorFilter);
				anythingAdded = true;
			}
		}

		if (anythingAdded)
			return newMonitorFilter;	
		else
			return null;
	}

	public synchronized long addMonitorListener(MonitorFilter monitorFilter, long reportRate, boolean includeCumulative, MonitorListener monitorListener) throws MonitorException {

		validateMonitorFilter(monitorFilter, reportRate);	// if validation fails, it will throw an exception

		if (includeCumulative)
			validateCumulativeMonitorFilter(monitorFilter);	// if validation fails, it will throw an exception

		int reportRateIndex = getReportRateIndex(reportRate);

		try {
			monitorFilter = (MonitorFilter) DocumentSerializableUtilities.copyDocumentSerializable(monitorFilter);	// make a copy of the filter
		} catch (DocumentSerializationException e) {
			throw new MonitorException(MonitorException.SERIALIZATION, "Error trying to copy MonitorFilter");			
		}
		
		MonitorListenerInfo monitorListenerInfo = new MonitorListenerInfo(monitorListener, reportRate, monitorFilter, includeCumulative);
		monitorListenerInfos.add(monitorListenerInfo); 
		filtersPerRate[reportRateIndex]++;

		if ((filtersPerRate[reportRateIndex] == 1) && (pulseRateIndex != NOT_PULSING) && (reportRateIndex > pulseRateIndex))
			previousReportTimes[reportRateIndex] = previousReportTimes[pulseRateIndex];

		for (Iterator i = monitorFilter.getModuleClassIDs(); i.hasNext(); ) {
			ModuleClassID moduleClassID = (ModuleClassID)i.next();
			
			ServiceMonitorFilter serviceMonitorFilter = monitorFilter.getServiceMonitorFilter(moduleClassID);	
			ServiceMonitorPulseInfo serviceMonitorPulseInfo = (ServiceMonitorPulseInfo) serviceMonitorPulseInfos.get(moduleClassID);
			serviceMonitorPulseInfo.registerServiceMonitorFilter(serviceMonitorFilter, reportRateIndex, reportRate);
		}		
		
		resetPulseRate();	

		return reportRate;
	}

	private MonitorListenerInfo getMonitorListenerInfo(MonitorListener monitorListener) {
		for (Iterator i = monitorListenerInfos.iterator(); i.hasNext(); ) {
			MonitorListenerInfo monitorListenerInfo = (MonitorListenerInfo)i.next();

			if (monitorListenerInfo.monitorListener == monitorListener)
				return monitorListenerInfo;
		}

		return null;
	}
		

	public synchronized int removeMonitorListener(MonitorListener monitorListener) {
		int numRemoved = 0;

		for(;;) {	// remove all instances of this listener
			MonitorListenerInfo monitorListenerInfo = getMonitorListenerInfo(monitorListener);

			if (monitorListenerInfo == null)
				break;
			else {
				MonitorFilter monitorFilter = monitorListenerInfo.monitorFilter;
				long reportRate = monitorListenerInfo.reportRate;
				int reportRateIndex = monitorListenerInfo.reportRateIndex;
			
				monitorListenerInfos.remove(monitorListenerInfo);
				numRemoved++;
				filtersPerRate[reportRateIndex]--;

				for (Iterator i = monitorFilter.getModuleClassIDs(); i.hasNext(); ) {
					ModuleClassID moduleClassID = (ModuleClassID)i.next();
					
					ServiceMonitorFilter serviceMonitorFilter = monitorFilter.getServiceMonitorFilter(moduleClassID);	
					ServiceMonitorPulseInfo serviceMonitorPulseInfo = (ServiceMonitorPulseInfo) serviceMonitorPulseInfos.get(moduleClassID);
					serviceMonitorPulseInfo.deregisterServiceMonitorFilter(serviceMonitorFilter, reportRateIndex, reportRate);
				}
			}
		}
		
		resetPulseRate();

		return numRemoved;
	}

	public synchronized MonitorReport getCumulativeMonitorReport(MonitorFilter monitorFilter) throws MonitorFilterException, MonitorException  {
		validateCumulativeMonitorFilter(monitorFilter);

		long beginReportTime = System.currentTimeMillis();

		MonitorReport monitorReport = new MonitorReport(startTime, beginReportTime, true);

		for (Iterator i = monitorFilter.getModuleClassIDs(); i.hasNext(); ) {
			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 supportedModuleClassIDsList = new LinkedList();
			
			for (int i = 0; i < registeredModuleClassIDs.length; i++) {
				ModuleClassID registeredModuleClassID = registeredModuleClassIDs[i];

				if (isLocalMonitoringAvailable(registeredModuleClassID)) 
					supportedModuleClassIDsList.add(registeredModuleClassID);
			}
	
			supportedModuleClassIDs = (ModuleClassID[])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 = (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 = (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 e = serviceMonitorPulseInfos.elements(); e.hasMoreElements(); ) {
			ServiceMonitorPulseInfo serviceMonitorPulseInfo = (ServiceMonitorPulseInfo) e.nextElement();
			int servicePulseRateIndex = serviceMonitorPulseInfo.getPulseRateIndex();

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

		for (Iterator i = monitorListenerInfos.iterator(); i.hasNext(); ) {
			MonitorListenerInfo monitorListenerInfo = (MonitorListenerInfo)i.next();
			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 e = serviceMonitorPulseInfos.elements(); e.hasMoreElements(); ) {
			ServiceMonitorPulseInfo 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 e = serviceMonitorPulseInfos.elements(); e.hasMoreElements(); ) {
			ServiceMonitorPulseInfo serviceMonitorPulseInfo = (ServiceMonitorPulseInfo) e.nextElement();
			ServiceMonitor serviceMonitor = serviceMonitorPulseInfo.serviceMonitor;

			serviceMonitor.destroy();
		}
	}

	private static long start = System.currentTimeMillis();

	private static MonitorListener createMonitorListener(final String label) {
		return new MonitorListener() {
			
			public void processMonitorReport(MonitorEvent monitorEvent) {
				MonitorReport monitorReport = monitorEvent.getMonitorReport();
			}
			
			public void monitorReportingCancelled(MonitorEvent monitorEvent) { }
			public void monitorRequestFailed(MonitorEvent monitorEvent) { }
		};
	}

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

	private static Hashtable monitorManagers = new Hashtable();

	/**
	  * 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 = (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 = (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");
		}
	}
}