Revision 47366
Added by Michele Artini almost 7 years ago
| RegisterWorkflowFromTemplateJobNode.java | ||
|---|---|---|
| 19 | 19 |
|
| 20 | 20 |
import eu.dnetlib.clients.is.InformationServiceClient; |
| 21 | 21 |
import eu.dnetlib.miscutils.templates.TemplateUtil; |
| 22 |
import eu.dnetlib.msro.annotations.Direction; |
|
| 23 |
import eu.dnetlib.msro.annotations.EnvParam; |
|
| 24 | 22 |
import eu.dnetlib.msro.annotations.ProcessNode; |
| 25 | 23 |
import eu.dnetlib.msro.workflows.Arc; |
| 26 | 24 |
import eu.dnetlib.msro.workflows.WorkflowStatus; |
| 27 |
import eu.dnetlib.msro.workflows.nodes.AbstractParallelProcessNode;
|
|
| 25 |
import eu.dnetlib.msro.workflows.nodes.AbstractProcessNode; |
|
| 28 | 26 |
import eu.dnetlib.msro.workflows.procs.ProcessAware; |
| 27 |
import eu.dnetlib.msro.workflows.procs.Token; |
|
| 29 | 28 |
import eu.dnetlib.msro.workflows.procs.WorkflowProcess; |
| 30 | 29 |
|
| 31 | 30 |
@Component |
| 32 | 31 |
@Scope("prototype")
|
| 33 | 32 |
@ProcessNode("RegisterWorkflowFromTemplate")
|
| 34 |
public class RegisterWorkflowFromTemplateJobNode extends AbstractParallelProcessNode implements ProcessAware {
|
|
| 33 |
public class RegisterWorkflowFromTemplateJobNode extends AbstractProcessNode implements ProcessAware {
|
|
| 35 | 34 |
|
| 36 | 35 |
private String wfName; |
| 37 | 36 |
private String wfDescription; |
| ... | ... | |
| 45 | 44 |
@Autowired |
| 46 | 45 |
private InformationServiceClient isClient; |
| 47 | 46 |
|
| 48 |
@EnvParam(value = "wfId", direction = Direction.OUT) |
|
| 49 |
private String wfId; |
|
| 50 |
|
|
| 51 | 47 |
private WorkflowProcess process; |
| 52 | 48 |
|
| 53 | 49 |
private static final Log log = LogFactory.getLog(RegisterWorkflowFromTemplateJobNode.class); |
| 54 | 50 |
|
| 55 | 51 |
@Override |
| 56 |
protected String execute() throws Exception {
|
|
| 57 |
final Map<String, Object> map = new HashMap<>(); |
|
| 52 |
public void execute(final Token token) {
|
|
| 53 |
try {
|
|
| 54 |
final Map<String, Object> map = new HashMap<>(); |
|
| 58 | 55 |
|
| 59 |
if (simpleParams != null) {
|
|
| 60 |
envParams.forEach((k, v) -> map.put(k, StringEscapeUtils.escapeXml11(v))); |
|
| 61 |
} |
|
| 56 |
if (simpleParams != null) {
|
|
| 57 |
envParams.forEach((k, v) -> map.put(k, StringEscapeUtils.escapeXml11(v)));
|
|
| 58 |
}
|
|
| 62 | 59 |
|
| 63 |
if (envParams != null) {
|
|
| 64 |
envParams.forEach((k, v) -> map.put(k, StringEscapeUtils.escapeXml11(process.getEnv().getAttribute(v).toString()))); |
|
| 65 |
} |
|
| 60 |
if (envParams != null) {
|
|
| 61 |
envParams.forEach((k, v) -> {
|
|
| 62 |
if (token.getEnv().hasAttribute(v)) {
|
|
| 63 |
final Object rv = token.getEnv().getAttribute(v); |
|
| 64 |
if (rv != null) {
|
|
| 65 |
log.info(String.format("%s -> env(%s) = %s", k, v, rv));
|
|
| 66 |
map.put(k, StringEscapeUtils.escapeXml11(rv.toString())); |
|
| 67 |
} else {
|
|
| 68 |
log.warn(String.format("Env param %s is NULL", v));
|
|
| 69 |
} |
|
| 70 |
} else {
|
|
| 71 |
log.warn(String.format("Env param %s is missing, valid params are %s", v, token.getEnv().getAttributes().keySet()));
|
|
| 72 |
} |
|
| 73 |
}); |
|
| 74 |
} |
|
| 66 | 75 |
|
| 67 |
map.put("name", StringEscapeUtils.escapeXml11(wfName));
|
|
| 68 |
map.put("desc", StringEscapeUtils.escapeXml11(wfDescription));
|
|
| 69 |
map.put("worker", StringEscapeUtils.escapeXml11(process.getWorkerId()));
|
|
| 70 |
map.put("dsId", StringEscapeUtils.escapeXml11(process.getDsId()));
|
|
| 71 |
map.put("iface", StringEscapeUtils.escapeXml11(process.getDsInterface()));
|
|
| 76 |
map.put("name", StringEscapeUtils.escapeXml11(wfName));
|
|
| 77 |
map.put("desc", StringEscapeUtils.escapeXml11(wfDescription));
|
|
| 78 |
map.put("worker", StringEscapeUtils.escapeXml11(process.getWorkerId()));
|
|
| 79 |
map.put("dsId", StringEscapeUtils.escapeXml11(process.getDsId()));
|
|
| 80 |
map.put("iface", StringEscapeUtils.escapeXml11(process.getDsInterface()));
|
|
| 72 | 81 |
|
| 73 |
final String wfProfile = templateUtil.processTemplateProfile(wfTemplate, map).get("wf");
|
|
| 82 |
final String wfProfile = templateUtil.processTemplateProfile(wfTemplate, map).get("wf");
|
|
| 74 | 83 |
|
| 75 |
final Document doc = new SAXReader().read(new StringReader(wfProfile)); |
|
| 76 |
final Node paramsNode = doc.selectSingleNode("//CONFIGURATION/PARAMETERS");
|
|
| 84 |
final Document doc = new SAXReader().read(new StringReader(wfProfile));
|
|
| 85 |
final Node paramsNode = doc.selectSingleNode("//CONFIGURATION/PARAMETERS");
|
|
| 77 | 86 |
|
| 78 |
final List<?> emptySysParams = paramsNode.selectNodes("//PARAM[@managedBy='system' and @required='true' and not(text())]");
|
|
| 79 |
if (emptySysParams.size() > 0) {
|
|
| 80 |
((Element) doc.selectSingleNode("//CONFIGURATION")).addAttribute("status", WorkflowStatus.WAIT_SYS_SETTINGS.toString());
|
|
| 81 |
log.warn("The following system parameters are empty: "
|
|
| 82 |
+ emptySysParams.stream().map(p -> ((Node) p).valueOf("@name")).collect(Collectors.joining()));
|
|
| 83 |
} else if (paramsNode.selectNodes("//PARAM[@managedBy='user' and @required='true' and not(text())]").size() > 0) {
|
|
| 84 |
((Element) doc.selectSingleNode("//CONFIGURATION")).addAttribute("status", WorkflowStatus.WAIT_USER_SETTINGS.toString());
|
|
| 85 |
} else {
|
|
| 86 |
((Element) doc.selectSingleNode("//CONFIGURATION")).addAttribute("status", WorkflowStatus.EXECUTABLE.toString());
|
|
| 87 |
} |
|
| 87 |
final List<?> emptySysParams = paramsNode.selectNodes("//PARAM[@managedBy='system' and @required='true' and not(text())]");
|
|
| 88 |
if (emptySysParams.size() > 0) {
|
|
| 89 |
((Element) doc.selectSingleNode("//CONFIGURATION")).addAttribute("status", WorkflowStatus.WAIT_SYS_SETTINGS.toString());
|
|
| 90 |
log.warn("The following system parameters are empty: "
|
|
| 91 |
+ emptySysParams.stream().map(p -> ((Node) p).valueOf("@name")).collect(Collectors.joining()));
|
|
| 92 |
} else if (paramsNode.selectNodes("//PARAM[@managedBy='user' and @required='true' and not(text())]").size() > 0) {
|
|
| 93 |
((Element) doc.selectSingleNode("//CONFIGURATION")).addAttribute("status", WorkflowStatus.WAIT_USER_SETTINGS.toString());
|
|
| 94 |
} else {
|
|
| 95 |
((Element) doc.selectSingleNode("//CONFIGURATION")).addAttribute("status", WorkflowStatus.EXECUTABLE.toString());
|
|
| 96 |
}
|
|
| 88 | 97 |
|
| 89 |
wfId = isClient.register(doc.asXML()); |
|
| 98 |
final String wfId = isClient.register(doc.asXML());
|
|
| 90 | 99 |
|
| 91 |
return Arc.DEFAULT_ARC; |
|
| 100 |
token.getEnv().setAttribute("wfId", wfId);
|
|
| 101 |
|
|
| 102 |
token.release(Arc.DEFAULT_ARC); |
|
| 103 |
} catch (final Throwable e) {
|
|
| 104 |
token.releaseAsFailed(e); |
|
| 105 |
} |
|
| 92 | 106 |
} |
| 93 | 107 |
|
| 94 | 108 |
public String getWfName() {
|
| ... | ... | |
| 136 | 150 |
this.process = process; |
| 137 | 151 |
} |
| 138 | 152 |
|
| 139 |
public String getWfId() {
|
|
| 140 |
return wfId; |
|
| 141 |
} |
|
| 142 |
|
|
| 143 |
public void setWfId(final String wfId) {
|
|
| 144 |
this.wfId = wfId; |
|
| 145 |
} |
|
| 146 |
|
|
| 147 | 153 |
} |
Also available in: Unified diff
partial repohi implementation