launcher.java

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

package org.osu.ogsa.stream.util;

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

import org.osu.ogsa.stream.Stream.StreamServiceGridLocator;
import org.osu.ogsa.stream.Stream.StreamPortType;
import org.osu.ogsa.stream.Stream.StreamServiceLocator;

import org.globus.ogsa.impl.ogsi.GridServiceImpl;
import org.globus.ogsa.GridContext;
import org.globus.ogsa.GridServiceException;
import org.globus.ogsa.utils.AnyHelper;
import org.globus.ogsa.OperationProvider;
import org.globus.ogsa.GridServiceCallback;
import org.globus.ogsa.GridServiceBase;


import org.gridforum.ogsi.ExtensibilityType;
import org.gridforum.ogsi.GridService;
import org.gridforum.ogsi.HandleType;
import org.gridforum.ogsi.OGSIServiceGridLocator;
import org.gridforum.ogsi.ServiceDataValuesType;

import org.gridforum.ogsi.Factory;
import org.gridforum.ogsi.LocatorType;
import org.globus.ogsa.utils.GridServiceFactory;



import java.io.*;
import java.nio.*;
import java.nio.channels.*;
import java.nio.channels.spi.*;
import java.nio.charset.*;
import java.net.*;
import java.util.*;
import java.lang.*;
import org.apache.log4j.*;
import org.apache.axis.client.Stub;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class Launcher{

        static Log log = LogFactory.getLog(Launcher.class.getName());
//	private static int port[] = {58888, 57777, 56666, 55555};
	private static int iPort = 58888;//, 57777, 56666, 55555};

	public boolean start(InputStream in){
		try{
			Integer nInt;
			XMLConfigurator.init(in);
			if((nInt = (Integer)XMLConfigurator.getParameter("numStages")) == null)
			{
				log.fatal("null");
				System.exit(-1);
			}
			else if(nInt.intValue() < 1)
			{
				log.fatal("the number of stage has to be larger than 0");
				System.exit(-1);
			}
//			XMLConfigurator.setParameter("__currentStage", new Integer(1));
			
//			ExtensibilityType content = AnyHelper.getExtensibility(new XMLConfigurator());
			Integer ii = (Integer)XMLConfigurator.getParameter("stages|stage1|numPlacements");
			log.debug(ii);
			int numPlacements = ii.intValue();
//			int numPlacements = ((Integer)XMLConfigurator.getParameter("stages|stage1|numPlacements")).intValue();
			log.debug("number of placements is " + numPlacements);
			
			int i;

			for(i = 1; i <= numPlacements; i ++)
			{
				
			URL GSH = (URL)XMLConfigurator.getParameter("stages|stage1|placement"+i);
			String curDownstreamHandle = GSH.toString()+"/stage1-placement"+i;

//			URL GSH = new URL("http://localhost:8080/ogsa/services/Stream/StreamService");
			//Debug region
			log.debug("****************");
//			log.debug("__currentStage" + ((Integer)XMLConfigurator.getParameter("__currentStage")).intValue());
			log.debug(curDownstreamHandle);
			log.debug("****************");
			
			// Get a reference to the MathService Factory
			OGSIServiceGridLocator gridLocator = new OGSIServiceGridLocator();
			Factory factory = gridLocator.getFactoryPort(GSH);
			GridServiceFactory streamFactory = new GridServiceFactory(factory);
			// Create a new StreamService instance and get a reference to its to its Stream PortType
/*			LocatorType locator = streamFactory.createService(content);
			log.debug(content);*/
			LocatorType locator = streamFactory.createService(new String("stage1-placement"+i));
			log.debug(locator);

			StreamServiceGridLocator streamLocator =   new StreamServiceGridLocator();
			StreamPortType stream = streamLocator.getStreamService(locator);
                        org.apache.axis.client.Stub s = (Stub) stream;
                        s.setTimeout(500000);  // 500 second, in miliseconds
			log.debug(stream);
			//Get the string from InputStream
			/*BufferedReader inReader = new BufferedReader(new InputStreamReader(in));
			char [] buf1 = new char[5000];
			int ret = inReader.read(buf1, 0, 5000);
			if(ret != -1)
			{
				log.error("can't read the end of the configuration file");
				return false;
			}*/
			ByteArrayInputStream inReader = (ByteArrayInputStream)in;
			inReader.reset();
			int nAvail = inReader.available();
			log.debug("navaiL:"+ nAvail);
			byte [] buf1 = new byte[nAvail];
                        for(int f = 0; f < buf1.length; f ++)
				System.out.println((char)buf1[f]);

			inReader.read(buf1, 0, nAvail);
			/*if(ret != -1)
			{
				log.error("can't read the end of the configuration file");
				return false;
			}*/


			String strConfig  = new String(buf1,0, nAvail);
			log.debug(strConfig);

			//String c = stream.setConfigFile(args[0]);
			String c = stream.setConfigInfo(strConfig);
			int b = stream.setCurrentStage(1, i);
//			stream.setSocketPort(port[i-1]);
			stream.setSocketPort(iPort);
			iPort -= 1111;
			stream.setLoadWorkClassFlag(true);
			log.debug(c + b);

			boolean a = stream.start();
			log.debug("result " + a);
			if(a)
			{
				System.out.println(i + "work.....");
			}
			else
			{
				System.out.println("can't start the web services");
			}
			}
			
		}
/*		catch(java.lang.reflect.InvocationTargetException e)
		{
			System.out.println(e.getCause());
		}*/
		catch( Exception e)
		{
		//	System.out.println(e.getCause());
			System.out.println(e);
		}
		return false;
	}
}