hostinfoprovider.java.1

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

                    // 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)
            {
                this.setState(PROVIDER_ERROR);
                this.errorString = e.getMessage();
                logger.error(errorString);
                break;
            }
        }
        logger.debug("Thread terminating.");
        synchronized (this) { 
            this.notify();
        }
    }
    
    public void terminate()
    {
        logger.debug("Sending thread termination message."); 
        synchronized (this) {
            this.setState(PROVIDER_TERMINATED);            
        }               
    }
   
    private Document composeDocument()  throws javax.xml.parsers.ParserConfigurationException
    {
try {

    logger.debug("Composing document");
     Document doc = XMLUtils.newDocument(dataFile);
        return doc;
} catch(Exception npex)
{
        return null;
} /*catch(Exception ex) // TODO: better exception handling here
{
        logger.warn("Exception assembing document"+ex);
	return null;
}*/
/* catch(Exception ex)
{
        logger.error("General exception in composeDocument: "+ex);
        throw ex;
} */
    }
    
    /*
     * The SimpleDataProvider methods are implemented below as well
     */


     /** this should perform a one-off query of the provider, but it 
     probably won't work properly as we need to have started up most
     of the async stuff. */
    
    public void update(String argStr) throws Exception
    {              
        parseArgs(argStr);

        synchronized(this)
        {
            Document resultDoc = composeDocument();
            if (resultDoc != null)
                this.doc = resultDoc;
        }        
    }
    
    protected void parseArgs(String args) throws Exception
    {
        StringTokenizer toker = new StringTokenizer(args);
        String token = "";
        while (toker.hasMoreTokens()) 
        {
           token = toker.nextToken();
           if (token.compareToIgnoreCase("-i") == 0) {
               this.frequencyMillis = Long.parseLong(toker.nextToken());
               continue;
           }
           else if (token.compareToIgnoreCase("-f") == 0) {
                setSampleFile(toker.nextToken());
                continue;
           }
           else throw new Exception("Unknown Argument: " + token);          
        }        
    }
    
    public void output(java.io.OutputStream outStream) throws Exception
    {
        synchronized(this)
        {
            XMLUtils.DocumentToStream(this.doc,outStream);
        }
    }
    
    public void run(String args, java.io.OutputStream outStream) throws Exception 
    {
        synchronized(this)
        {
            try {
                this.update(args);
                this.output(outStream);
            }
            catch (Exception e)
            {
                logger.error(e.getMessage());
                throw e;
            }
        }
    }    

    private Document loadConfigFile() throws Exception
        {   
        InputStream input = null;
        Document doc = null;
        try
        {   
            input = new FileInputStream(this.configFile);
                if (input != null) {
                doc = XMLUtils.newDocument(input);
            }
            }
            catch (Exception e)
            {   
                if (input != null) {
                    input.close();
                }
                this.errorString = "Config file loading error: " + e.getMessage();
                logger.warn(errorString);
            }
            return doc;
        }




/** One ExecutionThread is created to handle each script. */
class ExecutionThread extends Thread
{

    /** refresh period, in ms */
    private long period = 20000;
    private String commandLine = null;

    /** contains the most recent complete result. must either be null
    or a complete usable document.
    */

    Document recentResult = null;

    /** if true, we'll strip of the outer element of the result and 
        add in each contained element to the composite result
	separately. */

    boolean stripOuter = false;

    ExecutionThread(String commandLine, long period, boolean stripOuter)
    {
        this.commandLine = commandLine;
        this.period = period;
	this.stripOuter = stripOuter;
        logger.debug("Creating ExecutionThread for "+commandLine+" with period of "+period+"ms");
    }

    public void run()
    {
        /* we'll fall out of this loop (and hence the thread will go away)
           when the enclosing provider finishes or dies. */

        while (getState() != PROVIDER_TERMINATED && getState() != PROVIDER_ERROR) 
        {

            logger.debug("Executing "+commandLine);
            Runtime runtime = Runtime.getRuntime();

            try 
            {
                int scriptExitCode;
                Process process;

                // start the script
                process = runtime.exec(commandLine);


                // parse the results and store for later
                logger.debug("Parsing results of "+commandLine);

                InputStream processInputStream = process.getInputStream();
                this.recentResult = XMLUtils.newDocument(processInputStream);

                // TODO: is it the case that builder.parse can return 
                // non-null / non-valid Documents ?
//		if (logger.isDebugEnabled()) {
		    logger.debug("Parsed result of "+commandLine+" is "+DOM2Writer.nodeToString(this.recentResult));
//		}

                // TODO: there should be a timeout on this so that
                // if a script never exits, we eventually kill it 
                // (and it'll run again)

                /* waiting for the script to exit means that
                   multiple instances of long running / hanging 
                   scripts won't build up.
                */

                scriptExitCode = process.waitFor();
                processInputStream.close();
                process.destroy();

                /* examine error code

                   following standard unix semantics, we expect a zero 
                   exit code to indicate successful completion.

                   if non-zero, we log a warning, but other than that
                   we don't change the way that we handle the output.
                */
                if(0!=scriptExitCode)
                {
                    logger.warn(commandLine+" returned a non-zero exit code: "+scriptExitCode);
                }

                // notify upwards that a change has occurred
                synchronized(updateLock) { updateLock.notifyAll(); }

            }
            catch (Exception e) 
            {
                String errorString = "Execution error: " + e.getMessage ()+ " while processing "+commandLine;
                logger.error(errorString);
                // can't send this exception anywhere
                // throw new Exception(errorString);
            }
            try 
            {


                // now wait a while before we go round again
                logger.debug("Sleeping after "+commandLine);
                thread.sleep(this.period);
            } 
            catch (Exception e) 
            {
                String errorString = "Sleep error: " + e.getMessage ()+ " while processing "+commandLine;
                logger.error(errorString);
                // can't send this exception anywhere
                // throw new Exception(errorString);
            }

        }

    }

}

}