This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch staging
in repository https://gitbox.apache.org/repos/asf/airavata.git
The following commit(s) were added to refs/heads/staging by this push:
new 4cc9438 Adding new parameter resolutions and bug fixed to parser wm
4cc9438 is described below
commit 4cc94389ad3ff9681620e899fc99adfab8773343
Author: Dimuthu Wannipurage <[email protected]>
AuthorDate: Thu Jul 25 16:15:17 2019 -0400
Adding new parameter resolutions and bug fixed to parser wm
---
.../helix/impl/workflow/ParserWorkflowManager.java | 66 ++++++++++++++--------
1 file changed, 42 insertions(+), 24 deletions(-)
diff --git
a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/ParserWorkflowManager.java
b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/ParserWorkflowManager.java
index a86b8a4..bb27638 100644
---
a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/ParserWorkflowManager.java
+++
b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/ParserWorkflowManager.java
@@ -55,7 +55,7 @@ public class ParserWorkflowManager extends WorkflowManager {
private final static Logger logger =
LoggerFactory.getLogger(ParserWorkflowManager.class);
- private String parserStorageResourceId =
"pgadev.scigap.org_7ddf28fd-d503-4ff8-bbc5-3279a7c3b99e";
+ private String parserStorageResourceId =
ServerSettings.getSetting("parser.storage.resource.id");
public ParserWorkflowManager() throws ApplicationSettingsException {
super(ServerSettings.getSetting("parser.workflow.manager.name"),
@@ -117,20 +117,24 @@ public class ParserWorkflowManager extends
WorkflowManager {
for (ParsingTemplate template : parsingTemplates) {
logger.info("Launching parsing template " + template.getId());
- String parentParserId = null;
- for (String parentId :
parentToChildParsers.get(template.getId()).keySet()) {
- boolean found = false;
- for (Set<ParserConnector> dagElements :
parentToChildParsers.get(template.getId()).values()) {
- Optional<ParserConnector> first =
dagElements.stream().filter(dagElement ->
dagElement.getChildParserId().equals(parentId)).findFirst();
- if (first.isPresent()) {
- found = true;
+ ParserInput parserInput =
registryClient.getParserInput(template.getInitialInputs().get(0).getTargetInputId(),
template.getGatewayId());
+ String parentParserId = parserInput.getParserId();
+
+ if (!parentToChildParsers.isEmpty()) {
+ for (String parentId :
parentToChildParsers.get(template.getId()).keySet()) {
+ boolean found = false;
+ for (Set<ParserConnector> dagElements :
parentToChildParsers.get(template.getId()).values()) {
+ Optional<ParserConnector> first =
dagElements.stream().filter(dagElement ->
dagElement.getChildParserId().equals(parentId)).findFirst();
+ if (first.isPresent()) {
+ found = true;
+ break;
+ }
+ }
+ if (!found) {
+ parentParserId = parentId;
break;
}
}
- if (!found) {
- parentParserId = parentId;
- break;
- }
}
if (parentParserId == null ) {
@@ -143,8 +147,10 @@ public class ParserWorkflowManager extends WorkflowManager
{
List<AbstractTask> allTasks = new ArrayList<>();
allTasks.add(parentParserTask);
- createParserDagRecursively(allTasks, parentParser,
parentParserTask, parentToChildParsers.get(template.getId()),
completionMessage, registryClient);
+ if (parentToChildParsers.containsKey(template.getId())) {
+ createParserDagRecursively(allTasks, parentParser,
parentParserTask, parentToChildParsers.get(template.getId()),
completionMessage, registryClient);
+ }
String workflow =
getWorkflowOperator().launchWorkflow("Parser-" +
completionMessage.getProcessId() + UUID.randomUUID().toString(),
allTasks, true, false);
// TODO: figure out processId and register
@@ -234,20 +240,32 @@ public class ParserWorkflowManager extends
WorkflowManager {
return parsingTask;
}
- private String processExpression(String expression,
ProcessCompletionMessage completionMessage) {
- if (expression != null) {
- if (expression.startsWith("{{") && expression.endsWith("}}")) {
- switch (expression) {
- case "{{experimentId}}":
- return completionMessage.getExperimentId();
- case "{{processId}}":
- return completionMessage.getProcessId();
- case "{{gateway}}":
- return completionMessage.getGatewayId();
+ private String processExpression(String expression,
ProcessCompletionMessage completionMessage) throws Exception {
+ RegistryService.Client registryClient =
getRegistryClientPool().getResource();
+
+ try {
+ if (expression != null) {
+ if (expression.startsWith("{{") && expression.endsWith("}}")) {
+ switch (expression) {
+ case "{{experiment}}":
+ return completionMessage.getExperimentId();
+ case "{{process}}":
+ return completionMessage.getProcessId();
+ case "{{gateway}}":
+ return completionMessage.getGatewayId();
+ case "{{user}}":
+ return
registryClient.getProcess(completionMessage.getProcessId()).getUserName();
+ case "{{project}}":
+ return
registryClient.getExperiment(completionMessage.getExperimentId()).getProjectId();
+ }
}
}
+ getRegistryClientPool().returnResource(registryClient);
+ return expression;
+ } catch (Exception e) {
+ getRegistryClientPool().returnBrokenResource(registryClient);
+ throw new Exception("Failed to resolve expression " + expression,
e);
}
- return expression;
}
private void createParserDagRecursively(List<AbstractTask> allTasks,
Parser parentParserInfo, DataParsingTask parentTask, Map<String,
Set<ParserConnector>> parentToChild,
