hostinfoprovider.java

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

/*
 * HostScriptProvider.java
 *
 * Constructs Host service data from the output of multiple scripts
 *
 */

package org.osu.ogsa.stream.info;

import com.ibm.wsdl.util.xml.DOM2Writer;
import org.globus.ogsa.impl.base.providers.servicedata.AsyncDataProvider;
import org.globus.ogsa.impl.base.providers.servicedata.ServiceDataProviderManager;
import org.globus.ogsa.impl.base.providers.servicedata.ServiceDataProviderDocumentCallback;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.axis.utils.XMLUtils;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import java.io.InputStream;
import java.io.FileInputStream;
import java.util.Enumeration;
import java.util.StringTokenizer;
import java.util.Vector;
import java.lang.reflect.Method;
import org.apache.log4j.*;

public class HostInfoProvider implements AsyncDataProvider, Runnable
{
    private final String name = "HostInfoProvider";
    private final String desc = "Constructs Host service data from the output of multiple scripts";
    private String errorString = "";
    private ServiceDataProviderDocumentCallback callback = null;
    private Thread thread = null;
    private String args = "";
    private String callbackName = "";
    private Object context = null;
    private Document doc = null;
    private long frequencyMillis = 20000;
    private String configFile = "etc/globus-host-providers.conf";
    private int state = PROVIDER_IDLE;

    /** One ExecutionThread for each script that will provide a partial
        result. */
    private Vector components = new Vector();

    /** An ExecutionThread that will provider the top level Host element */
    ExecutionThread hostThread;

    /** used for notifications on new data. We don't make any attempt to
        ensure that a new result is generated for every single notification,
	which might lead to delay of data appearing if a result comes in
	whilst we are not waiting. */
    Object updateLock = new Object();

    private static Category logger = Category.getInstance(HostInfoProvider.class.getName());
    boolean isDebug;
    
    public static void main(String args[]) 
    {
        try 
        {
            // test the synchronous interface
            HostInfoProvider provider =  new HostInfoProvider();
            String argstr = "";
            if (args.length !=0) {
                for (int i=0;i<args.length;i++) {
                    argstr += args[i] + " ";
                }                
            } else {
                argstr = provider.getDefaultArgs();
            }
	    logger.debug(argstr);
	    provider.run();
//            provider.run(argstr, System.out);
        } 
        catch(Exception e) 
        {
            e.printStackTrace();
        }
    }
    
    /** Creates a new instance of AsyncXMLDocumentProvider */
    public HostInfoProvider() 
    {
        this.thread = new Thread(this);
    }    
    
    /**
     * If the provider has a set of default arguments, they can be retrieved
     * with this function.
     */
    public String getDefaultArgs() 
    {
        return "-i " + Long.toString(this.frequencyMillis) + " -f " + this.configFile;
    }
    
    /**
     * Returns a description of the provider's functionality.
     */
    public String getDescription() 
    {
        return this.desc;        
    }
    
    /**
    * The provider should return a string representation of the current error, if any 
    */
    public String getErrorString() 
    {
         return this.errorString;
    }
    
    public int getState() 
    {
        int state;
        synchronized(this) {
            state = this.state;
        }
        return state;
    }
    
    /**
     * Returns the display name of the provider.
     */
    public String getName() 
    {
        return this.name;
    }
        
    public synchronized void setSampleFile(String fileName)
    {
        this.configFile = fileName;
    }
    
    public synchronized void setSampleRefresh(long millis) {
        this.frequencyMillis = millis;
    }   
    
    private synchronized void setState(int state)
    {
        this.state = state;
    }    
    
    /**
     * Triggers the asynchronous execution of the provider, sending the output to
     * the specified callback Object. 
     */   

    public void run(String args, 
                    String callbackName,
                    ServiceDataProviderDocumentCallback callback,
                    Object context) 
        throws Exception 
    {
        // are we already running?
        if (this.getState() == PROVIDER_RUNNING) { 
            return;
        }

        this.parseArgs(args);
        this.context = context;
        this.callback = callback;
        this.callbackName = callbackName;
        this.thread.start();
//	logger.fatal("args" + args +" callbackname"+callbackName+" callback" + callback);

        synchronized (this) { 
            this.wait(15000);
        }
        return;       
    }


    public void run()
    {
        Enumeration en;

        /* this will hold the composite result document */
        Document resultDoc = null;

        /* create a group of threads, one per provider.
        each provider runs in its own thread.
        
        The scripts that we are running are unreliable:

         + they may take a long time to return, or may hang forever.
         + they may not return valid XML.

        If one script goes wrong, it should effect as little as 
        possible -- for example, other scripts should continue
        to be executed.

        */

        /* we must have parsed args by this point, so that config file
	   overrides have happened.
	*/

        Document configDoc;
	try 
	{
	    configDoc = loadConfigFile();
	}
	catch(Exception e)
	{
	    /* if we get an exception loading the config file, give up. */
            this.setState(PROVIDER_ERROR);
            this.errorString = e.getMessage();
            logger.error("Couldn't read config file. Provider aborting: "+errorString);
	    return;
	}
	

        /* now iterate over config and for each provider element, 
	   create an ExecutionThread.
	*/
        Node providersElement = configDoc.getFirstChild(); // gives us the Providers element
	// TODO: potentially the above could return a comment not the
	// providers element, I think? If so, need to handle.

//	if (logger.isDebugEnabled()) {
	    logger.debug("providersElement is "+DOM2Writer.nodeToString(providersElement));
//	}

        Node providerNode = providersElement.getFirstChild();

	while(providerNode!=null)
        {
//	    if (logger.isDebugEnabled()) {
		logger.debug("providerNode is "+DOM2Writer.nodeToString(providerNode));
//	    }
	    if(providerNode instanceof Element && ((Element)providerNode).getNodeName().equals("provider"))
	    {
	        Element providerElement = (Element)providerNode;
	        String commandString = providerElement.getAttribute("command");
	        String refreshString = providerElement.getAttribute("refresh");
	        String stripOuterString = providerElement.getAttribute("stripOuter");
		int refresh = Integer.parseInt(refreshString);
		boolean strip = Boolean.valueOf(stripOuterString).booleanValue();
                logger.debug("Configuring "+commandString);
		ExecutionThread thr = new ExecutionThread(commandString,refresh*1000,strip);
		components.add(thr);
	    }
            providerNode = providerNode.getNextSibling();
	}



        // the following is temporary hack 'till the config file stuff works
/*

        components.add(  new ExecutionThread("GT3/libexec/grid-info-os-uname",100000,false) );
        components.add( new ExecutionThread("GT3/libexec/grid-info-runtime",180000,false));
        components.add( new ExecutionThread("GT3/libexec/grid-info-mem-linux",100000,false));
        components.add( new ExecutionThread("GT3/libexec/grid-info-partial-host GT3/libexec/grid-info-net-linux",300000,true));

*/

	// start up all the threads
        // hostThread.start(); -- no more hostThread

	for(en=components.elements();en.hasMoreElements();)
	{
            ((ExecutionThread)en.nextElement()).start();
	}

        while (getState() != PROVIDER_TERMINATED) 
        {           
            try 
            {                 
                // build the composite document here 
                resultDoc = composeDocument();

                // only pass results up if the 
                // result doc is not null
                // otherwise, just wait and loop again
                if (resultDoc != null) { 
                    this.doc = resultDoc;

                Integer state = new Integer(this.getState());
                String methodName = this.callbackName;
                if (methodName.length() == 0) {
                    methodName = callback.getDefaultCallbackMethodName();
                    }
                    // test the parameter signature for the callback
                    Class[] paramSig = callback.getCallbackParamSig(methodName);
                    if (paramSig == null) {
                        throw new Exception("Fatal - Unable to locate support for the requested callback handler: " + methodName);
                    }                
                    if ((paramSig.length == 3) && 
                        (paramSig[0].isAssignableFrom(this.doc.getClass())) &&
                        (paramSig[1].isAssignableFrom(this.context.getClass())) && 
                        (paramSig[2].isAssignableFrom(state.getClass())))
                    {
                        Object[] params = { this.doc, this.context, state };                                         
    
                        Class callbackClass = this.callback.getClass();
                        Method method = callbackClass.getDeclaredMethod(methodName, paramSig);
    
                        // pass the result document to the specified callback handler
                        method.invoke(this.callback, params);
                    
                        this.setState(PROVIDER_RUNNING);
                        synchronized (this) { 
                            this.notify();
                       }

                    }
                }
                /* now do nothing until someone notifies up that there
		   has been a change. 
		*/
                synchronized(updateLock) { updateLock.wait(); }
                //thread.sleep(this.frequencyMillis);                    
            }
            catch (Exception e)
            {