Menu
▾
▴
[Cogkit-svn] SF.net SVN: cogkit: [1852] trunk/cyberaide/src/axis2
|
From: <je...@us...> - 2007-12-07 05:39:58
|
Revision: 1852
http://cogkit.svn.sourceforge.net/cogkit/?rev=1852&view=rev
Author: jenvor
Date: 2007-12-06 21:39:56 -0800 (Thu, 06 Dec 2007)
Log Message:
-----------
add state service, modify executor implementation
Modified Paths:
--------------
trunk/cyberaide/src/axis2/axis2ws_client/SimpleHandlerClient.java
trunk/cyberaide/src/axis2/axis2ws_client/SimpleHandlerInterface.java
trunk/cyberaide/src/axis2/axis2ws_server/SimpleHandlerInterface.java
trunk/cyberaide/src/axis2/axis2ws_server/SimpleHandlerInterfaceSkeleton.java
Added Paths:
-----------
trunk/cyberaide/src/axis2/axis2ws_server/StatusServiceInterface.java
trunk/cyberaide/src/axis2/axis2ws_server/StatusServiceInterfaceSkeleton.java
Modified: trunk/cyberaide/src/axis2/axis2ws_client/SimpleHandlerClient.java
===================================================================
--- trunk/cyberaide/src/axis2/axis2ws_client/SimpleHandlerClient.java 2007-12-05 05:06:57 UTC (rev 1851)
+++ trunk/cyberaide/src/axis2/axis2ws_client/SimpleHandlerClient.java 2007-12-07 05:39:56 UTC (rev 1852)
@@ -14,30 +14,72 @@
import java.io.File;
import java.io.BufferedWriter;
+import java.io.BufferedReader;
import java.io.FileWriter;
+import java.io.FileReader;
+import java.io.FileOutputStream;
import java.io.FileInputStream;
+import java.io.StringWriter;
+
import java.util.Properties;
+import java.util.ArrayList;
+import java.util.HashMap;
import org.cogkit.cyberaide.axis2ws.SimpleHandlerInterfaceStub.Echo;
import org.cogkit.cyberaide.axis2ws.SimpleHandlerInterfaceStub.EchoResponse;
import org.cogkit.cyberaide.axis2ws.SimpleHandlerInterfaceStub.CmdCOGExecute;
import org.cogkit.cyberaide.axis2ws.SimpleHandlerInterfaceStub.CmdCOGExecuteResponse;
+import org.cogkit.cyberaide.axis2ws.SimpleHandlerInterfaceStub.CmdWFSubmit;
+import org.cogkit.cyberaide.axis2ws.SimpleHandlerInterfaceStub.CmdWFSubmitResponse;
+import javax.xml.parsers.DocumentBuilder;
+import javax.xml.parsers.DocumentBuilderFactory;
+import javax.xml.transform.Transformer;
+import javax.xml.transform.TransformerFactory;
+import javax.xml.transform.dom.DOMSource;
+import javax.xml.transform.stream.StreamResult;
-public class SimpleHandlerClient{
- private String prefix = "(This is done by using Web servie)\n";
- private String propertiesFile = "webservice.properties";
- private String wsendpoint = "";
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
+import org.w3c.dom.Node;
+import org.cogkit.cyberaide.axis2ws.StatusServiceInterfaceStub;
+import org.cogkit.cyberaide.axis2ws.StatusServiceInterfaceStub.AddWorkflowFromString;
+import org.cogkit.cyberaide.axis2ws.StatusServiceInterfaceStub.GetJSONStatusByUID;
+import org.cogkit.cyberaide.axis2ws.StatusServiceInterfaceStub.GetJSONStatusByUIDResponse;
+
+public class SimpleHandlerClient implements SimpleHandlerClientInterface{
+ private static String prefix = "(This is done by using Web service)\n";
+ //private static String propertiesFile = "/home/tomcat/apache-tomcat-6.0.14/webapps/webservice.properties";
+ private static String propertiesFile = "/home/tomcat/apache-tomcat-6.0.14/webapps/agent.properties";
+ private static String wsendpoint = "";
+ private static StatusServiceNotifier _notifier = new StatusServiceNotifier();
+ private static AgentWFDB _agentDb = null;
+ private static SimpleHandlerInterfaceStub stub = null;
+
public SimpleHandlerClient(){
try{
- Properties prop = new Properties();
- prop.load( new FileInputStream( propertiesFile ) );
- String endpoint = prop.getProperty( "serviceendpoint" );
- if( endpoint == null ){
- System.out.println( "you must specify 'serviceendpoint' property in file service.properties" );
- }else{
- wsendpoint = endpoint;
+ //open property file specified by propertiesFile
+ File propfile = new File(propertiesFile);
+ if( propfile.exists() == false ){
+ System.out.println("property file: "+propertiesFile+" does not exist!");
+ return ;
+ }
+
+ // get service end point
+ Properties prop = new Properties();
+ prop.load( new FileInputStream( propertiesFile ) );
+ String endpoint = prop.getProperty( "serviceendpoint" );
+ if( endpoint == null ){
+ System.out.println( "you must specify 'serviceendpoint' property in file service.properties" );
+ }else{
+ wsendpoint = endpoint;
+ stub = new SimpleHandlerInterfaceStub(wsendpoint);
+ }
+
+ if( _agentDb == null ){
+ String dbfile = prop.getProperty("agentdb");
+ _agentDb = new AgentWFDB(dbfile);
}
}catch( Exception e ){
System.out.println( e.toString() );
@@ -57,9 +99,6 @@
br.write( input );
br.close();
- //call the corresponding web service
- //SimpleHandlerInterfaceStub stub = new SimpleHandlerInterfaceStub("http://localhost:8080/axis2/services/SimpleHandler");
- SimpleHandlerInterfaceStub stub = new SimpleHandlerInterfaceStub(wsendpoint);
Echo req = new Echo();
req.setParam0( input );
EchoResponse resp = new EchoResponse();
@@ -72,27 +111,33 @@
}
/**
- * @param input workflow description which is represented by a string
+ * @param workflow workflow description which is represented by a string
* This invokes remote 'cmdCOGExecute' web service which will return a string.
* <b>Note</b>: the workflow description from user is logged in file <b>/tmp/workflow.xml</b>.
*/
public String cmdCOGExecute( String workflow ){
String output = "";
try {
+ String oldWF = workflow;
+ WorkflowTransformer transformer = new WorkflowTransformer();
+ transformer.LoadFromString(workflow);
+ workflow = transformer.getNewWF();
+
String filename = "/tmp/workflow.xml";
File recordfile = new File(filename);
BufferedWriter br = new BufferedWriter(new FileWriter(recordfile));
br.write(workflow);
br.close();
- //call the corresponding web service
- //SimpleHandlerInterfaceStub stub = new SimpleHandlerInterfaceStub("http://localhost:8080/axis2/services/SimpleHandler");
- SimpleHandlerInterfaceStub stub = new SimpleHandlerInterfaceStub(wsendpoint);
+ //send this workflow to status server
+ _notifier.addWorkFlowFromString("user1", "file1", oldWF);
+
CmdCOGExecute req = new CmdCOGExecute();
req.setParam0( workflow );
CmdCOGExecuteResponse resp = new CmdCOGExecuteResponse();
resp = stub.cmdCOGExecute( req );
output = resp.get_return();
+
} catch (IllegalThreadStateException e) {
return e.toString();
} catch (Exception e){
@@ -100,4 +145,482 @@
}
return prefix + output;
}
+ /**
+ */
+ public String WFSubmit(String username, String workflow){
+ String result = "", myprefix = "(This is done by using web service with status monitoring enabled)\n";
+ if( username.compareTo("")==0 ){
+ result = "You must specify your username!!";
+ return result;
+ }
+ System.out.println("this id received by WFSubmit method");
+ //TODO
+ //check whether the user exists in the system
+ //how to generate the wfid
+ String output = "", wfid = _agentDb.addWF( username );
+ try {
+ String oldWF = workflow;
+ WorkflowTransformer transformer = new WorkflowTransformer();
+ transformer.LoadFromString(workflow);
+ workflow = transformer.getNewWF();
+
+ //send this workflow to status server
+ _notifier.addWorkFlowFromString(username, wfid, oldWF);
+
+ CmdWFSubmit req = new CmdWFSubmit();
+ req.setParam0( username );
+ req.setParam1( wfid );
+ req.setParam2( workflow );
+ CmdWFSubmitResponse resp = stub.cmdWFSubmit( req );
+ output = resp.get_return();
+ } catch (IllegalThreadStateException e) {
+ return e.toString();
+ } catch (Exception e){
+ return e.toString();
+ }
+ return myprefix + output;
+ }
+
+ public String getStatusByUID(String struid){
+ return _notifier.getStatusByUID(struid);
+ }
}
+
+class WorkflowTransformer{
+ private String newWorkflow = "";
+ public String getNewWF(){
+ return newWorkflow;
+ }
+
+ public void LoadFromFile(String filename) {
+ try {
+ File karajanfile = new File(filename);
+ if (karajanfile.exists() == false) {
+ System.out.println("The karajan workflow file " + filename
+ + " does not exist!!");
+ return;
+ }
+ String content = "", line ="";
+ BufferedReader br = new BufferedReader(new FileReader(karajanfile));
+ boolean isFirstLine = true;
+ while( (line=br.readLine())!=null ){
+ if( isFirstLine ){
+ isFirstLine = false;
+ }else
+ content +="\n";
+ content += line;
+ }
+// buildTree(content);
+ LoadFromString(content);
+ } catch (Exception e) {
+ System.out.println(e);
+ }
+ }
+ public void LoadFromString(String strworkflow) {
+ String tempFilename = "_tempxx345";
+ File tempfile = new File(tempFilename);
+ BufferedWriter fw = null;
+ try {
+ fw = new BufferedWriter(new FileWriter(tempfile));
+ fw.write(strworkflow);
+ fw.close();
+ buildTree(tempFilename);
+ tempfile.delete();
+ } catch (Exception e) {
+ tempfile.delete();
+ System.out.println(e);
+ }
+ }
+ protected void buildTree(String workflowfilename){
+ try{
+ DocumentBuilderFactory domfactory = DocumentBuilderFactory.newInstance();
+ domfactory.setNamespaceAware(false);
+ DocumentBuilder builder = domfactory.newDocumentBuilder();
+
+ Document doc = builder.parse(new FileInputStream(workflowfilename));
+ Element root = doc.getDocumentElement();
+
+ //transformSubTree(root, doc);
+ transformSubTreeWithPath(root,doc,"");
+
+ /*
+ String newWF = "test.xml";
+ File newfile = new File(newWF);
+ if( newfile.exists() )
+ newfile.delete();
+ newfile.createNewFile();
+ */
+
+ DOMSource src = new DOMSource(doc);
+ TransformerFactory transformerfactory = TransformerFactory.newInstance();
+ Transformer transformer = transformerfactory.newTransformer();
+ //transformer.transform(src, new StreamResult(new FileWriter(newfile)));
+ StringWriter sw = new StringWriter();
+ transformer.transform(src, new StreamResult(sw));
+ newWorkflow = sw.toString();
+ } catch (Exception e) {
+ System.out.println(e);
+ }
+ }
+ protected void transformSubTree(Node root, Document doc){
+ if(root.getChildNodes().getLength()==0)
+ return;
+ Node parent = root;
+ Node node = root.getFirstChild();
+ boolean isfirstChild = true;
+ String previousNode ="";
+
+ do{
+ if( node.getNodeType() == Node.TEXT_NODE )
+ continue;
+ if( node.getNodeType() == Node.ELEMENT_NODE){
+ if( node.getNodeName().compareToIgnoreCase("include")==0)
+ continue;
+ String nodename = node.getNodeName();
+ if( nodename.trim().compareToIgnoreCase("execute")==0 ){
+ nodename = "job:"+nodename+"("+((Element)node).getAttribute("executable")+") ";
+ }else
+ nodename = "job:"+nodename;
+
+ if( isfirstChild ){
+ isfirstChild = false;
+ Element newChild = doc.createElement("echo");
+ newChild.setAttribute("message", nodename+" started");
+ parent.insertBefore(newChild, node);
+ Node newline = doc.createTextNode("\n");
+ parent.insertBefore(newline, node);
+ }else{
+ Element newChild = doc.createElement("echo");
+ newChild.setAttribute("message", previousNode+" completed");
+ parent.insertBefore(newChild, node);
+ Node newline = doc.createTextNode("\n");
+ parent.insertBefore(newline, node);
+ newChild = doc.createElement("echo");
+ newChild.setAttribute("message", nodename+" started");
+ parent.insertBefore(newChild, node);
+ newline = doc.createTextNode("\n");
+ parent.insertBefore(newline, node);
+ }
+ previousNode = nodename;
+ transformSubTree(node, doc);
+ }
+ }while( (node=node.getNextSibling())!=null);
+
+ Element newChild = doc.createElement("echo");
+ newChild.setAttribute("message", previousNode+" completed");
+ parent.appendChild(newChild);
+ }
+
+ protected void transformSubTreeWithPath(Node root, Document doc, String path){
+ if(root.getChildNodes().getLength()==0)
+ return;
+ Node parent = root;
+ Node node = root.getFirstChild();
+ boolean isfirstChild = true;
+ String previousNode ="", basepath = path, localpath="", oldpath="", oldEchoMsg="";
+ String echoMsg = "";
+ int childindex = 0;
+ do{
+ if( node.getNodeType() == Node.TEXT_NODE )
+ continue;
+ if( node.getNodeType() == Node.ELEMENT_NODE){
+ if( node.getNodeName().compareToIgnoreCase("include")==0){
+ childindex ++;
+ continue;
+ }
+ childindex ++;
+
+ String nodename = node.getNodeName();
+ if( nodename.trim().compareToIgnoreCase("execute")==0 )
+ nodename = nodename+"("+((Element)node).getAttribute("executable")+") ";
+
+ localpath = path+"/"+childindex;
+
+ if( isfirstChild ){
+ isfirstChild = false;
+ Element newChild = doc.createElement("echo");
+ echoMsg = localpath+"|job:"+nodename;
+ newChild.setAttribute("message", echoMsg+" started|1");
+ parent.insertBefore(newChild, node);
+ Node newline = doc.createTextNode("\n");
+ parent.insertBefore(newline, node);
+ }else{
+ Element newChild = doc.createElement("echo");
+ newChild.setAttribute("message", oldEchoMsg+" completed|2");
+ parent.insertBefore(newChild, node);
+
+ Node newline = doc.createTextNode("\n");
+ parent.insertBefore(newline, node);
+
+ echoMsg = localpath+"|job:"+nodename;
+ newChild = doc.createElement("echo");
+ newChild.setAttribute("message", echoMsg+" started|1");
+ parent.insertBefore(newChild, node);
+
+ newline = doc.createTextNode("\n");
+ parent.insertBefore(newline, node);
+ }
+ oldEchoMsg = echoMsg;
+ previousNode = nodename;
+ transformSubTreeWithPath(node, doc, localpath);
+ }
+ }while( (node=node.getNextSibling())!=null);
+
+ Element newChild = doc.createElement("echo");
+ newChild.setAttribute("message", oldEchoMsg+" completed|2");
+ parent.appendChild(newChild);
+ }
+}
+
+class StatusServiceNotifier{
+ private String _wsendpoint = "";
+ private StatusServiceInterfaceStub _stub = null;
+
+ public StatusServiceNotifier(){
+ try{
+ _wsendpoint = "http://156.56.104.196:8080/axis2/services/StatusServiceInterface";
+ _stub = new StatusServiceInterfaceStub(_wsendpoint);
+ } catch(Exception e){
+ System.out.println(e);
+ }
+ }
+
+ public String getStatusByUID(String struid){
+ try{
+ if( _stub == null ) return "_stub is null";
+ GetJSONStatusByUID req = new GetJSONStatusByUID();
+ req.setParam0( struid );
+ GetJSONStatusByUIDResponse resp = new GetJSONStatusByUIDResponse();
+ resp = _stub.getJSONStatusByUID(req);
+ return resp.get_return();
+ }catch(Exception e){
+ return e.toString();
+ }
+ }
+
+ public void addWorkFlowFromString(String userid, String wfid, String workflow){
+ try{
+ if( _stub == null ) return ;
+
+ String filename = "/tmp/statusservice_agent_client.test";
+ File file = new File(filename);
+ file.createNewFile();
+ BufferedWriter bw = new BufferedWriter(new FileWriter(file));
+
+ bw.write("add a new workflow\n");
+ //add a workflow
+ /*
+ File currentDir = new File(".");
+ if( true )
+ return currentDir.getAbsolutePath();
+ */
+
+ AddWorkflowFromString addWF1 = new AddWorkflowFromString();
+ addWF1.setParam0(userid);
+ addWF1.setParam1(wfid);
+ addWF1.setParam2(workflow);
+ _stub.addWorkflowFromString(addWF1);
+
+ bw.close();
+ }catch(Exception e){
+ System.out.println(e);
+ }
+ }
+}
+
+/**
+ * Format of the db file:
+ * username1 workflow1
+ * username2 workflow1
+ * username1 workflow2
+ * ... ...
+ */
+class AgentWFDB{
+ private ArrayList<Pair> _db = new ArrayList<Pair>();
+ private HashMap<String, Integer> _userwf = new HashMap<String,Integer>();
+
+ private String _dbfilename = "";//source file
+ private BufferedWriter _bw = null;
+ private boolean _load = false;
+
+ public AgentWFDB(String dbfilename){
+ _dbfilename = dbfilename;
+ File dbfile = new File(dbfilename);
+ if( dbfile.exists() )
+ _load = loadFromFile(dbfilename);
+ }
+ private void addPair(StringPair pair){
+ _db.add(pair);
+ }
+ private void addPair(String first, String second){
+ _db.add( new StringPair(first, second) );
+ }
+ /**
+ * This function first generates a unique workflow id for the user
+ * and add the entry to db.
+ * @param userid
+ * @return
+ */
+ public String addWF(String userid){
+ try{
+ File file = new File("/tmp/client.test");
+ file.createNewFile();
+ BufferedWriter bw = new BufferedWriter( new FileWriter( file ) );
+ bw.write( userid );
+ bw.close();
+ }catch(Exception e){
+ }
+ Integer cn = _userwf.get(userid);
+ int count;
+ if( cn == null ){
+ _userwf.put(userid, 1);
+ count = 1;
+ }else{
+ count = cn+1;
+ _userwf.put(userid, count);
+ }
+ String wfid = userid+"_"+count;
+ addPair(userid, wfid);
+ serialize2file();
+ return wfid;
+ }
+ public boolean contains(String username, String wfid){
+ StringPair pair = new StringPair(username, wfid);
+ return this.contains(pair);
+ }
+ public boolean contains(StringPair pair){
+ for( int i = 0 ; i < _db.size(); i++ ){
+ if( _db.get(i).isEqual(pair) )
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * get workflow list owned by a specified user.
+ * @param userid
+ * @return JSON string
+ */
+ public String getJSONByUserID(String userid){
+ String output = "{\"user\":\""+userid+"\", \"workflow\":[";
+ boolean isfirst = true;
+ for(int i = 0 ; i < _db.size(); i++){
+ if( ((String)_db.get(i).getFirst()).compareTo(userid)==0 ){
+ if( isfirst ){
+ isfirst = false;
+ }else
+ output += ",";
+ output += "\""+(String)_db.get(i).getSecond()+"\"";
+ }
+ }
+ output += "]}";
+ return output;
+ }
+
+ private boolean loadFromFile(String filename){
+ BufferedReader br = null;
+ String line = "";
+ try {
+ br = new BufferedReader(new FileReader(filename));
+ while( (line = br.readLine()) != null){
+ StringPair pair = new StringPair(line);
+ Integer cn = _userwf.get(pair.getFirst());
+ int count;
+ if( cn == null )
+ count = 1;
+ else
+ count = cn + 1;
+ _userwf.put(pair.getFirst(), count);
+ _db.add( pair );
+ }
+ br.close();
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ public String serialize2string(){
+ String output = "";
+ for(int i=0; i<_db.size(); i++){
+ output += _db.get(i).serialize() +"\n";
+ }
+ return output;
+ }
+ public void serialize2file(){
+ try {
+ if( !_load ){
+ File file = new File(_dbfilename);
+ if( !file.exists() )
+ file.createNewFile();
+ if (_bw == null) {
+ _bw = new BufferedWriter(new FileWriter(_dbfilename));
+ }
+ _load = true;
+ }
+ String output = serialize2string();
+ if (_bw == null) {
+ _bw = new BufferedWriter(new FileWriter(_dbfilename));
+ }
+ _bw.write(output);
+ _bw.close();
+ _bw = null;
+ } catch (Exception e) {
+ System.out.println(e);
+ }
+ }
+}
+
+abstract class Pair{
+ abstract Object getFirst();
+ abstract Object getSecond();
+ abstract void setFirst(Object first);
+ abstract void setSecond(Object second);
+ abstract String serialize();
+ abstract void deserialize(String str);
+ abstract boolean isEqual(Pair pair);
+}
+
+class StringPair extends Pair{
+ private String _first = "";
+ private String _second = "";
+ public StringPair(String first, String second){
+ _first = first;
+ _second = second;
+ }
+ public StringPair(String str){
+ deserialize(str);
+ }
+ public void setFirst(Object first){
+ if( first instanceof String)
+ _first = (String)first;
+ }
+ public void setSecond(Object second){
+ if( second instanceof String)
+ _second = (String)second;
+ }
+ public String getFirst(){
+ return _first;
+ }
+ public String getSecond(){
+ return _second;
+ }
+ public String serialize(){
+ return _first+"\t"+_second;
+ }
+ public void deserialize(String str){
+ int index = str.indexOf('\t');
+ _first = str.substring( 0, index);
+ _second = str.substring(index+1);
+ }
+ public boolean isEqual( Pair p ){
+ if( p instanceof StringPair){
+ StringPair pair = (StringPair)p;
+ String first = pair.getFirst();
+ String second = pair.getSecond();
+ return (_first.compareTo(first)==0)&&(_second.compareTo(second)==0);
+ }else{
+ return false;
+ }
+ }
+}
Modified: trunk/cyberaide/src/axis2/axis2ws_client/SimpleHandlerInterface.java
===================================================================
--- trunk/cyberaide/src/axis2/axis2ws_client/SimpleHandlerInterface.java 2007-12-05 05:06:57 UTC (rev 1851)
+++ trunk/cyberaide/src/axis2/axis2ws_client/SimpleHandlerInterface.java 2007-12-07 05:39:56 UTC (rev 1852)
@@ -8,4 +8,11 @@
interface SimpleHandlerInterface{
String echo( String input );
String cmdCOGExecute( String workflow );
+ /**
+ * This method submits a workflow to CogKit.
+ * Notice: the user who submits the workflow is identified by <b><i>uid</i></b>
+ * the workflow is uniquely identified by wfid so that different workflows
+ * submitted by the same user can be distinguished.
+ */
+ String cmdWFSubmit(String uid, String wfid, String workflow);
}
Modified: trunk/cyberaide/src/axis2/axis2ws_server/SimpleHandlerInterface.java
===================================================================
--- trunk/cyberaide/src/axis2/axis2ws_server/SimpleHandlerInterface.java 2007-12-05 05:06:57 UTC (rev 1851)
+++ trunk/cyberaide/src/axis2/axis2ws_server/SimpleHandlerInterface.java 2007-12-07 05:39:56 UTC (rev 1852)
@@ -8,4 +8,11 @@
interface SimpleHandlerInterface{
String echo( String input );
String cmdCOGExecute( String workflow );
+ /**
+ * This method submits a workflow to CogKit.
+ * Notice: the user who submits the workflow is identified by <b><i>uid</i></b>
+ * the workflow is uniquely identified by wfid so that different workflows
+ * submitted by the same user can be distinguished.
+ */
+ String cmdWFSubmit(String uid, String wfid, String workflow);
}
Modified: trunk/cyberaide/src/axis2/axis2ws_server/SimpleHandlerInterfaceSkeleton.java
===================================================================
--- trunk/cyberaide/src/axis2/axis2ws_server/SimpleHandlerInterfaceSkeleton.java 2007-12-05 05:06:57 UTC (rev 1851)
+++ trunk/cyberaide/src/axis2/axis2ws_server/SimpleHandlerInterfaceSkeleton.java 2007-12-07 05:39:56 UTC (rev 1852)
@@ -32,11 +32,58 @@
import org.globus.common.CoGProperties;
import org.globus.util.Util;
+
+/* following imported packages are used by StatusService client */
+import org.cogkit.cyberaide.axis2ws.StatusServiceInterfaceStub;
+import org.cogkit.cyberaide.axis2ws.StatusServiceInterfaceStub.*;
+
/**
* SimpleHandlerInterfaceSkeleton java skeleton for the axisService
*/
-public class SimpleHandlerInterfaceSkeleton implements SimpleHandlerInterfaceSkeletonInterface {
+public class SimpleHandlerInterfaceSkeleton implements SimpleHandlerInterfaceSkeletonInterface {
+/* private String _uid = "user1";
+ private String _wfid = "file1";
+ private String _workflow ="";
+*/
+ private static String _statusFilename = "/tmp/status.txt";
+ private static String _statusserviceendpoint = "";
+ private static String _configFilename = "/home/tomcat/apache-tomcat-6.0.14/webapps/executor.properties";
+ private static StatusServiceClient _statusclient = null;
+
+ public SimpleHandlerInterfaceSkeleton(){
+ super();
+ try{
+ //open property file specified by propertiesFile
+ File propfile = new File(_configFilename);
+ if( propfile.exists() == false ){
+ System.out.println("property file: "+_configFilename+" does not exist!");
+ return ;
+ }
+
+ // get status service end point
+ Properties prop = new Properties();
+ prop.load( new FileInputStream( _configFilename ) );
+ String endpoint = prop.getProperty( "statusserviceendpoint" );
+ if( endpoint == null ){
+ System.out.println( "you must specify 'serviceendpoint' property in file service.properties" );
+ }else{
+ _statusserviceendpoint = endpoint;
+ if( _statusclient == null )
+ _statusclient = new StatusServiceClient(_statusserviceendpoint);
+ }
+ }catch(Exception e){
+ System.out.println(e);
+ }
+ }
+ /**
+ * Currently, the events generated by CoGKit during workflow execution is
+ * written into a file specified by variable _statusFilename. When end users
+ * check status, a routine reads this file and sends the content to clients.
+ * If the workflow is finished, another file is created to indicate it.
+ * I can not think of a better way now to tell status checker the completion
+ * of execution of workflow.
+ */
/**
* Auto generated method signature
* @param cmdCOGExecute0
@@ -44,20 +91,39 @@
public org.cogkit.cyberaide.axis2ws.CmdCOGExecuteResponse cmdCOGExecute(
org.cogkit.cyberaide.axis2ws.CmdCOGExecute cmdCOGExecute0) {
String workflow = cmdCOGExecute0.getParam0();
+ /* actually we invoke cmdWFSubmit !!! */
+ if( true ){
+ String result = "";
+ CmdWFSubmit req = new CmdWFSubmit();
+ req.setParam0("user1");
+ req.setParam1("file1");
+ req.setParam2(workflow);
+ CmdWFSubmitResponse resp = cmdWFSubmit( req );
+ CmdCOGExecuteResponse reply = new CmdCOGExecuteResponse();
+ reply.set_return( resp.get_return() );
+ return reply;
+ }
+
Process wfproc = null;
String output = "";
int exit = -1;
try {
- //First, get the proxy certificate from the myproxy server
- getCertFromMyproxy();//"gf1.ucs.indiana.edu", 7512, "gerald", "....", 60*60*24*10);
-
+ String tempEOFFilename = "/tmp/status_end";
+ File tempEOFFile = new File(tempEOFFilename);
+ tempEOFFile.delete();
+
//store the input(it is a workflow description actually) in a file
- String filename = "/tmp/workflow.xml";
+ String filename = "/tmp/workflow_ws.xml";
File recordfile = new File(filename);
- BufferedWriter br = new BufferedWriter(new FileWriter(recordfile));
- br.write(workflow);
- br.close();
+ BufferedWriter bw = new BufferedWriter(new FileWriter(recordfile));
+ bw.write(workflow);
+ bw.flush();
+ //First, get the proxy certificate from the myproxy server
+ getCertFromMyproxy();//"gf1.ucs.indiana.edu", 7512, "gerald", "....", 60*60*24*10);
+
+ bw.write("get proxy\n");
+
//filename = "\"" + filename +"\"";
//String cmd = "E:\\my_program\\cog-4_1_5\\bin\\cog-workflow.bat";// + filename;
// get value of environment variable COG_INSTALL_PATH
@@ -86,16 +152,41 @@
System.out.println("Execute a new command");
wfproc = Runtime.getRuntime().exec(cmdarray);
+ //open status file
+ int count = 0;
+ File statusFile = new File(_statusFilename);
+ BufferedWriter statusbw = new BufferedWriter( new FileWriter(statusFile) );
+
//get output from the executed tool
line = "";
input = new BufferedReader (new InputStreamReader(wfproc.getInputStream()));
+ boolean isfirstline = true;
while ((line = input.readLine()) != null) {
- output += line;
+ //create a new status file
+// File statusFile = new File(statusFilename+count);
+// BufferedWriter statusbw = new BufferedWriter( new FileWriter(statusFile) );
+
+ if( isfirstline ){
+ isfirstline = false;
+ output += "Firstline:"+line;
+ statusbw.write(line,0,line.length());
+ statusbw.flush();
+ }else{
+ output += "\nFollowing line:"+line;
+ String newline = "\n"+line;
+ statusbw.write(newline,0,newline.length());
+ statusbw.flush();
+ }
+ count ++;
+// statusbw.close();
}
input.close();
+
+ tempEOFFile.createNewFile();
+ statusbw.close();
//exit = wfproc.exitValue();
//output = Integer.toString( exit );
- output = "succeeds!"+output;
+ output = "succeeds!\n"+output;
wfproc.destroy();
} catch (IllegalThreadStateException e) {
wfproc.destroy();
@@ -127,7 +218,7 @@
*/
private void getCertFromMyproxy(){//String host, int port, String user, String pwd, int lifetime){
try{
- String propertiesFile = "myproxy.properties";
+ String propertiesFile = "/home/tomcat/apache-tomcat-6.0.14/webapps/myproxy.properties";
String host = "";
int port = -1;
String user = "";
@@ -192,4 +283,191 @@
}
}
+ /**
+ * Auto generated method signature
+ * This method will be used by agent to submit a workflow.
+ * @param cmdWFSubmit0
+ */
+ public org.cogkit.cyberaide.axis2ws.CmdWFSubmitResponse cmdWFSubmit(
+ org.cogkit.cyberaide.axis2ws.CmdWFSubmit cmdWFSubmit0) {
+ String uid = cmdWFSubmit0.getParam0();
+ String wfid = cmdWFSubmit0.getParam1();
+ String workflow = cmdWFSubmit0.getParam2();
+
+ /*
+ //internally, invoke cmdCogExecution
+ org.cogkit.cyberaide.axis2ws.CmdCOGExecute execution = new org.cogkit.cyberaide.axis2ws.CmdCOGExecute();
+ execution.setParam0(_workflow);
+ org.cogkit.cyberaide.axis2ws.CmdCOGExecuteResponse resp = new org.cogkit.cyberaide.axis2ws.CmdCOGExecuteResponse();
+ resp = cmdCOGExecute(execution);
+ org.cogkit.cyberaide.axis2ws.CmdWFSubmitResponse wfresp = new org.cogkit.cyberaide.axis2ws.CmdWFSubmitResponse();
+ wfresp.set_return(resp.get_return());
+ */
+
+ Process wfproc = null;
+ String output = "";
+ int exit = -1;
+ try {
+ //First, get the proxy certificate from the myproxy server
+ getCertFromMyproxy();//"gf1.ucs.indiana.edu", 7512, "gerald", "....", 60*60*24*10);
+
+ //store the input(it is a workflow description actually) in a file
+ String filename = "/tmp/workflow_WFSubmit.xml";
+ File recordfile = new File(filename);
+ BufferedWriter br = new BufferedWriter(new FileWriter(recordfile));
+ br.write(workflow);
+ br.close();
+
+ //filename = "\"" + filename +"\"";
+ //String cmd = "E:\\my_program\\cog-4_1_5\\bin\\cog-workflow.bat";// + filename;
+ // get value of environment variable COG_INSTALL_PATH
+ String cog_workflow = "";
+ String getEnvVar = "printenv COG_INSTALL_PATH";
+ String line = "";
+
+ wfproc = Runtime.getRuntime().exec(getEnvVar);
+ BufferedReader input = new BufferedReader (new InputStreamReader(wfproc.getInputStream()));
+ while ((line = input.readLine()) != null) {
+ cog_workflow += line;
+ }
+ input.close();
+ wfproc.destroy();
+ if( cog_workflow.equals("") ){
+ throw new Exception("Environment variable COG_INSTALL_PATH is not set!!");
+ }
+
+ wfproc = null;
+ //call the cog command line tool to execute the workflow
+ String cmd = cog_workflow+"/bin/cog-workflow";
+ String []cmdarray = new String[2];
+ cmdarray[0] = cmd; //command
+ cmdarray[1] = filename;//argument
+ long timeout = 1000 * 30;//30 seconds
+ System.out.println("Execute a new command");
+ wfproc = Runtime.getRuntime().exec(cmdarray);
+
+ int statuscodeStarted = 1;
+ int statuscodeCompleted = 2;
+
+ // notify status service that the workflow has started running
+ _statusclient.notifyStatusOfWF(uid, wfid, statuscodeStarted, "workflow execution started");
+
+ //get output from the executed tool
+ line = "";
+ input = new BufferedReader (new InputStreamReader(wfproc.getInputStream()));
+ boolean isfirstline = true;
+ while ((line = input.readLine()) != null) {
+ if( isfirstline ){
+ isfirstline = false;
+ output += "WFSubmit_Firstline:"+line;
+ }else{
+ output += "\nWFSubmit_Following line:"+line;
+ }
+ /**
+ * Key point:
+ * Every time a new message is generated during execution of workflow, the
+ * message will be sent to status service based on web service.
+ * First, we need to parse the generated message because it contains some
+ * more information than pure message.
+ * Note: this must correspond to the method used to transform the received
+ * workflow by the agent server.
+ *
+ * Format:
+ * Currently, the generated status message has the format like this:
+ * /1/2/1/3|job:...|0
+ * Note: /1/2/1/3 denotes path of the target element in the workflow.
+ * The last number after '|' denotes status code. status code is
+ * 0 in the example above. The meaning of status code MUST match the
+ * definition in StatusService program!!
+ * Our job is to extract this part.
+ */
+ int index = line.indexOf('|');
+ if( index == -1 ){
+ //this message is not status message, it is generated natively by
+ //original workflow.
+ continue;
+ }else{
+ String path = line.substring(0, index);
+ String msg = line.substring(index+1);
+ int endindex = msg.lastIndexOf('|');
+ String statuscodestr = msg.substring(endindex+1);
+ msg = msg.substring(index+1, endindex);
+ int statuscode = Integer.parseInt(statuscodestr);
+ _statusclient.notifyStatusOfElement(uid, wfid, path, statuscode, msg);
+ }
+ }
+ input.close();
+
+ //notify status service that this workflow finishes executing
+ _statusclient.notifyStatusOfWF(uid, wfid, statuscodeCompleted, "workflow execution completed");
+
+ output = "succeeds!\n"+output;
+ wfproc.destroy();
+ } catch (IllegalThreadStateException e) {
+ wfproc.destroy();
+ output = e.toString();
+ } catch (Exception e){
+ wfproc.destroy();
+ output = e.toString();
+ }
+
+ org.cogkit.cyberaide.axis2ws.CmdWFSubmitResponse wfresp = new org.cogkit.cyberaide.axis2ws.CmdWFSubmitResponse();
+ wfresp.set_return( output );
+
+ return wfresp;
+ }
}
+
+
+class StatusServiceClient{
+// public static void main(String []args){
+ private StatusServiceInterfaceStub _stub = null;
+
+ public StatusServiceClient( String wsendpoint ){
+ // currently, the wsendpoint should be
+ // "http://156.56.104.196:8080/axis2/services/StatusServiceInterface";
+ try{
+ _stub = new StatusServiceInterfaceStub(wsendpoint);
+ }catch(Exception e){
+ System.out.println("in constructor of class StatusServiceClient in file SimpleHandlerInterfaceSkeleton.java"+e);
+ }
+ }
+
+ public void notifyStatusOfWF( String userid, String wfid, int statuscode, String statusmsg ){
+ try{
+ if( _stub == null ){
+ System.out.println("In method StatusServiceClient#StatusOfElement, _stub has not been set");
+ return ;
+ }
+ ModifyStatusOfWholeWF req = new ModifyStatusOfWholeWF();
+ req.setParam0(userid);
+ req.setParam1(wfid);
+ req.setParam2(statuscode);
+ req.setParam3(statusmsg);
+ _stub.modifyStatusOfWholeWF(req);
+ }catch(Exception e){
+ System.out.println(e);
+ }
+ }
+
+ public void notifyStatusOfElement(String userid, String wfid, String path, int statuscode, String statusmsg){
+ try{
+ if( _stub == null ){
+ System.out.println("In method StatusServiceClient#StatusOfElement, _stub has not been set");
+ return ;
+ }
+
+ //modify status of an element
+ ModifyStatusByPathString modifyReq = new ModifyStatusByPathString();
+ modifyReq.setParam0(userid);
+ modifyReq.setParam1(wfid);
+ modifyReq.setParam2(path);
+ modifyReq.setParam3(statuscode);
+ modifyReq.setParam4(statusmsg);
+ _stub.modifyStatusByPathString(modifyReq);
+
+ }catch(Exception e){
+ System.out.println(e);
+ }
+ }
+}
Added: trunk/cyberaide/src/axis2/axis2ws_server/StatusServiceInterface.java
===================================================================
--- trunk/cyberaide/src/axis2/axis2ws_server/StatusServiceInterface.java (rev 0)
+++ trunk/cyberaide/src/axis2/axis2ws_server/StatusServiceInterface.java 2007-12-07 05:39:56 UTC (rev 1852)
@@ -0,0 +1,66 @@
+package org.cogkit.cyberaide.axis2ws;
+
+/**
+ * This interface provides functionality needed by external entities.
+ */
+interface StatusServiceInterface {
+ /**
+ * This method is used to add a new workflow from a string passed as an
+ * argument.
+ *
+ * @param struid
+ * @param wfid
+ * @param workflow
+ */
+ public void addWorkflowFromString(String struid, String wfid, String workflow);
+ /**
+ * This method is used to add a new workflow from a workflow file.
+ *
+ * @param struid
+ * @param wfid
+ * @param filename
+ */
+ public void addWorkflowFromFile(String struid, String wfid, String filename);
+ /**
+ * get status of all workflows submitted by a user.
+ *
+ * @param struid
+ * @return a string in JSON format
+ */
+ public String getJSONStatusByUID(String struid);
+ public String getXMLStatusByUID(String struid);
+
+ /**
+ * @deprecated
+ * modify status of an element in the workflow.
+ *
+ * @param struid
+ * identify a user
+ * @param wfid
+ * identify a workflow
+ * @param element
+ * identify an element
+ * @param statusmsg
+ * status message
+ */
+ // public void modifyStatus(String struid, String wfid, String element, int statuscode, String statusmsg);
+ /**
+ * modify status of an element. The element of which status to be modified is located by parameter
+ * <i>path</i>. The from of path is not defined here. It depends on implementation.
+ * @param struid identify a user
+ * @param wfid identify a workflow
+ * @param path identify an element
+ * @param statuscode status code
+ * @param statusmsg status message
+ */
+ public void modifyStatusByPathString(String struid, String wfid, String path, int statuscode, String statusmsg);
+
+ /**
+ * modify status of a whole workflow.
+ * @param struid
+ * @param wfid
+ * @param statuscode
+ * @param statusmsg
+ */
+ public void modifyStatusOfWholeWF(String struid, String wfid, int statuscode, String statusmsg);
+}
Added: trunk/cyberaide/src/axis2/axis2ws_server/StatusServiceInterfaceSkeleton.java
===================================================================
--- trunk/cyberaide/src/axis2/axis2ws_server/StatusServiceInterfaceSkeleton.java (rev 0)
+++ trunk/cyberaide/src/axis2/axis2ws_server/StatusServiceInterfaceSkeleton.java 2007-12-07 05:39:56 UTC (rev 1852)
@@ -0,0 +1,942 @@
+/**
+ * StatusServiceInterfaceSkeleton.java
+ *
+ * This file was auto-generated from WSDL
+ * by the Apache Axis2 version: 1.3 Built on : Aug 10, 2007 (04:45:47 LKT)
+ */
+
+package org.cogkit.cyberaide.axis2ws;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.io.*;
+
+import org.globus.cog.karajan.workflow.ElementTree;
+import org.globus.cog.karajan.workflow.nodes.FlowElement;
+import org.globus.cog.karajan.Loader;
+
+/**
+ * StatusServiceInterfaceSkeleton java skeleton for the axisService
+ */
+public class StatusServiceInterfaceSkeleton
+ implements StatusServiceInterfaceSkeletonInterface {
+
+ private static String TAGUID = "uid";
+ private static String TAGWORKFLOWS = "workflows";
+
+ /**
+ * following three variables represent three ways to implement the
+ * status service.
+ * (1)event driven
+ * this means we base on the event model of cogkit.
+ * This does not work well because events generated by CoGKit are out of
+ * order. This may result from underlying Globus event model.
+ * (2)echo driven
+ * This is the default method.
+ * echo elements are added to original workflow automatically to
+ * record progress.
+ * (3)workflow element overwriting
+ * overwrite original workflow element, such as task:execution, task:transfer
+ * so that additional log/debug information is generated besides execution of
+ * tasks.
+ */
+ private static int EVENTDRIVEN = 1;
+ private static int ECHODRIVEN = 2;
+ private static int WFELEOVERWRITE = 3;
+
+ private int serviceImplType = this.ECHODRIVEN;
+
+ private static StatusDB _db = new StatusDB();
+
+ private String _logfile = "/tmp/status_service.log";
+ private RandomAccessFile _log = null;
+
+ private void writeLog(String log){
+ try{
+ if( _log == null ){
+ _log = new RandomAccessFile(_logfile, "rws");
+ _log.seek( _log.length() );
+ }
+ _log.writeChars(log);
+ }catch(Exception e){
+ System.out.println(e);
+ }
+ }
+ /**
+ * Auto generated method signature
+ * @param modifyStatus0
+ */
+ /*
+ public void modifyStatus(
+ org.cogkit.cyberaide.axis2ws.ModifyStatus modifyStatus0) {
+ String struid = modifyStatus0.getParam0();
+ String wfid = modifyStatus0.getParam1();
+ String element = modifyStatus0.getParam2();
+ int statuscode = modifyStatus0.getParam3();
+ String statusmsg = modifyStatus0.getParam4();
+ UID uid = new UIDUsername(struid);
+ _db.setElementStatus(uid, wfid, element, statuscode, statusmsg);
+ }
+ */
+
+ public void modifyStatusByPathString(
+ org.cogkit.cyberaide.axis2ws.ModifyStatusByPathString modifyStatusByPathString0) {
+ String struid = modifyStatusByPathString0.getParam0();
+ String wfid = modifyStatusByPathString0.getParam1();
+ String path = modifyStatusByPathString0.getParam2();
+ int statuscode = modifyStatusByPathString0.getParam3();
+ String statusmsg= modifyStatusByPathString0.getParam4();
+
+ writeLog("modifyStatusByPathString "+struid+":"+wfid+":"+path+":"+statuscode+":"+statusmsg+"\n");
+
+ UID uid = new UIDUsername(struid);
+ _db.setElementStatusByPathString(uid, wfid, path, statuscode, statusmsg);
+
+ String json = _db.getStatusByUID(uid).toJSON();
+ writeLog(json+"\n");
+ }
+ /**
+ * Auto generated method signature
+ * @param getXMLStatusByUID1
+ */
+ public org.cogkit.cyberaide.axis2ws.GetXMLStatusByUIDResponse getXMLStatusByUID(
+ org.cogkit.cyberaide.axis2ws.GetXMLStatusByUID getXMLStatusByUID1) {
+ String struid = getXMLStatusByUID1.getParam0();
+ UID uid = new UIDUsername(struid);
+ String result = _db.getStatusByUID(uid).toXML();
+ org.cogkit.cyberaide.axis2ws.GetXMLStatusByUIDResponse resp = new org.cogkit.cyberaide.axis2ws.GetXMLStatusByUIDResponse();
+ resp.set_return(result);
+ return resp;
+ }
+
+ /**
+ * Auto generated method signature
+ * @param getJSONStatusByUID3
+ */
+ public org.cogkit.cyberaide.axis2ws.GetJSONStatusByUIDResponse getJSONStatusByUID(
+ org.cogkit.cyberaide.axis2ws.GetJSONStatusByUID getJSONStatusByUID3) {
+ try{
+ String struid = getJSONStatusByUID3.getParam0();
+
+ File file = new File("/tmp/json_request.txt");
+ file.createNewFile();
+ BufferedWriter bw = new BufferedWriter( new FileWriter(file) );
+ bw.write("user:"+struid);
+ bw.close();
+
+ UID uid = new UIDUsername(struid);
+ StatusList sl = _db.getStatusByUID(uid);
+
+ String result = "{";
+ if( sl == null){
+ result = "No workflow found";
+ }else{
+ result = "{";
+ result += "\""+TAGUID+"\":\""+struid+"\"";
+ result += ",";
+ result += "\""+TAGWORKFLOWS+"\":";
+ result += sl.toJSON();
+ result += "}";
+ }
+ org.cogkit.cyberaide.axis2ws.GetJSONStatusByUIDResponse resp = new org.cogkit.cyberaide.axis2ws.GetJSONStatusByUIDResponse();
+ resp.set_return(result);
+ return resp;
+ }catch(Exception e){
+ return null;
+ }
+ }
+
+ /**
+ * Auto generated method signature
+ * @param addWorkflowFromString5
+ */
+ public void addWorkflowFromString(
+ org.cogkit.cyberaide.axis2ws.AddWorkflowFromString addWorkflowFromString5) {
+ writeLog("addWorkflowFromString:");
+ String struid = addWorkflowFromString5.getParam0();
+ String wfid = addWorkflowFromString5.getParam1();
+ String workflow = addWorkflowFromString5.getParam2();
+
+ writeLog("addWorkflowFromString:"+struid+":"+wfid+":"+workflow+"\n");
+
+ UID uid = new UIDUsername(struid);
+ KarajanTree tree = new KarajanTree();
+ tree.LoadFromString(workflow);
+ WorkflowStatus wfs = new WorkflowStatus(tree);
+ wfs.setWfid(wfid);
+ _db.addStatus(uid, wfs);
+ }
+
+ /**
+ * Auto generated method signature
+ * @param addWorkflowFromFile6
+ */
+ public void addWorkflowFromFile(
+ org.cogkit.cyberaide.axis2ws.AddWorkflowFromFile addWorkflowFromFile6) {
+ String struid = addWorkflowFromFile6.getParam0();
+ String wfid = addWorkflowFromFile6.getParam1();
+ String filename = addWorkflowFromFile6.getParam2();
+
+ UIDUsername uid = new UIDUsername(struid);
+ KarajanTree tree = new KarajanTree();
+ tree.LoadFromFile(filename);
+ WorkflowStatus wfs = new WorkflowStatus(tree);
+ wfs.setWfid(wfid);
+ _db.addStatus(uid, wfs);
+ }
+
+ public void modifyStatusOfWholeWF(org.cogkit.cyberaide.axis2ws.ModifyStatusOfWholeWF modifyStatusOfWholeWF7){
+ String struid = modifyStatusOfWholeWF7.getParam0();
+ String wfid = modifyStatusOfWholeWF7.getParam1();
+ int statuscode = modifyStatusOfWholeWF7.getParam2();
+ String statusmsg = modifyStatusOfWholeWF7.getParam3();
+
+ writeLog("modifyStatusOfWholeWF:"+struid+":"+wfid+":"+statuscode+":"+statusmsg+"\n");
+
+ UID uid = new UIDUsername(struid);
+ _db.setStatusOfWF(uid, wfid, statuscode, statusmsg);
+
+ String json = _db.getStatusByUID(uid).toJSON();
+ writeLog(json+"\n");
+ }
+}
+
+/**
+ * Interface which all tree nodes must implement. Note: every node can contain
+ * its data which can be gotten by invoking method <b>getData()</b>.
+ */
+interface ITreeNode {
+ // TreeNode getParent(); //I am not sure currently whether parent pointer
+ // needs to be maintained.
+ // ITreeNode getFirstChild();
+ ITreeNode[] getChildren();
+
+ void addChild(ITreeNode child);
+
+ /**
+ * remove all children. Note: actually this will remove all direct and
+ * indirect children. In other words, a subtree of which root is current
+ * node will be removed.
+ */
+ void removeAllChildren();
+
+ ITreeNodeData getData();
+ void setData(ITreeNodeData data);
+ ITreeNode searchNode(ITreeNodeData criteria); // return one node
+ ITreeNode getNodeByPath(String path,String delimiter);
+ ITreeNode getNodeByPath(int []path);
+ ITreeNode getChildByIndex(int index);
+}
+
+/**
+ * In my case, the tree does not contain duplicate nodes, which means only a
+ * result node can be returned by function <b>searchXXX()</b>.
+ *
+ * @author gerald
+ *
+ */
+interface ITree {
+ ITreeNode getRoot();
+
+ /**
+ * Search. Actually criteria will be compared with data contained in every
+ * node.
+ */
+ ITreeNode searchNode(ITreeNodeData criteria);
+ ITreeNode getNodeByPath(String path, String delimiter);
+}
+
+/**
+ * This interface represents the data contained in tree node.
+ *
+ * @author gerald
+ *
+ */
+interface ITreeNodeData {
+ boolean isEqual(ITreeNodeData data);
+}
+
+class TreeNode implements ITreeNode {
+ private ArrayList<ITreeNode> _children = new ArrayList<ITreeNode>();
+ private ITreeNodeData _data = null;
+
+ public ITreeNodeData getData() {
+ return _data;
+ }
+
+ public ITreeNode[] getChildren() {
+ ITreeNode[] children =(ITreeNode[]) _children.toArray(new ITreeNode[0]);
+ return children;
+ }
+
+ public void addChild(ITreeNode child) {
+ _children.add(child);
+ }
+
+ public void removeAllChildren() {
+ Iterator it = _children.iterator();
+ while (it.hasNext()) {
+ ITreeNode node = (ITreeNode) it.next();
+ node.removeAllChildren();
+ }
+ _children.clear();
+ }
+
+ public void setData(ITreeNodeData data) {
+ _data = data;
+ }
+ public ITreeNode getChildByIndex(int index){
+ if( index < _children.size() )
+ return _children.get(index);
+ return null;
+ }
+ public ITreeNode searchNode(ITreeNodeData criteria){
+ return internalSearch(this, criteria);
+ }
+ private ITreeNode internalSearch(ITreeNode root, ITreeNodeData criteria) {
+ ITreeNodeData data = root.getData();
+ if (data.isEqual(criteria)){
+// System.out.println("to find:"+((KarajanTreeNodeData)criteria).getNodeType());
+// System.out.println("find:"+((KarajanTreeNodeData)data).getNodeType()+","+((KarajanTreeNodeData)data).getStatus());
+ return root;
+ }
+ else {
+ ITreeNode node;
+ ITreeNode[] children = root.getChildren();
+ for (int i = 0; i < children.length; i++) {
+ node = children[i];
+ node = internalSearch(node, criteria);
+ if ( node != null)
+ return node;
+ }
+ return null;
+ }
+ }
+ /**
+ * Path format is like this:
+ * /1/2/3/4
+ */
+ public ITreeNode getNodeByPath(String path, String delimiter){
+ String pathstr = path;
+ if(pathstr.indexOf(delimiter)==0){
+ pathstr = pathstr.substring(delimiter.length());
+ }
+ String pathSeg[] = pathstr.split(delimiter);
+ int indexArr[] = new int[pathSeg.length];
+ for(int i = 0 ; i < pathSeg.length; i++){
+ int childindex = Integer.valueOf(pathSeg[i]);
+ indexArr[i] = childindex;
+ }
+ //find node along the path
+ return getNodeByPath(indexArr);
+ }
+
+ public ITreeNode getNodeByPath(int []path){
+ ITreeNode currentNode = this;
+ for( int i = 0 ; i < path.length; i++){
+ currentNode = currentNode.getChildByIndex(path[i]-1);
+ }
+ return currentNode;
+ }
+}
+
+/**
+ * This class is closely related to flow element in Karajan.
+ *
+ * @author gerald
+ *
+ */
+class KarajanTreeNodeData implements ITreeNodeData {
+ private String _nodeType = "";
+ private String _status = "";
+ private int _statuscode = StatusConstant.IDLE;
+ // Currently, just compare the node type and statuscode.
+ // Maybe in the future, some more
+ // comparisons need to be added.
+ public boolean isEqual(ITreeNodeData data) {
+ if(_nodeType.compareTo(((KarajanTreeNodeData)data).getNodeType())==0 &&
+ _statuscode == ((KarajanTreeNodeData)data).getStatuscode() ){
+// System.out.println("I find it:"+_nodeType);
+ return true;
+ }
+ return false;
+ }
+ public KarajanTreeNodeData(String nodeType, String status){
+ _nodeType = nodeType;
+ _status = status;
+ }
+ public KarajanTreeNodeData(String nodeType){
+ _nodeType = nodeType;
+ // TODO set initial status here
+ _status = "idle";
+ }
+ public String getNodeType(){
+ return _nodeType;
+ }
+ public String getStatus(){
+ return _status;
+ }
+ public void setStatus(String status) {
+ _status = status;
+ }
+ public int getStatuscode() {
+ return _statuscode;
+ }
+ public void setStatuscode(int _statuscode) {
+ this._statuscode = _statuscode;
+ }
+}
+
+
+class Tree implements ITree {
+ private ITreeNode _root = null;
+
+ public ITreeNode getRoot() {
+ return _root;
+ }
+
+ public void setRoot(ITreeNode root) {
+ if (_root != null) {
+ System.out.println("Root of the tree has been set!!");
+ return;
+ }
+ _root = root;
+ }
+
+ public ITreeNode searchNode(ITreeNodeData criteria) {
+ if (_root == null) {
+ System.out.println("You have not built your tree yet!!");
+ return null;
+ }
+ return _root.searchNode(criteria);
+ }
+ public ITreeNode getNodeByPath(String path, String delimiter){
+ return _root.getNodeByPath(path, delimiter);
+ }
+}
+
+/**
+ * This interface provides necessary method which can be used to represent a
+ * unique user. Note: we don't specify the internal data structure. All we need
+ * is that the returned value must be a <b>String</b>.
+ */
+interface UID {
+ String getUID();
+ boolean isEqual(UID uid2);
+}
+
+interface IStatus {
+ boolean isEqual(IStatus obj);
+}
+
+/**
+ * This class uses user name as uid. My assumption is that user name is unique
+ * for every user.
+ */
+class UIDUsername implements UID {
+ private String _uid = "";
+
+ public UIDUsername(String uid) {
+ _uid = uid;
+ }
+
+ public String getUID() {
+ return _uid;
+ }
+
+ public boolean isEqual(UID uid) {
+ return _uid.compareTo(uid.getUID()) == 0;
+ }
+ public String toString(){
+ return _uid;
+ }
+}
+/**
+ * This class represents status of execution of a workflow. Two kinds of
+ * information are provided here: (1) status about whole workflow (2) status
+ * about a certain step in the workflow. <b>Layout:</b>
+ * ___________________________________________ | message | | code | | | - step 1
+ * (information) | | flowtree -- | - step 2 (information) | | | - step 3
+ * (information) | |___________________________________________|
+ */
+class WorkflowStatus {
+ private static String WFID = "wfid";
+ private static String STATUS = "status";
+ private static String ELEMENTS = "elements";
+ /**
+ * following two variables represent the status of the whole workflow. If
+ * you need information about a certain sub task in the workflow, you should
+ * consult <b><i>_flowtree</i></b> variable.
+ */
+ private String _statusMsg = "";
+ private int _statusCode;
+ /**
+ * This variable represents the logic structure of the karajan workflow.
+ */
+ private KarajanTree _flowtree = null;
+
+ /**
+ * This variable uniquely identifies a workflow so that we can distinguish
+ * different workflows submitted by a single user.
+ */
+ private String _wfid = "";
+
+ //private static int COMPLETED = 1;
+ //private static int FAILED = 2;
+ private static String[] MSG= {
+ "idle", // corresponding to idle
+ "started", // corresponding to started
+ "completed" // corresponding to completed
+ };
+ public WorkflowStatus(KarajanTree tree){
+ _flowtree = tree;
+ _statusCode = StatusConstant.IDLE;
+ _statusMsg = this.MSG[_statusCode];
+ }
+
+ public String getStatusMsg() {
+ return _statusMsg;
+ }
+
+ public void setStatusMsg(String msg) {
+ _statusMsg = msg;
+ }
+ public int get_statusCode() {
+ return _statusCode;
+ }
+ public void set_statusCode(int code) {
+ _statusCode = code;
+ }
+ /*
+ public void onCompleted(){
+ setStatusMsg(MSG[COMPLETED]);
+ }
+ public void onFailed(){
+ setStatusMsg(MSG[FAILED]);
+ }*/
+ /**
+ * This class can modify status of a certain element in the workflow.
+ *
+ * @param element
+ * @param statusmsg
+ */
+ /*
+ public void setElementStatus(String element, int statuscode, String statusmsg){
+ try {
+ _flowtree.modifyStatus(element, statuscode, statusmsg);
+ } catch (Exception e) {
+ System.out.println(e);
+ }
+ }*/
+
+ public void setElementStatusByPathString(String path, int statuscode, String statusmsg){
+ try {
+ _flowtree.modifyStatusByPathString(path, statuscode, statusmsg);
+ } catch (Exception e) {
+ System.out.println(e);
+ }
+ }
+
+ public String getWfid() {
+ return _wfid;
+ }
+ public void setWfid(String wfid) {
+ _wfid = wfid;
+ }
+ public String toJSON(){
+ String result = "{";
+ result += "\""+WFID+"\":\""+_wfid+"\"";
+ result += ",";
+ result += "\""+STATUS+"\":\""+_statusMsg+"\"";
+ result += ",";
+ result += "\""+this.ELEMENTS+"\":[";
+ result += _flowtree.toJSON();
+ result += "]}";
+ return result;
+ }
+}
+
+class StatusData implements ITreeNodeData{
+
+ public boolean isEqual(ITreeNodeData data) {
+ return true;
+ }
+}
+
+/**
+ * This class represents an entry in the status database. Layout:
+ * _________________________________ | | | | user id | WorkflowStatus |
+ * |___________|_____________________|
+ *
+ */
+/*
+ * class StatusEntry { private UID _uid; private WorkflowStatus _status;
+ *
+ * public UID getUID() { return _uid; }
+ *
+ * public WorkflowStatus getStatus() { return _status; }
+ *
+ * public boolean isEqual(ITreeNodeData data) { StatusEntry entry =
+ * (StatusEntry) data; return _uid.isEqual(entry.getUID()); } }
+ */
+
+class KarajanTree extends Tree {
+ private boolean _isload = false;
+ /** contains list of nodes which are executed */
+// private ArrayList<ITreeNode> _currentNodes = new ArrayList<ITreeNode>();
+ public KarajanTree(){
+ }
+ public void LoadFromFile(String filename) {
+ if(_isload)
+ return;
+ try {
+ File karajanfile = new File(filename);
+ if (karajanfile.exists() == false) {
+ System.out.println("The karajan workflow file " + filename
+ + " does not exist!!");
+ return;
+ }
+ ElementTree tree = Loader.load(filename);
+ buildTree(tree);
+ _isload = true;
+ } catch (Exception e) {
+ System.out.println(e);
+ }
+ }
+ public void LoadFromString(String strworkflow) {
+ if(_isload)
+ return;
+ try {
+ ElementTree tree = Loader.loadFromString(strworkflow);
+ buildTree(tree);
+ _isload = true;
+ } catch (Exception e) {
+ System.out.println(e);
+ }
+ }
+ protected void buildTree(ElementTree eletree){
+ setRoot( addSubTree(eletree.getRoot()));
+// _currentNodes.add(getRoot());
+ }
+ private ITreeNode addSubTree(FlowElement fe){
+ ITreeNode root = new TreeNode();
+ KarajanTreeNodeData nodeData = new KarajanTreeNodeData(getPureName(fe.getElementType()));
+ root.setData(nodeData);
+
+ List clist = fe.elements();
+ Iterator it = clist.iterator();
+ while( it.hasNext() ){
+ FlowElement flowele = (FlowElement)it.next();
+ ITreeNode child = addSubTree(flowele);
+ root.addChild( child );
+ }
+ return root;
+ }
+ private String getPureName(String fullname){
+ int npos = fullname.indexOf(':');
+ if( npos != -1 ){
+ return fullname.substring(npos+1);
+ }
+ return fullname;
+ }
+ public String toJSON(){
+ return toJSON(getRoot());
+ }
+ private String toJSON(ITreeNode node){
+ if( node == null )
+ return "";
+ KarajanTreeNodeData data = (KarajanTreeNodeData)node.getData();
+ ITreeNode []children = node.getChildren();
+ String result = "{";
+ result += "\""+KarajanNodeJSON.NODENAME+"\":\""+data.getNodeType()+"\"";
+ result += ",";
+ result += "\""+KarajanNodeJSON.STATUS+"\":\""+data.getStatus()+"\"";
+ result += ",";
+ result += "\""+KarajanNodeJSON.STATUSCODE+"\":"+data.getStatuscode()+"";
+ result += ",";
+ result += "\""+KarajanNodeJSON.CHILDREN+"\":[";
+ for( int i = 0 ; i < children.length ; i ++ ){
+ result += toJSON(children[i]);
+ if( i < children.length-1 )
+ result +=",";
+ }
+ result += "]}";
+ return result;
+ }
+ /**
+ * This function is used to modify status of an element.
+ * The destination node is found by following the path specified by parameter
+ * <i>path</i>
+ * Note: StatusContant.PATHDELIMITER specifies the delimiter between path parts.
+ * Currently, the delimiter is "/". This must accord with the format of event
+ * generated by backend grid service because the parameter <i>path</i> actually is
+ * passed from that part(backend grid service).
+ * @param path specifies path of the node of which status will be modified
+ * @param statuscode status code
+ * @param statusmsg status message
+ */
+ public synchronized void modifyStatusByPathString(String path, int statuscode, String statusmsg){
+ ITreeNode node = getRoot().getNodeByPath(path, StatusConstant.PATHDELIMITER);
+ if(node!=null){
+ KarajanTreeNodeData data = (KarajanTreeNodeData)node.getData();
+ data.setStatuscode(statuscode);
+ data.setStatus(statusmsg);
+ }
+ }
+
+ public synchronized void modifyStatusByPathIndex(int []path, int statuscode, String statusmsg){
+ ITreeNode node = getRoot().getNodeByPath(path);
+ if(node!=null){
+ KarajanTreeNodeData data = (KarajanTreeNodeData)node.getData();
+ data.setStatuscode(statuscode);
+ data.setStatus(statusmsg);
+ }
+ }
+ /**
+ * This function must be synchronized because concurrent access is possible.
+ *
+ * @param strele
+ * name of the element, e.g. transfer, execution, echo ...
+ * @param statusmsg
+ * description about the status corresponding to the element.
+ */
+ /*
+ public synchronized void modifyStatus(String eleType, int statuscode, String statusmsg) throws Exception{
+ eleType = getPureName(eleType);
+ switch (statuscode){
+ case StatusConstant.STARTED:
+/* for(int i = 0 ; i < _currentNodes.size(); i++){
+ ITreeNode node = _currentNodes.get(i);
+ KarajanTreeNodeData data = (KarajanTreeNodeData)node.getData();
+ if( data.getNodeType().compareTo(eleType) == 0 ){
+ data.setStatus(statusmsg);
+ data.setStatuscode(statuscode);
+ return;
+ }
+ }* /
+// System.out.println(eleType);
+ ITreeNode newnode = null;
+ KarajanTreeNodeData criteria = new KarajanTreeNodeData(eleType);
+ criteria.setStatuscode(StatusConstant.IDLE);
+
+ for(int i = 0 ; i < _currentNodes.size(); i++){
+ newnode = _currentNodes.get(i).searchNode(criteria);
+ if( newnode != null ){
+// System.out.println("---find it!!!!!!!!!!!!!"+((KarajanTreeNodeData)newnode.getData()).getNodeType());
+ ((KarajanTreeNodeData)newnode.getData()).setStatus(statusmsg);
+ ((KarajanTreeNodeData)newnode.getData()).setStatuscode(statuscode);
+ _currentNodes.add(newnode);
+ break;
+ }
+ }
+ break;
+
+ case StatusConstant.COMPLETED:
+ case StatusConstant.FAILED:
+ for(int i = 0 ; i < _currentNodes.size(); i++){
+ ITreeNode node = _currentNodes.get(i);
+ KarajanTreeNodeData data = (KarajanTreeNodeData)node.getData();
+ if( data.getNodeType().compareTo(eleType) == 0 ){
+ data.setStatus(statusmsg);
+ data.setStatuscode(statuscode);
+ _currentNodes.remove(...
[truncated message content] |
