You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by sa...@apache.org on 2014/07/19 16:08:30 UTC
[2/3] moving about status listener classes out of hte gfac
http://git-wip-us.apache.org/repos/asf/airavata/blob/92838e4a/modules/gfac/gfac-local/src/test/java/org/apache/airavata/core/gfac/services/impl/LocalProviderTest.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-local/src/test/java/org/apache/airavata/core/gfac/services/impl/LocalProviderTest.java b/modules/gfac/gfac-local/src/test/java/org/apache/airavata/core/gfac/services/impl/LocalProviderTest.java
index 818e4be..9ba7e5d 100644
--- a/modules/gfac/gfac-local/src/test/java/org/apache/airavata/core/gfac/services/impl/LocalProviderTest.java
+++ b/modules/gfac/gfac-local/src/test/java/org/apache/airavata/core/gfac/services/impl/LocalProviderTest.java
@@ -20,15 +20,21 @@
*/
package org.apache.airavata.core.gfac.services.impl;
-import com.google.common.eventbus.EventBus;
+import java.io.File;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.List;
-import org.apache.airavata.commons.gfac.type.*;
+import org.apache.airavata.common.utils.MonitorPublisher;
+import org.apache.airavata.commons.gfac.type.ActualParameter;
+import org.apache.airavata.commons.gfac.type.ApplicationDescription;
+import org.apache.airavata.commons.gfac.type.HostDescription;
+import org.apache.airavata.commons.gfac.type.ServiceDescription;
import org.apache.airavata.gfac.GFacConfiguration;
import org.apache.airavata.gfac.GFacException;
import org.apache.airavata.gfac.core.context.ApplicationContext;
import org.apache.airavata.gfac.core.context.JobExecutionContext;
import org.apache.airavata.gfac.core.context.MessageContext;
-import org.apache.airavata.gfac.core.notification.MonitorPublisher;
import org.apache.airavata.gfac.core.provider.GFacProviderException;
import org.apache.airavata.gfac.local.handler.LocalDirectorySetupHandler;
import org.apache.airavata.gfac.local.provider.impl.LocalProvider;
@@ -45,10 +51,7 @@ import org.apache.commons.lang.SystemUtils;
import org.testng.annotations.BeforeTest;
import org.testng.annotations.Test;
-import java.io.File;
-import java.net.URL;
-import java.util.ArrayList;
-import java.util.List;
+import com.google.common.eventbus.EventBus;
public class LocalProviderTest {
private JobExecutionContext jobExecutionContext;
http://git-wip-us.apache.org/repos/asf/airavata/blob/92838e4a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/core/AiravataAbstractMonitor.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/core/AiravataAbstractMonitor.java b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/core/AiravataAbstractMonitor.java
index 26f25b8..c754d3c 100644
--- a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/core/AiravataAbstractMonitor.java
+++ b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/core/AiravataAbstractMonitor.java
@@ -20,7 +20,7 @@
*/
package org.apache.airavata.gfac.monitor.core;
-import org.apache.airavata.gfac.core.notification.MonitorPublisher;
+import org.apache.airavata.common.utils.MonitorPublisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
http://git-wip-us.apache.org/repos/asf/airavata/blob/92838e4a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/handlers/GridPushMonitorHandler.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/handlers/GridPushMonitorHandler.java b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/handlers/GridPushMonitorHandler.java
index 66f8467..c69b516 100644
--- a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/handlers/GridPushMonitorHandler.java
+++ b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/handlers/GridPushMonitorHandler.java
@@ -20,23 +20,20 @@
*/
package org.apache.airavata.gfac.monitor.handlers;
-import com.google.common.eventbus.EventBus;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.LinkedBlockingQueue;
+
import org.apache.airavata.common.exception.ApplicationSettingsException;
import org.apache.airavata.common.utils.ServerSettings;
import org.apache.airavata.gfac.core.context.JobExecutionContext;
import org.apache.airavata.gfac.core.cpi.GFacImpl;
import org.apache.airavata.gfac.core.handler.GFacHandlerException;
import org.apache.airavata.gfac.core.handler.ThreadedHandler;
-
-import java.util.*;
-import java.util.concurrent.LinkedBlockingQueue;
-
import org.apache.airavata.gfac.core.monitor.MonitorID;
-import org.apache.airavata.gfac.core.notification.MonitorPublisher;
import org.apache.airavata.gfac.monitor.HPCMonitorID;
-import org.apache.airavata.gfac.monitor.exception.AiravataMonitorException;
import org.apache.airavata.gfac.monitor.impl.push.amqp.AMQPMonitor;
-import org.apache.airavata.gfac.monitor.util.CommonUtils;
import org.apache.airavata.gsi.ssh.api.authentication.AuthenticationInfo;
import org.apache.airavata.gsi.ssh.impl.authentication.MyProxyAuthenticationInfo;
import org.slf4j.Logger;
http://git-wip-us.apache.org/repos/asf/airavata/blob/92838e4a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/pull/qstat/HPCPullMonitor.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/pull/qstat/HPCPullMonitor.java b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/pull/qstat/HPCPullMonitor.java
index fa6714d..4208c85 100644
--- a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/pull/qstat/HPCPullMonitor.java
+++ b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/pull/qstat/HPCPullMonitor.java
@@ -20,20 +20,25 @@
*/
package org.apache.airavata.gfac.monitor.impl.pull.qstat;
-import com.google.common.eventbus.EventBus;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
+import org.apache.airavata.common.utils.MonitorPublisher;
import org.apache.airavata.common.utils.ServerSettings;
import org.apache.airavata.commons.gfac.type.HostDescription;
import org.apache.airavata.gfac.GFacException;
import org.apache.airavata.gfac.core.cpi.GFac;
-import org.apache.airavata.gfac.core.monitor.ExperimentIdentity;
import org.apache.airavata.gfac.core.monitor.MonitorID;
import org.apache.airavata.gfac.core.monitor.TaskIdentity;
-import org.apache.airavata.gfac.core.monitor.state.ExperimentStatusChangedEvent;
import org.apache.airavata.gfac.core.monitor.state.JobStatusChangeRequest;
import org.apache.airavata.gfac.core.monitor.state.TaskStatusChangeRequest;
-import org.apache.airavata.gfac.core.monitor.state.TaskStatusChangedEvent;
-import org.apache.airavata.gfac.core.notification.MonitorPublisher;
import org.apache.airavata.gfac.monitor.HostMonitorData;
import org.apache.airavata.gfac.monitor.UserMonitorData;
import org.apache.airavata.gfac.monitor.core.PullMonitor;
@@ -41,19 +46,14 @@ import org.apache.airavata.gfac.monitor.exception.AiravataMonitorException;
import org.apache.airavata.gfac.monitor.util.CommonUtils;
import org.apache.airavata.gsi.ssh.api.SSHApiException;
import org.apache.airavata.gsi.ssh.api.authentication.AuthenticationInfo;
-import org.apache.airavata.model.workspace.experiment.ExperimentState;
import org.apache.airavata.model.workspace.experiment.JobState;
import org.apache.airavata.model.workspace.experiment.TaskState;
import org.apache.airavata.schemas.gfac.GsisshHostType;
import org.apache.airavata.schemas.gfac.SSHHostType;
-import org.apache.openjpa.lib.log.Log;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.sql.Timestamp;
-import java.util.*;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingDeque;
+import com.google.common.eventbus.EventBus;
/**
* This monitor is based on qstat command which can be run
http://git-wip-us.apache.org/repos/asf/airavata/blob/92838e4a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/push/amqp/AMQPMonitor.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/push/amqp/AMQPMonitor.java b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/push/amqp/AMQPMonitor.java
index fa17c4e..010d3bc 100644
--- a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/push/amqp/AMQPMonitor.java
+++ b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/push/amqp/AMQPMonitor.java
@@ -20,16 +20,20 @@
*/
package org.apache.airavata.gfac.monitor.impl.push.amqp;
-import com.google.common.eventbus.EventBus;
-import com.google.common.eventbus.Subscribe;
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+
+import org.apache.airavata.common.utils.MonitorPublisher;
import org.apache.airavata.common.utils.ServerSettings;
import org.apache.airavata.commons.gfac.type.HostDescription;
import org.apache.airavata.gfac.core.monitor.JobIdentity;
import org.apache.airavata.gfac.core.monitor.MonitorID;
import org.apache.airavata.gfac.core.monitor.state.JobStatusChangeRequest;
-import org.apache.airavata.gfac.core.notification.MonitorPublisher;
import org.apache.airavata.gfac.monitor.core.PushMonitor;
import org.apache.airavata.gfac.monitor.exception.AiravataMonitorException;
import org.apache.airavata.gfac.monitor.util.AMQPConnectionUtil;
@@ -38,9 +42,10 @@ import org.apache.airavata.model.workspace.experiment.JobState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
-import java.util.*;
-import java.util.concurrent.BlockingQueue;
+import com.google.common.eventbus.EventBus;
+import com.google.common.eventbus.Subscribe;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
/**
* This is the implementation for AMQP based finishQueue, this uses
http://git-wip-us.apache.org/repos/asf/airavata/blob/92838e4a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/push/amqp/BasicConsumer.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/push/amqp/BasicConsumer.java b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/push/amqp/BasicConsumer.java
index b8ec08a..b992f85 100644
--- a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/push/amqp/BasicConsumer.java
+++ b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/push/amqp/BasicConsumer.java
@@ -20,17 +20,18 @@
*/
package org.apache.airavata.gfac.monitor.impl.push.amqp;
-import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.Consumer;
-import com.rabbitmq.client.Envelope;
-import com.rabbitmq.client.ShutdownSignalException;
+import org.apache.airavata.common.utils.MonitorPublisher;
import org.apache.airavata.gfac.core.monitor.MonitorID;
-import org.apache.airavata.gfac.core.notification.MonitorPublisher;
import org.apache.airavata.gfac.monitor.core.MessageParser;
import org.apache.airavata.gfac.monitor.exception.AiravataMonitorException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.Envelope;
+import com.rabbitmq.client.ShutdownSignalException;
+
public class BasicConsumer implements Consumer {
private final static Logger logger = LoggerFactory.getLogger(AMQPMonitor.class);
http://git-wip-us.apache.org/repos/asf/airavata/blob/92838e4a/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/AMQPMonitorTest.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/AMQPMonitorTest.java b/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/AMQPMonitorTest.java
index 9c3d08b..b17f9b8 100644
--- a/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/AMQPMonitorTest.java
+++ b/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/AMQPMonitorTest.java
@@ -20,12 +20,18 @@
*/
package org.apache.airavata.job;
-import com.google.common.eventbus.EventBus;
-import com.google.common.eventbus.Subscribe;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.apache.airavata.common.utils.MonitorPublisher;
import org.apache.airavata.commons.gfac.type.HostDescription;
import org.apache.airavata.gfac.core.monitor.MonitorID;
import org.apache.airavata.gfac.core.monitor.state.JobStatusChangeRequest;
-import org.apache.airavata.gfac.core.notification.MonitorPublisher;
+import org.apache.airavata.gfac.monitor.impl.push.amqp.AMQPMonitor;
import org.apache.airavata.gsi.ssh.api.Cluster;
import org.apache.airavata.gsi.ssh.api.SSHApiException;
import org.apache.airavata.gsi.ssh.api.ServerInfo;
@@ -33,18 +39,13 @@ import org.apache.airavata.gsi.ssh.api.authentication.GSIAuthenticationInfo;
import org.apache.airavata.gsi.ssh.api.job.JobDescriptor;
import org.apache.airavata.gsi.ssh.impl.PBSCluster;
import org.apache.airavata.gsi.ssh.impl.authentication.MyProxyAuthenticationInfo;
-import org.apache.airavata.gfac.monitor.impl.push.amqp.AMQPMonitor;
import org.apache.airavata.schemas.gfac.GsisshHostType;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
-import java.io.File;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
+import com.google.common.eventbus.EventBus;
+import com.google.common.eventbus.Subscribe;
public class AMQPMonitorTest {
http://git-wip-us.apache.org/repos/asf/airavata/blob/92838e4a/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/QstatMonitorTestWithMyProxyAuth.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/QstatMonitorTestWithMyProxyAuth.java b/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/QstatMonitorTestWithMyProxyAuth.java
index 3a762ec..6dd4d68 100644
--- a/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/QstatMonitorTestWithMyProxyAuth.java
+++ b/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/QstatMonitorTestWithMyProxyAuth.java
@@ -20,13 +20,19 @@
*/
package org.apache.airavata.job;
-import com.google.common.eventbus.EventBus;
-import com.google.common.eventbus.Subscribe;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.apache.airavata.common.utils.MonitorPublisher;
import org.apache.airavata.commons.gfac.type.HostDescription;
import org.apache.airavata.gfac.core.monitor.MonitorID;
import org.apache.airavata.gfac.core.monitor.state.JobStatusChangeRequest;
-import org.apache.airavata.gfac.core.notification.MonitorPublisher;
import org.apache.airavata.gfac.monitor.HPCMonitorID;
+import org.apache.airavata.gfac.monitor.UserMonitorData;
+import org.apache.airavata.gfac.monitor.impl.pull.qstat.HPCPullMonitor;
import org.apache.airavata.gsi.ssh.api.Cluster;
import org.apache.airavata.gsi.ssh.api.SSHApiException;
import org.apache.airavata.gsi.ssh.api.ServerInfo;
@@ -35,17 +41,12 @@ import org.apache.airavata.gsi.ssh.api.job.JobDescriptor;
import org.apache.airavata.gsi.ssh.impl.PBSCluster;
import org.apache.airavata.gsi.ssh.impl.authentication.MyProxyAuthenticationInfo;
import org.apache.airavata.gsi.ssh.util.CommonUtils;
-import org.apache.airavata.gfac.monitor.UserMonitorData;
-import org.apache.airavata.gfac.monitor.impl.pull.qstat.HPCPullMonitor;
import org.apache.airavata.schemas.gfac.GsisshHostType;
import org.junit.Assert;
import org.testng.annotations.Test;
-import java.io.File;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
+import com.google.common.eventbus.EventBus;
+import com.google.common.eventbus.Subscribe;
public class QstatMonitorTestWithMyProxyAuth {
private String myProxyUserName;
http://git-wip-us.apache.org/repos/asf/airavata/blob/92838e4a/modules/workflow-model/workflow-engine/src/main/java/org/apache/airavata/workflow/engine/WorkflowUtils.java
----------------------------------------------------------------------
diff --git a/modules/workflow-model/workflow-engine/src/main/java/org/apache/airavata/workflow/engine/WorkflowUtils.java b/modules/workflow-model/workflow-engine/src/main/java/org/apache/airavata/workflow/engine/WorkflowUtils.java
new file mode 100644
index 0000000..e21a2e2
--- /dev/null
+++ b/modules/workflow-model/workflow-engine/src/main/java/org/apache/airavata/workflow/engine/WorkflowUtils.java
@@ -0,0 +1,56 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.airavata.workflow.engine;
+
+import java.util.List;
+
+import org.airavata.appcatalog.cpi.AppCatalogException;
+import org.airavata.appcatalog.cpi.ApplicationInterface;
+import org.apache.aiaravata.application.catalog.data.impl.AppCatalogFactory;
+import org.apache.airavata.model.util.ExecutionType;
+import org.apache.airavata.model.workspace.experiment.Experiment;
+import org.apache.airavata.workflow.catalog.WorkflowCatalogException;
+import org.apache.airavata.workflow.catalog.WorkflowCatalogFactory;
+
+public class WorkflowUtils {
+
+ public static ExecutionType getExecutionType(Experiment experiment){
+ try {
+ ApplicationInterface applicationInterface = AppCatalogFactory.getAppCatalog().getApplicationInterface();
+ List<String> allApplicationInterfaceIds = applicationInterface.getAllApplicationInterfaceIds();
+ String applicationId = experiment.getApplicationId();
+ if (allApplicationInterfaceIds.contains(applicationId)){
+ return ExecutionType.SINGLE_APP;
+ } else {
+ List<String> allWorkflows = WorkflowCatalogFactory.getWorkflowCatalog().getAllWorkflows();
+ if (allWorkflows.contains(applicationId)){
+ return ExecutionType.WORKFLOW;
+ }
+ }
+ } catch (AppCatalogException e) {
+ e.printStackTrace();
+ } catch (WorkflowCatalogException e) {
+ e.printStackTrace();
+ }
+ return ExecutionType.UNKNOWN;
+ }
+}
http://git-wip-us.apache.org/repos/asf/airavata/blob/92838e4a/modules/workflow-model/workflow-engine/src/main/java/org/apache/airavata/workflow/engine/interpretor/WorkflowInterpreter.java
----------------------------------------------------------------------
diff --git a/modules/workflow-model/workflow-engine/src/main/java/org/apache/airavata/workflow/engine/interpretor/WorkflowInterpreter.java b/modules/workflow-model/workflow-engine/src/main/java/org/apache/airavata/workflow/engine/interpretor/WorkflowInterpreter.java
index 3fe82f0..98fc560 100644
--- a/modules/workflow-model/workflow-engine/src/main/java/org/apache/airavata/workflow/engine/interpretor/WorkflowInterpreter.java
+++ b/modules/workflow-model/workflow-engine/src/main/java/org/apache/airavata/workflow/engine/interpretor/WorkflowInterpreter.java
@@ -41,10 +41,13 @@ import javax.xml.xpath.XPathExpressionException;
import javax.xml.xpath.XPathFactory;
import org.apache.airavata.client.api.exception.AiravataAPIInvocationException;
+import org.apache.airavata.common.utils.AbstractActivityListener;
import org.apache.airavata.common.utils.StringUtil;
import org.apache.airavata.common.utils.XMLUtil;
+import org.apache.airavata.gfac.core.monitor.state.TaskStatusChangedEvent;
import org.apache.airavata.model.util.ExperimentModelUtil;
import org.apache.airavata.model.workspace.experiment.DataObjectType;
+import org.apache.airavata.model.workspace.experiment.ExecutionUnit;
import org.apache.airavata.model.workspace.experiment.Experiment;
import org.apache.airavata.model.workspace.experiment.ExperimentState;
import org.apache.airavata.model.workspace.experiment.TaskDetails;
@@ -62,6 +65,7 @@ import org.apache.airavata.workflow.engine.invoker.DynamicInvoker;
import org.apache.airavata.workflow.engine.invoker.Invoker;
import org.apache.airavata.workflow.engine.util.AmazonUtil;
import org.apache.airavata.workflow.engine.util.InterpreterUtil;
+import org.apache.airavata.workflow.engine.util.ProxyMonitorPublisher;
import org.apache.airavata.workflow.model.component.Component;
import org.apache.airavata.workflow.model.component.amazon.InstanceComponent;
import org.apache.airavata.workflow.model.component.amazon.TerminateInstanceComponent;
@@ -112,7 +116,9 @@ import org.xmlpull.infoset.XmlElement;
import xsul5.XmlConstants;
-public class WorkflowInterpreter {
+import com.google.common.eventbus.Subscribe;
+
+public class WorkflowInterpreter implements AbstractActivityListener{
private static final Logger log = LoggerFactory.getLogger(WorkflowInterpreter.class);
public static final String WORKFLOW_STARTED = "Workflow Running";
@@ -133,6 +139,9 @@ public class WorkflowInterpreter {
private OrchestratorService.Client orchestratorClient;
+ private Map<String, Node> awaitingTasks;
+ private Map<Node, Map<String,String>> nodeOutputData;
+
public static ThreadLocal<WorkflowInterpreterConfiguration> workflowInterpreterConfigurationThreadLocal =
new ThreadLocal<WorkflowInterpreterConfiguration>();
@@ -151,6 +160,9 @@ public class WorkflowInterpreter {
//TODO set act of provenance
nodeInstanceList=new HashMap<Node, WorkflowNodeDetails>();
setWorkflowInterpreterConfigurationThreadLocal(config);
+ awaitingTasks = new HashMap<String, Node>();
+ nodeOutputData = new HashMap<Node, Map<String,String>>();
+ ProxyMonitorPublisher.registerListener(this);
}
public WorkflowInterpreterInteractor getInteractor(){
@@ -361,7 +373,16 @@ public class WorkflowInterpreter {
}
private WorkflowNodeDetails createWorkflowNodeDetails(Node node) {
- WorkflowNodeDetails workflowNode = ExperimentModelUtil.createWorkflowNode(node.getName(), null);
+ WorkflowNodeDetails workflowNode = ExperimentModelUtil.createWorkflowNode(node.getName(), null);
+ ExecutionUnit executionUnit = ExecutionUnit.APPLICATION;
+ if (node instanceof InputNode){
+ executionUnit = ExecutionUnit.INPUT;
+ } else if (node instanceof OutputNode){
+ executionUnit = ExecutionUnit.OUTPUT;
+ } if (node instanceof WSNode){
+ executionUnit = ExecutionUnit.APPLICATION;
+ }
+ workflowNode.setExecutionUnit(executionUnit);
nodeInstanceList.put(node, workflowNode);
return workflowNode;
}
@@ -401,7 +422,7 @@ public class WorkflowInterpreter {
// next run
// even if the next run runs before the notification arrives
WorkflowNodeDetails workflowNodeDetails = createWorkflowNodeDetails(node);
- getRegistry().update(RegistryModelType.WORKFLOW_NODE_DETAIL, workflowNodeDetails, getExperiment().getExperimentID());
+ workflowNodeDetails.setNodeInstanceId((String)getRegistry().add(ChildDataType.WORKFLOW_NODE_DETAIL, workflowNodeDetails, getExperiment().getExperimentID()));
node.setState(NodeExecutionState.EXECUTING);
updateWorkflowNodeStatus(workflowNodeDetails, WorkflowNodeState.EXECUTING);
// OutputNode node = (OutputNode) outputNode;
@@ -606,11 +627,37 @@ public class WorkflowInterpreter {
}
protected void handleWSComponent(Node node) throws WorkflowException, TException, RegistryException {
- TaskDetails taskDetails = ExperimentModelUtil.cloneTaskFromWorkflowNodeDetails(experiment, nodeInstanceList.get(node));
- taskDetails.setTaskID(getRegistry().add(ChildDataType.TASK_DETAIL, taskDetails,nodeInstanceList.get(node).getNodeInstanceId()).toString());
+ TaskDetails taskDetails = createTaskDetails(node);
getOrchestratorClient().launchTask(taskDetails.getTaskID(), getCredentialStoreToken());
}
+
+ private void addToAwaitingTaskList(String taskId, Node node){
+ synchronized (awaitingTasks) {
+ awaitingTasks.put(taskId, node);
+ }
+ }
+
+ private boolean isTaskAwaiting(String taskId){
+ boolean result;
+ synchronized (awaitingTasks) {
+ result = awaitingTasks.containsKey(taskId);
+ }
+ return result;
+ }
+ private Node getAwaitingNodeForTask(String taskId){
+ Node node;
+ synchronized (awaitingTasks) {
+ node = awaitingTasks.get(taskId);
+ }
+ return node;
+ }
+
+ private void removeAwaitingTask(String taskId){
+ synchronized (awaitingTasks) {
+ awaitingTasks.remove(taskId);
+ }
+ }
private void handleDynamicComponent(Node node) throws WorkflowException {
DynamicComponent dynamicComponent = (DynamicComponent) node.getComponent();
String className = dynamicComponent.getClassName();
@@ -930,49 +977,57 @@ public class WorkflowInterpreter {
node.setState(NodeExecutionState.FINISHED);
}
+
private String createInvokerForEachSingleWSNode(Node foreachWSNode, WSComponent wsComponent) throws WorkflowException, RegistryException, TException {
- TaskDetails taskDetails = ExperimentModelUtil.cloneTaskFromWorkflowNodeDetails(experiment, nodeInstanceList.get(foreachWSNode));
- taskDetails.setTaskID(getRegistry().add(ChildDataType.TASK_DETAIL, taskDetails,nodeInstanceList.get(foreachWSNode).getNodeInstanceId()).toString());
+ TaskDetails taskDetails = createTaskDetails(foreachWSNode);
getOrchestratorClient().launchTask(taskDetails.getTaskID(), getCredentialStoreToken());
-
-// if (null == wsdlLocation) {
- // WorkflowInterpreter is no longer using gfac in service mode. we only support embedded mode.
-// if (gfacURLString.startsWith("https")) {
-// LeadContextHeader leadCtxHeader = null;
-// try {
-// leadCtxHeader = XBayaUtil.buildLeadContextHeader(this.getWorkflow(), this.getConfig().getConfiguration(), new MonitorConfiguration(this
-// .getConfig().getConfiguration().getBrokerURL(), this.config.getTopic(), true, this.getConfig().getConfiguration()
-// .getMessageBoxURL()), foreachWSNode.getID(), null);
-// } catch (URISyntaxException e) {
-// throw new WorkflowException(e);
-// }
-// invoker = new WorkflowInvokerWrapperForGFacInvoker(portTypeQName, gfacURLString, this.getConfig().getConfiguration().getMessageBoxURL()
-// .toString(), leadCtxHeader, this.config.getNotifier().createServiceNotificationSender(foreachWSNode.getID()));
-// } else {
-
-// if (this.config.isGfacEmbeddedMode()) {
-// invoker = new EmbeddedGFacInvoker(portTypeQName, WSDLUtil.wsdlDefinitions5ToWsdlDefintions3(((WSNode)foreachWSNode).getComponent().getWSDL()), foreachWSNode.getID(),
-// this.config.getMessageBoxURL().toASCIIString(), this.config.getMessageBrokerURL().toASCIIString(), this.config.getNotifier(),
-// this.config.getTopic(), this.config.getAiravataAPI(), portTypeQName.getLocalPart(), this.config.getConfiguration());
-// } else {
-// invoker = new GenericInvoker(portTypeQName, WSDLUtil.wsdlDefinitions5ToWsdlDefintions3(((WSNode)foreachWSNode).getComponent().getWSDL()), foreachWSNode.getID(),
-// this.config.getMessageBoxURL().toASCIIString(), gfacURLString, this.config.getNotifier());
-// }
-// }
-
-// } else {
-// if (wsdlLocation.endsWith("/")) {
-// wsdlLocation = wsdlLocation.substring(0, wsdlLocation.length() - 1);
-// }
-// if (!wsdlLocation.endsWith("?wsdl")) {
-// wsdlLocation += "?wsdl";
-// }
-// invoker = new GenericInvoker(portTypeQName, wsdlLocation, foreachWSNode.getID(), this.getConfig().getConfiguration().getMessageBoxURL().toString(),
-// gfacURLString, this.config.getNotifier());
-// }
return taskDetails.getTaskID();
}
+ private void setupNodeDetailsInput(Node node, WorkflowNodeDetails nodeDetails){
+ List<DataPort> inputPorts = node.getInputPorts();
+ for (DataPort dataPort : inputPorts) {
+ Map<String, String> outputData = nodeOutputData.get(dataPort.getFromNode());
+ String portInputValue = outputData.get(dataPort.getName());
+ DataObjectType elem = new DataObjectType();
+ elem.setKey(dataPort.getName());
+ elem.setValue(portInputValue);
+ nodeDetails.addToNodeInputs(elem);
+ }
+ try {
+ getRegistry().update(RegistryModelType.WORKFLOW_NODE_DETAIL, nodeDetails, nodeDetails.getNodeInstanceId());
+ } catch (RegistryException e) {
+ e.printStackTrace();
+ }
+ }
+
+ private void setupNodeDetailsOutput(Node node){
+ WorkflowNodeDetails nodeDetails = nodeInstanceList.get(node);
+ List<DataPort> outputPorts = node.getOutputPorts();
+ Map<String, String> outputData = nodeOutputData.get(node);
+ for (DataPort dataPort : outputPorts) {
+ String portInputValue = outputData.get(dataPort.getName());
+ DataObjectType elem = new DataObjectType();
+ elem.setKey(dataPort.getName());
+ elem.setValue(portInputValue);
+ nodeDetails.addToNodeOutputs(elem);
+ }
+ try {
+ getRegistry().update(RegistryModelType.WORKFLOW_NODE_DETAIL, nodeDetails, nodeDetails.getNodeInstanceId());
+ } catch (RegistryException e) {
+ e.printStackTrace();
+ }
+ }
+
+ private TaskDetails createTaskDetails(Node node)
+ throws RegistryException {
+ setupNodeDetailsInput(node, nodeInstanceList.get(node));
+ TaskDetails taskDetails = ExperimentModelUtil.cloneTaskFromWorkflowNodeDetails(experiment, nodeInstanceList.get(node));
+ taskDetails.setTaskID(getRegistry().add(ChildDataType.TASK_DETAIL, taskDetails,nodeInstanceList.get(node).getNodeInstanceId()).toString());
+ addToAwaitingTaskList(taskDetails.getTaskID(), node);
+ return taskDetails;
+ }
+
private void runInThread(final LinkedList<String> listOfValues, ForEachNode forEachNode, final Node middleNode, List<Node> endForEachNodes,
Map<Node, Invoker> tempInvoker, AtomicInteger counter, final Integer[] inputNumber) throws WorkflowException, RegistryException, TException {
@@ -1326,83 +1381,55 @@ public class WorkflowInterpreter {
this.credentialStoreToken = credentialStoreToken;
}
- public static class TmpRegistry implements Registry{
- public TmpRegistry() throws RegistryException {
- }
-
- @Override
- public Object add(ChildDataType dataType, Object newObjectToAdd,
- Object dependentIdentifier) throws RegistryException {
- return UUID.randomUUID().toString();
- }
+ @Override
+ public void setup(Object... configurations) {
+ // TODO Auto-generated method stub
- @Override
- public void update(RegistryModelType dataType, Object identifier,
- String fieldName, Object value) throws RegistryException {
- }
-
- @Override
- public Object add(ParentDataType dataType, Object newObjectToAdd)
- throws RegistryException {
- // TODO Auto-generated method stub
- return null;
- }
-
- @Override
- public void update(RegistryModelType dataType,
- Object newObjectToUpdate, Object identifier)
- throws RegistryException {
- // TODO Auto-generated method stub
-
- }
-
- @Override
- public Object get(RegistryModelType dataType, Object identifier)
- throws RegistryException {
- // TODO Auto-generated method stub
- return null;
- }
-
- @Override
- public List<Object> get(RegistryModelType dataType, String fieldName,
- Object value) throws RegistryException {
- // TODO Auto-generated method stub
- return null;
- }
-
- @Override
- public List<Object> search(RegistryModelType dataType,
- Map<String, String> filters) throws RegistryException {
- // TODO Auto-generated method stub
- return null;
- }
-
- @Override
- public Object getValue(RegistryModelType dataType, Object identifier,
- String field) throws RegistryException {
- // TODO Auto-generated method stub
- return null;
- }
-
- @Override
- public List<String> getIds(RegistryModelType dataType,
- String fieldName, Object value) throws RegistryException {
- // TODO Auto-generated method stub
- return null;
- }
-
- @Override
- public void remove(RegistryModelType dataType, Object identifier)
- throws RegistryException {
- // TODO Auto-generated method stub
-
- }
-
- @Override
- public boolean isExist(RegistryModelType dataType, Object identifier)
- throws RegistryException {
- // TODO Auto-generated method stub
- return false;
- }
}
+
+ @Subscribe
+ public void taskStatusChanged(TaskStatusChangedEvent taskStatus){
+ String taskId = taskStatus.getIdentity().getTaskId();
+ if (isTaskAwaiting(taskId)){
+ WorkflowNodeState state=WorkflowNodeState.UNKNOWN;
+ switch(taskStatus.getState()){
+ case CANCELED:
+ ; break;
+ case COMPLETED:
+ //task is completed
+ try {
+ TaskDetails task = (TaskDetails)getRegistry().get(RegistryModelType.TASK_DETAIL, taskId);
+ List<DataObjectType> applicationOutputs = task.getApplicationOutputs();
+ Map<String, String> outputData = new HashMap<String, String>();
+ Node node = getAwaitingNodeForTask(taskId);
+ for (DataObjectType outputObj : applicationOutputs) {
+ List<DataPort> outputPorts = node.getOutputPorts();
+ for (DataPort dataPort : outputPorts) {
+ if (dataPort.getName().equals(outputObj.getKey())){
+ outputData.put(outputObj.getKey(), outputObj.getValue());
+ }
+ }
+ }
+ nodeOutputData.put(node, outputData);
+ setupNodeDetailsOutput(node);
+ } catch (RegistryException e) {
+ e.printStackTrace();
+ }
+ break;
+ case CONFIGURING_WORKSPACE:
+ break;
+ case FAILED:
+ break;
+ case EXECUTING: case WAITING: case PRE_PROCESSING: case POST_PROCESSING: case OUTPUT_DATA_STAGING: case INPUT_DATA_STAGING:
+ break;
+ case STARTED:
+ break;
+ case CANCELING:
+ break;
+ default:
+ break;
+ }
+ }
+
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/airavata/blob/92838e4a/modules/workflow-model/workflow-engine/src/main/java/org/apache/airavata/workflow/engine/util/ProxyMonitorPublisher.java
----------------------------------------------------------------------
diff --git a/modules/workflow-model/workflow-engine/src/main/java/org/apache/airavata/workflow/engine/util/ProxyMonitorPublisher.java b/modules/workflow-model/workflow-engine/src/main/java/org/apache/airavata/workflow/engine/util/ProxyMonitorPublisher.java
new file mode 100644
index 0000000..507e1bb
--- /dev/null
+++ b/modules/workflow-model/workflow-engine/src/main/java/org/apache/airavata/workflow/engine/util/ProxyMonitorPublisher.java
@@ -0,0 +1,62 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.airavata.workflow.engine.util;
+
+import org.apache.airavata.common.utils.AbstractActivityListener;
+import org.apache.airavata.common.utils.MonitorPublisher;
+
+public class ProxyMonitorPublisher implements AbstractActivityListener{
+
+ private static Object[] setupConfigurations;
+
+ @Override
+ public void setup(Object... configurations) {
+ setupConfigurations=configurations;
+ }
+
+ private static MonitorPublisher getPublisher(){
+ if (setupConfigurations!=null) {
+ for (Object configuration : setupConfigurations) {
+ if (configuration instanceof MonitorPublisher){
+ return (MonitorPublisher) configuration;
+ }
+ }
+ }
+ return null;
+ }
+
+ public static void registerListener(Object listener) {
+ if (listener instanceof AbstractActivityListener){
+ ((AbstractActivityListener) listener).setup(setupConfigurations);
+ }
+ getPublisher().registerListener(listener);
+ }
+
+ public static void unregisterListener(Object listener) {
+ getPublisher().unregisterListener(listener);
+ }
+
+ public void publish(Object o) {
+ getPublisher().publish(o);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/airavata/blob/92838e4a/modules/workflow-model/workflow-model-core/src/main/java/org/apache/airavata/workflow/model/component/ws/WSComponent.java
----------------------------------------------------------------------
diff --git a/modules/workflow-model/workflow-model-core/src/main/java/org/apache/airavata/workflow/model/component/ws/WSComponent.java b/modules/workflow-model/workflow-model-core/src/main/java/org/apache/airavata/workflow/model/component/ws/WSComponent.java
index ac86b05..606c88b 100644
--- a/modules/workflow-model/workflow-model-core/src/main/java/org/apache/airavata/workflow/model/component/ws/WSComponent.java
+++ b/modules/workflow-model/workflow-model-core/src/main/java/org/apache/airavata/workflow/model/component/ws/WSComponent.java
@@ -204,7 +204,7 @@ public class WSComponent extends Component {
buf.append("<h1>Application: " + getName() + "</h1>\n");
buf.append("<h2>Description:</h2>\n");
- buf.append(this.description==null?"":this.description);
+ buf.append(application.getApplicationId()+"<br />"+(this.description==null?"":this.description));
if (getInputPorts().size()>0) {
buf.append("<h3>Input parameter(s)</h3>\n");
http://git-wip-us.apache.org/repos/asf/airavata/blob/92838e4a/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/ui/dialogs/registry/RegistryWindow.java
----------------------------------------------------------------------
diff --git a/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/ui/dialogs/registry/RegistryWindow.java b/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/ui/dialogs/registry/RegistryWindow.java
index 61040ac..8bcc2bc 100644
--- a/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/ui/dialogs/registry/RegistryWindow.java
+++ b/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/ui/dialogs/registry/RegistryWindow.java
@@ -141,7 +141,7 @@ public class RegistryWindow {
XBayaLabel serverAddressLabel = new XBayaLabel("Server Address", this.serverTextField);
XBayaLabel serverPortLabel = new XBayaLabel("Server Port", this.portTextField);
XBayaLabel gatewayNameLabel = new XBayaLabel("Gateway Name", this.gatewayNameTextField);
- XBayaLabel gatewayUserLabel = new XBayaLabel("Gateway TUser", this.usernameTextField);
+ XBayaLabel gatewayUserLabel = new XBayaLabel("Gateway User", this.usernameTextField);
serviceTypeModel = new DefaultComboBoxModel(ThriftServiceType.values());
serviceTypeModel.setSelectedItem(getServiceType());
this.serviceTypeCombo = new XBayaComboBox(serviceTypeModel);
@@ -188,7 +188,7 @@ public class RegistryWindow {
buttonPanel.add(cancelButton);
buttonPanel.getSwingComponent().setBorder(BorderFactory.createEtchedBorder());
- this.dialog = new XBayaDialog(this.engine.getGUI(), "Configure Airavata Registry", infoPanel, buttonPanel);
+ this.dialog = new XBayaDialog(this.engine.getGUI(), "Configure Airavata Thrift Service", infoPanel, buttonPanel);
this.dialog.setDefaultButton(okButton);
}
http://git-wip-us.apache.org/repos/asf/airavata/blob/92838e4a/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/ui/experiment/WorkflowInterpreterLaunchWindow.java
----------------------------------------------------------------------
diff --git a/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/ui/experiment/WorkflowInterpreterLaunchWindow.java b/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/ui/experiment/WorkflowInterpreterLaunchWindow.java
index 2671d30..fd6cb46 100644
--- a/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/ui/experiment/WorkflowInterpreterLaunchWindow.java
+++ b/modules/xbaya-gui/src/main/java/org/apache/airavata/xbaya/ui/experiment/WorkflowInterpreterLaunchWindow.java
@@ -22,7 +22,6 @@
package org.apache.airavata.xbaya.ui.experiment;
import java.awt.event.ActionEvent;
-import java.net.URI;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
@@ -30,10 +29,6 @@ import java.util.UUID;
//import org.apache.airavata.registry.api.AiravataRegistry2;
-
-
-
-
import javax.swing.AbstractAction;
import javax.swing.BorderFactory;
import javax.swing.JButton;
@@ -43,9 +38,14 @@ import javax.swing.JPanel;
import javax.swing.border.EtchedBorder;
import javax.xml.namespace.QName;
+import org.apache.airavata.api.Airavata.Client;
import org.apache.airavata.common.utils.ServerSettings;
-import org.apache.airavata.common.utils.StringUtil;
import org.apache.airavata.common.utils.XMLUtil;
+import org.apache.airavata.model.error.AiravataClientConnectException;
+import org.apache.airavata.model.error.AiravataClientException;
+import org.apache.airavata.model.error.AiravataSystemException;
+import org.apache.airavata.model.error.InvalidRequestException;
+import org.apache.airavata.model.workspace.Project;
import org.apache.airavata.model.workspace.experiment.DataObjectType;
import org.apache.airavata.model.workspace.experiment.DataType;
import org.apache.airavata.model.workspace.experiment.Experiment;
@@ -57,25 +57,21 @@ import org.apache.airavata.workflow.engine.interpretor.WorkflowInterpreterConfig
import org.apache.airavata.workflow.model.exceptions.WorkflowException;
import org.apache.airavata.workflow.model.graph.system.InputNode;
import org.apache.airavata.workflow.model.graph.util.GraphUtil;
-import org.apache.airavata.workflow.model.graph.ws.WSNode;
-import org.apache.airavata.workflow.model.ode.ODEClient;
import org.apache.airavata.workflow.model.wf.Workflow;
import org.apache.airavata.ws.monitor.MonitorConfiguration;
+import org.apache.airavata.xbaya.ThriftClientData;
+import org.apache.airavata.xbaya.ThriftServiceType;
import org.apache.airavata.xbaya.XBayaEngine;
-import org.apache.airavata.xbaya.graph.controller.NodeController;
-import org.apache.airavata.xbaya.jython.script.JythonScript;
import org.apache.airavata.xbaya.ui.dialogs.XBayaDialog;
-import org.apache.airavata.xbaya.ui.graph.ws.WSNodeGUI;
import org.apache.airavata.xbaya.ui.utils.ErrorMessages;
import org.apache.airavata.xbaya.ui.widgets.GridPanel;
import org.apache.airavata.xbaya.ui.widgets.XBayaLabel;
import org.apache.airavata.xbaya.ui.widgets.XBayaTextField;
+import org.apache.airavata.xbaya.util.XBayaUtil;
+import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xmlpull.infoset.XmlElement;
-import org.xmlpull.v1.builder.XmlInfosetBuilder;
-
-import xsul.XmlConstants;
public class WorkflowInterpreterLaunchWindow {
@@ -101,7 +97,7 @@ public class WorkflowInterpreterLaunchWindow {
private XBayaTextField instanceNameTextField;
- protected final static XmlInfosetBuilder builder = XmlConstants.BUILDER;
+// protected final static XmlInfosetBuilder builder = XmlConstants.BUILDER;
/**
* Constructs a WorkflowInterpreterLaunchWindow.
@@ -228,7 +224,23 @@ public class WorkflowInterpreterLaunchWindow {
JButton okButton = new JButton("Run");
okButton.addActionListener(new AbstractAction() {
public void actionPerformed(ActionEvent e) {
- execute();
+ try {
+ execute();
+ } catch (AiravataClientConnectException e1) {
+ e1.printStackTrace();
+ } catch (InvalidRequestException e1) {
+ // TODO Auto-generated catch block
+ e1.printStackTrace();
+ } catch (AiravataClientException e1) {
+ // TODO Auto-generated catch block
+ e1.printStackTrace();
+ } catch (AiravataSystemException e1) {
+ // TODO Auto-generated catch block
+ e1.printStackTrace();
+ } catch (TException e1) {
+ // TODO Auto-generated catch block
+ e1.printStackTrace();
+ }
}
});
@@ -247,10 +259,24 @@ public class WorkflowInterpreterLaunchWindow {
this.dialog.setDefaultButton(okButton);
}
- private void execute() {
- final List<String> arguments = new ArrayList<String>();
-
- String topic = UUID.randomUUID().toString();//this.topicTextField.getText();
+ private void execute() throws AiravataClientConnectException, InvalidRequestException, AiravataClientException, AiravataSystemException, TException {
+
+ if (!(engine.getGUI().setupThriftClientData(ThriftServiceType.API_SERVICE) && engine.getGUI().setupThriftClientData(ThriftServiceType.WORKFLOW_SERVICE))){
+ hide();
+ return;
+ }
+
+ ThriftClientData thriftClientData = engine.getConfiguration().getThriftClientData(ThriftServiceType.API_SERVICE);
+ Client airavataClient = XBayaUtil.getAiravataClient(thriftClientData);
+
+ org.apache.airavata.api.workflow.Workflow.Client workflowClient = XBayaUtil.getWorkflowClient(engine.getConfiguration().getThriftClientData(ThriftServiceType.WORKFLOW_SERVICE));
+
+ Workflow workflowClone = workflow.clone();
+ workflowClone.setName(workflowClone.getName()+UUID.randomUUID().toString());
+ org.apache.airavata.model.Workflow w = new org.apache.airavata.model.Workflow();
+ w.setName(workflowClone.getName());
+ w.setGraph(XMLUtil.xmlElementToString(workflowClone.toXML()));
+ w.setTemplateId(workflowClient.registerWorkflow(w));
String instanceName = this.instanceNameTextField.getText();
if (instanceName.trim().equals("")){
JOptionPane.showMessageDialog(engine.getGUI().getFrame(),
@@ -264,38 +290,22 @@ public class WorkflowInterpreterLaunchWindow {
if (!instanceNameTextField.getText().equals("")){
this.instanceNameTextField.setText("");
}
- final String instanceNameFinal=instanceName;
- if (topic.length() == 0) {
- this.engine.getGUI().getErrorWindow().error(ErrorMessages.TOPIC_EMPTY_ERROR);
- return;
- }
// Use topic as a base of workflow instance ID so that the monitor can
// find it.
- URI workfowInstanceID = URI.create(StringUtil.convertToJavaIdentifier(topic));
- this.workflow.setGPELInstanceID(workfowInstanceID);
-
- MonitorConfiguration notifConfig = this.engine.getMonitor().getConfiguration();
- engine.getMonitor().resetEventData();
- notifConfig.setTopic(topic);
- arguments.add("-" + JythonScript.TOPIC_VARIABLE);
- arguments.add(topic);
- Collection<WSNode> wsNodes = GraphUtil.getWSNodes(this.engine.getGUI().getWorkflow().getGraph());
- for (WSNode node : wsNodes) {
- ((WSNodeGUI) NodeController.getGUI(node)).setInteractiveMode(false);
- }
-
- // TODO error check for user inputs
+ Project project = new Project();
+ project.setName("project1");
+ project.setOwner(thriftClientData.getUsername());
+ project.setProjectID(airavataClient.createProject(project));
final List<InputNode> inputNodes = GraphUtil.getInputNodes(this.workflow.getGraph());
- builder.newFragment("inputs");
- new ODEClient();
+// builder.newFragment("inputs");
+// new ODEClient();
final Experiment experiment = new Experiment();
- experiment.setApplicationId(workflow.getName());
- experiment.setExperimentID("some-id");
- experiment.setProjectID("prj1");
- experiment.setName("some-exp");
- experiment.setUserName("me");
+ experiment.setApplicationId(w.getTemplateId());
+ experiment.setName(instanceName);
+ experiment.setProjectID(project.getProjectID());
+ experiment.setUserName(thriftClientData.getUsername());
for (int i = 0; i < inputNodes.size(); i++) {
InputNode inputNode = inputNodes.get(i);
XBayaTextField parameterTextField = this.parameterTextFields.get(i);
@@ -308,8 +318,9 @@ public class WorkflowInterpreterLaunchWindow {
elem.setValue(value);
experiment.addToExperimentInputs(elem );
}
-
-
+
+ experiment.setExperimentID(airavataClient.createExperiment(experiment));
+ airavataClient.launchExperiment(experiment.getExperimentID(), "testToken");
// final String workflowInterpreterUrl = this.workflowInterpreterTextField.getText();
// if (null != workflowInterpreterUrl && !"".equals(workflowInterpreterUrl)) {
// try {
@@ -327,20 +338,20 @@ public class WorkflowInterpreterLaunchWindow {
// this.engine.getGUI().getErrorWindow().error(e);
// }
// }
- this.engine.getConfiguration().setTopic(topic);
+// this.engine.getConfiguration().setTopic(topic);
new Thread() {
@Override
public void run() {
- WorkflowInterpreter workflowInterpreter = new WorkflowInterpreter(experiment, null, new WorkflowInterpreterConfiguration(workflow),getOrchestratorClient());
- try {
- workflowInterpreter.scheduleDynamically();
- } catch (WorkflowException e) {
- e.printStackTrace();
- } catch (RegistryException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
+// WorkflowInterpreter workflowInterpreter = new WorkflowInterpreter(experiment, null, new WorkflowInterpreterConfiguration(workflow),getOrchestratorClient());
+// try {
+// workflowInterpreter.scheduleDynamically();
+// } catch (WorkflowException e) {
+// e.printStackTrace();
+// } catch (RegistryException e) {
+// // TODO Auto-generated catch block
+// e.printStackTrace();
+// }
// try {
// List<WorkflowInput> workflowInputs=new ArrayList<WorkflowInput>();
// for (int i = 0; i < inputNodes.size(); i++) {