resquerytask.java

本人历尽千辛万苦找的clustream中的jar包

package org.osu.ogsa.stream.util;

//import org.osu.ogsa.stream.services.*;
import org.osu.ogsa.stream.util.xmlconfig.*;

import java.io.*;
import java.util.*;
import java.lang.Object;
import javax.swing.Timer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.awt.*;
import java.awt.event.*;
import org.w3c.dom.Element;

public class ResQueryTask implements ActionListener{
//	private StreamMonitorImpl stream;
        private static Log log = LogFactory.getLog(ResQueryTask.class.getName());
	
	private int intAvailMem = 0; 
	private int intAveLoad = 0;
	private int intCPUSpeed = 0;
	private Object synObject;
	private int DELAY_MEM = 1000 * 10; //10 sec. 
	private int DELAY_LOAD = 1000 * 60; //1 min
	private String service_handle = "http://localhost:8080/ogsa/services/base/index/IndexService";
	private String strXPathExpLoad = "//ce:Host/ce:ProcessorLoad";
	private	String strXPathExpAvailMem = "//ce:Host/ce:MainMemory";
	private	String strXPathExpCPUSpeed = "//ce:Host/ce:Processor";

	private Timer timer_mem;
	private Timer timer_load;
	private Element [] elementRes;
/*	public ResQueryTask(StreamMonitorImpl stream)
	{
		this.stream = stream;
	}*/

	public ResQueryTask()
	{
		timer_load = new Timer(DELAY_LOAD, this);
		timer_mem = new Timer(DELAY_MEM, this);
		synObject = new Object();
	}

	public void start()
	{
		//start the timer
		timer_load.start();
		timer_mem.start();
	}
	
	//3 min elapse
	public void actionPerformed(ActionEvent e)
	{
		if(timer_load.equals(e.getSource()))
			calculateAveCpuLoad_Speed();
		else if(timer_mem.equals(e.getSource()))
			calculateAveMem();
		else
			log.info("wrong timer!!!");
	}

	private void calculateAveCpuLoad_Speed()
	{
		//Query the resources from index services
/*		String location = stream.getCurrentInstanceLocation();
		if(location == null)
		{
			log.fatal("can't find where the current instance is running");
			stop();
		}*/ 
		
//		log.debug(location);
		//Try the service handle with the ip address 
//		String service_handle = "http://" + location + ":8080/ogsa/services/base/index/IndexService";

		//Query the load of cpu
		elementRes = QueryServiceDataByXPath.Query("HostScript", service_handle, strXPathExpLoad);
		if(elementRes == null)
		{
			log.error("can't query the processor's load");
		}
		else
		{
			int temp;
			temp = (Integer.valueOf(elementRes[0].getAttribute("ce:Last1Min"))).intValue();

			if(intAveLoad > 0)
			{
				temp += intAveLoad;
				temp /= 2;
			}

			synchronized(synObject)
			{
				intAveLoad = temp;
			}
		}

		//Query the frequery of cpu
		if(intCPUSpeed > 0)
			return;
		else
		{
			elementRes = QueryServiceDataByXPath.Query("HostScript", service_handle, strXPathExpCPUSpeed);
			if(elementRes == null)
				log.error("can't query the processor's speed");
			else
				synchronized(synObject)
				{
					intCPUSpeed = (Integer.valueOf(elementRes[0].getAttribute("ce:ClockSpeed"))).intValue();
				}
		}

	}
	
	private void calculateAveMem()
	{
		//Query the available memory
		int temp;
		Element [] elementRes = QueryServiceDataByXPath.Query("HostScript", service_handle, strXPathExpAvailMem);
		if(elementRes == null)
		{
			log.error("can't query the processor's available mem");
			return;
		}
		else
			temp = (Integer.valueOf(elementRes[0].getAttribute("ce:RAMAvailable"))).intValue();

		if(intAvailMem > 0)
		{
			temp += intAvailMem;
			temp = temp/2;
		}
		
		synchronized(synObject)
		{

			intAvailMem = temp;
		}
	}

	public int getAvailMem()
	{
		int temp;
		synchronized(synObject)
		{
			temp = intAvailMem;
			intAvailMem = 0;
		}
		return temp;
	}

	public int getCPUSpeed()
	{
		int temp;
		synchronized(synObject)
		{
			temp = intCPUSpeed;
			intCPUSpeed = 0;
		}
		return temp;
	}

	public int getAveCPULoad()
	{
		int temp;
		synchronized(synObject)
		{
			temp = intAveLoad;
			intAveLoad = 0;
		}
		return temp;
	}
}