* ============LICENSE_START=======================================================
* ONAP : APPC
* ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved.
* =============================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
public class FlowControlNode implements SvcLogicJavaPlugin {
- private static final EELFLogger log = EELFManager.getInstance().getLogger(FlowControlNode.class);
-
- private final FlowControlDBService dbService;
- private final FlowSequenceGenerator flowSequenceGenerator;
-
- public FlowControlNode() {
- this.dbService = FlowControlDBService.initialise();
- this.flowSequenceGenerator = new FlowSequenceGenerator();
- }
-
- FlowControlNode(FlowControlDBService dbService, FlowSequenceGenerator flowSequenceGenerator) {
- this.dbService = dbService;
- this.flowSequenceGenerator = flowSequenceGenerator;
- }
-
- public void processFlow(Map<String, String> inParams, SvcLogicContext ctx)
- throws SvcLogicException {
- log.debug("Received processParamKeys call with params : " + inParams);
- String responsePrefix = inParams.get(INPUT_PARAM_RESPONSE_PREFIX);
- try {
- responsePrefix = StringUtils.isNotBlank(responsePrefix) ? (responsePrefix + ".") : "";
- SvcLogicContext localContext = new SvcLogicContext();
-
- localContext.setAttribute(REQUEST_ID, ctx.getAttribute(REQUEST_ID));
- localContext.setAttribute(VNF_TYPE, ctx.getAttribute(VNF_TYPE));
- localContext.setAttribute(REQUEST_ACTION, ctx.getAttribute(REQUEST_ACTION));
- localContext.setAttribute(ACTION_LEVEL, ctx.getAttribute(ACTION_LEVEL));
- localContext.setAttribute(RESPONSE_PREFIX, responsePrefix);
- ctx.setAttribute(RESPONSE_PREFIX, responsePrefix);
-
- dbService.getFlowReferenceData(ctx, inParams, localContext);
-
- for (String key : localContext.getAttributeKeySet()) {
- log.debug("processFlow " + key + "=" + ctx.getAttribute(key));
- }
- processFlowSequence(inParams, ctx, localContext);
- if (!ctx.getAttribute(responsePrefix + OUTPUT_PARAM_STATUS).equals(OUTPUT_STATUS_SUCCESS)) {
- throw new SvcLogicException(ctx.getAttribute(responsePrefix + OUTPUT_STATUS_MESSAGE));
- }
- } catch (Exception e) {
- ctx.setAttribute(responsePrefix + OUTPUT_PARAM_STATUS, OUTPUT_STATUS_FAILURE);
- ctx.setAttribute(responsePrefix + OUTPUT_PARAM_ERROR_MESSAGE, e.getMessage());
- log.error("Error occurred in processFlow ", e);
- throw new SvcLogicException(e.getMessage());
+ private static final EELFLogger log = EELFManager.getInstance().getLogger(FlowControlNode.class);
+
+ private final FlowControlDBService dbService;
+ private final FlowSequenceGenerator flowSequenceGenerator;
+
+ public FlowControlNode() {
+ this.dbService = FlowControlDBService.initialise();
+ this.flowSequenceGenerator = new FlowSequenceGenerator();
}
- }
-
- private void processFlowSequence(Map<String, String> inParams, SvcLogicContext ctx, SvcLogicContext localContext)
- throws Exception {
-
- String fn = "FlowExecutorNode.processflowSequence";
- log.debug(fn + "Received model for flow : " + localContext.toString());
-
- localContext
- .getAttributeKeySet()
- .forEach(key -> log.debug(key + "=" + ctx.getAttribute(key)));
-
- String flowSequence = flowSequenceGenerator.getFlowSequence(inParams, ctx, localContext);
-
- log.debug("Received Flow Sequence : " + flowSequence);
- HashMap<Integer, Transaction> transactionMap = createTransactionMap(flowSequence, localContext);
- executeAllTransaction(transactionMap, ctx);
- log.info("Executed all the transaction successfully");
- }
-
- private void executeAllTransaction(HashMap<Integer, Transaction> transactionMap, SvcLogicContext ctx)
- throws Exception {
-
- String fn = "FlowExecutorNode.executeAllTransaction ";
- int retry = 0;
- FlowExecutorInterface flowExecutor;
- for (int key = 1; key <= transactionMap.size(); key++) {
- log.debug(fn + "Starting transactions ID " + key + " :)=" + retry);
- Transaction transaction = transactionMap.get(key);
- if (!preProcessor(transactionMap, transaction)) {
- log.info("Skipping Transaction ID " + transaction.getTransactionId());
- continue;
- }
- if (transaction.getExecutionType() != null) {
- switch (transaction.getExecutionType()) {
- case GRAPH:
- flowExecutor = new GraphExecutor();
- break;
- case NODE:
- flowExecutor = new NodeExecutor();
- break;
- case REST:
- flowExecutor = new RestExecutor();
- break;
- default:
- throw new Exception("No Executor found for transaction ID" + transaction.getTransactionId());
- }
- flowExecutor.execute(transaction, ctx);
- ResponseAction responseAction = handleResponse(transaction);
- if (responseAction.getWait() != null && Integer.parseInt(responseAction.getWait()) > 0) {
- log.debug(fn + "Going to Sleep .... " + responseAction.getWait());
- Thread.sleep(Integer.parseInt(responseAction.getWait()) * 1000L);
- }
- if (responseAction.isIntermediateMessage()) {
- log.debug(fn + "Sending Intermediate Message back .... ");
- sendIntermediateMessage();
- }
- if (responseAction.getRetry() != null && Integer.parseInt(responseAction.getRetry()) > retry) {
- log.debug(fn + "Ooppss!!! We will retry again ....... ");
- key--;
- retry++;
- log.debug(fn + "key =" + key + "retry =" + retry);
- }
- if (responseAction.isIgnore()) {
- log.debug(fn + "Ignoring this Error and moving ahead ....... ");
- continue;
- }
- if (responseAction.isStop()) {
- log.debug(fn + "Need to Stop ....... ");
- break;
+ FlowControlNode(FlowControlDBService dbService, FlowSequenceGenerator flowSequenceGenerator) {
+ this.dbService = dbService;
+ this.flowSequenceGenerator = flowSequenceGenerator;
+ }
+
+ public void processFlow(Map<String, String> inParams, SvcLogicContext ctx) throws SvcLogicException {
+ log.debug("Received processParamKeys call with params : " + inParams);
+ String responsePrefix = inParams.get(INPUT_PARAM_RESPONSE_PREFIX);
+ try {
+ responsePrefix = StringUtils.isNotBlank(responsePrefix) ? (responsePrefix + ".") : "";
+ SvcLogicContext localContext = new SvcLogicContext();
+
+ localContext.setAttribute(REQUEST_ID, ctx.getAttribute(REQUEST_ID));
+ localContext.setAttribute(VNF_TYPE, ctx.getAttribute(VNF_TYPE));
+ localContext.setAttribute(REQUEST_ACTION, ctx.getAttribute(REQUEST_ACTION));
+ localContext.setAttribute(ACTION_LEVEL, ctx.getAttribute(ACTION_LEVEL));
+ localContext.setAttribute(RESPONSE_PREFIX, responsePrefix);
+ ctx.setAttribute(RESPONSE_PREFIX, responsePrefix);
+
+ dbService.getFlowReferenceData(ctx, inParams, localContext);
+
+ for (String key : localContext.getAttributeKeySet()) {
+ log.debug("processFlow " + key + "=" + ctx.getAttribute(key));
+ }
+ processFlowSequence(inParams, ctx, localContext);
+ if (!ctx.getAttribute(responsePrefix + OUTPUT_PARAM_STATUS).equals(OUTPUT_STATUS_SUCCESS)) {
+ throw new SvcLogicException(ctx.getAttribute(responsePrefix + OUTPUT_STATUS_MESSAGE));
+ }
+ } catch (Exception e) {
+ ctx.setAttribute(responsePrefix + OUTPUT_PARAM_STATUS, OUTPUT_STATUS_FAILURE);
+ ctx.setAttribute(responsePrefix + OUTPUT_PARAM_ERROR_MESSAGE, e.getMessage());
+ log.error("Error occurred in processFlow ", e);
+ throw new SvcLogicException(e.getMessage());
}
- if (responseAction.getJump() != null && Integer.parseInt(responseAction.getJump()) > 0) {
- key = Integer.parseInt(responseAction.getJump());
- key--;
+ }
+
+ private void processFlowSequence(Map<String, String> inParams, SvcLogicContext ctx, SvcLogicContext localContext)
+ throws Exception {
+
+ String fn = "FlowExecutorNode.processflowSequence";
+ log.debug(fn + "Received model for flow : " + localContext.toString());
+
+ localContext.getAttributeKeySet().forEach(key -> log.debug(key + "=" + ctx.getAttribute(key)));
+
+ String flowSequence = flowSequenceGenerator.getFlowSequence(inParams, ctx, localContext);
+
+ log.debug("Received Flow Sequence : " + flowSequence);
+ HashMap<Integer, Transaction> transactionMap = createTransactionMap(flowSequence, localContext);
+ executeAllTransaction(transactionMap, ctx);
+ log.info("Executed all the transaction successfully");
+ }
+
+ private void executeAllTransaction(HashMap<Integer, Transaction> transactionMap, SvcLogicContext ctx)
+ throws Exception {
+
+ String fn = "FlowExecutorNode.executeAllTransaction ";
+ int retry = 0;
+ FlowExecutorInterface flowExecutor;
+ for (int key = 1; key <= transactionMap.size(); key++) {
+ log.debug(fn + "Starting transactions ID " + key + " :)=" + retry);
+ Transaction transaction = transactionMap.get(key);
+ if (!preProcessor(transactionMap, transaction)) {
+ log.info("Skipping Transaction ID " + transaction.getTransactionId());
+ continue;
+ }
+ if (transaction.getExecutionType() != null) {
+ switch (transaction.getExecutionType()) {
+ case GRAPH:
+ flowExecutor = new GraphExecutor();
+ break;
+ case NODE:
+ flowExecutor = new NodeExecutor();
+ break;
+ case REST:
+ flowExecutor = new RestExecutor();
+ break;
+ default:
+ throw new Exception("No Executor found for transaction ID" + transaction.getTransactionId());
+ }
+ flowExecutor.execute(transaction, ctx);
+ ResponseAction responseAction = handleResponse(transaction, ctx);
+
+ if (responseAction.getWait() != null && Integer.parseInt(responseAction.getWait()) > 0) {
+ log.debug(fn + "Going to Sleep .... " + responseAction.getWait());
+ Thread.sleep(Integer.parseInt(responseAction.getWait()) * 1000L);
+ }
+ if (responseAction.isIntermediateMessage()) {
+ log.debug(fn + "Sending Intermediate Message back .... ");
+ sendIntermediateMessage();
+ }
+ if (responseAction.getRetry() != null && Integer.parseInt(responseAction.getRetry()) > retry) {
+ log.debug(fn + "Ooppss!!! We will retry again ....... ");
+ key--;
+ retry++;
+ log.debug(fn + "key =" + key + "retry =" + retry);
+ }
+ if (responseAction.isIgnore()) {
+ log.debug(fn + "Ignoring this Error and moving ahead ....... ");
+ continue;
+ }
+ if (responseAction.isStop()) {
+ log.debug(fn + "Need to Stop ....... ");
+ break;
+ }
+ if (responseAction.getJump() != null && Integer.parseInt(responseAction.getJump()) > 0) {
+ key = Integer.parseInt(responseAction.getJump());
+ key--;
+ }
+ log.debug(fn + "key =" + key + "retry =" + retry);
+
+ } else {
+ throw new Exception("Don't know how to execute transaction ID " + transaction.getTransactionId());
+ }
}
- log.debug(fn + "key =" + key + "retry =" + retry);
+ }
- } else {
- throw new Exception("Don't know how to execute transaction ID " + transaction.getTransactionId());
- }
+ private void sendIntermediateMessage() {
+ // TODO Auto-generated method stub
}
- }
-
- private void sendIntermediateMessage() {
- // TODO Auto-generated method stub
- }
-
- private ResponseAction handleResponse(Transaction transaction) {
- log.info("Handling Response for transaction Id " + transaction.getTransactionId());
- DefaultResponseHandler defaultHandler = new DefaultResponseHandler();
- return defaultHandler.handlerResponse(transaction);
- }
-
- private boolean preProcessor(HashMap<Integer, Transaction> transactionMap, Transaction transaction)
- throws IOException {
-
- log.debug("Starting Preprocessing Logic ");
- boolean runThisStep = false;
- try {
- if (transaction.getPrecheck() != null
- && transaction.getPrecheck().getPrecheckOptions() != null
- && !transaction.getPrecheck().getPrecheckOptions().isEmpty()) {
-
- List<PrecheckOption> precheckOptions = transaction.getPrecheck().getPrecheckOptions();
- for (PrecheckOption precheck : precheckOptions) {
- Transaction trans = transactionMap.get(precheck.getpTransactionID());
- ObjectMapper mapper = new ObjectMapper();
- log.info("Mapper= " + mapper.writeValueAsString(trans));
- HashMap trmap = mapper.readValue(mapper.writeValueAsString(trans), HashMap.class);
- runThisStep = trmap.get(precheck.getParamName()) != null
- && ((String) trmap.get(precheck.getParamName()))
- .equalsIgnoreCase(precheck.getParamValue());
-
- if (("any").equalsIgnoreCase(transaction.getPrecheck().getPrecheckOperator()) && runThisStep) {
- break;
- }
+
+ private ResponseAction handleResponse(Transaction transaction, SvcLogicContext ctx) {
+ log.info("Handling Response for transaction Id " + transaction.getTransactionId());
+ DefaultResponseHandler defaultHandler = new DefaultResponseHandler();
+ return defaultHandler.handlerResponse(transaction, ctx);
+ }
+
+ private boolean preProcessor(HashMap<Integer, Transaction> transactionMap, Transaction transaction)
+ throws IOException {
+
+ log.debug("Starting Preprocessing Logic ");
+ boolean runThisStep = false;
+ try {
+ if (transaction.getPrecheck() != null && transaction.getPrecheck().getPrecheckOptions() != null
+ && !transaction.getPrecheck().getPrecheckOptions().isEmpty()) {
+
+ List<PrecheckOption> precheckOptions = transaction.getPrecheck().getPrecheckOptions();
+ for (PrecheckOption precheck : precheckOptions) {
+ Transaction trans = transactionMap.get(precheck.getpTransactionID());
+ ObjectMapper mapper = new ObjectMapper();
+ log.info("Mapper= " + mapper.writeValueAsString(trans));
+ HashMap trmap = mapper.readValue(mapper.writeValueAsString(trans), HashMap.class);
+ runThisStep = trmap.get(precheck.getParamName()) != null
+ && ((String) trmap.get(precheck.getParamName())).equalsIgnoreCase(precheck.getParamValue());
+
+ if (("any").equalsIgnoreCase(transaction.getPrecheck().getPrecheckOperator()) && runThisStep) {
+ break;
+ }
+ }
+ } else {
+ log.debug("No Pre check defined for transaction ID " + transaction.getTransactionId());
+ runThisStep = true;
+ }
+ } catch (Exception e) {
+ log.error("Error occured when Preprocessing Logic ", e);
+ throw e;
}
- } else {
- log.debug("No Pre check defined for transaction ID " + transaction.getTransactionId());
- runThisStep = true;
- }
- } catch (Exception e) {
- log.error("Error occured when Preprocessing Logic ", e);
- throw e;
+ log.debug("Returing process current Transaction = " + runThisStep);
+ return runThisStep;
}
- log.debug("Returing process current Transaction = " + runThisStep);
- return runThisStep;
- }
-
- private HashMap<Integer, Transaction> createTransactionMap(String flowSequence, SvcLogicContext localContext)
- throws Exception {
-
- ObjectMapper mapper = new ObjectMapper();
- Transactions transactions = mapper.readValue(flowSequence, Transactions.class);
- HashMap<Integer, Transaction> transMap = new HashMap<>();
- for (Transaction transaction : transactions.getTransactions()) {
- compileFlowDependencies(transaction, localContext);
- //parse the Transactions Object and create records in process_flow_status table
- //loadTransactionIntoStatus(transactions, ctx);
- transMap.put(transaction.getTransactionId(), transaction);
+
+ private HashMap<Integer, Transaction> createTransactionMap(String flowSequence, SvcLogicContext localContext)
+ throws Exception {
+
+ ObjectMapper mapper = new ObjectMapper();
+ Transactions transactions = mapper.readValue(flowSequence, Transactions.class);
+ HashMap<Integer, Transaction> transMap = new HashMap<>();
+ for (Transaction transaction : transactions.getTransactions()) {
+ compileFlowDependencies(transaction, localContext);
+ // parse the Transactions Object and create records in process_flow_status table
+ // loadTransactionIntoStatus(transactions, ctx);
+ transMap.put(transaction.getTransactionId(), transaction);
+ }
+ return transMap;
}
- return transMap;
- }
- private void compileFlowDependencies(Transaction transaction, SvcLogicContext localContext)
- throws Exception {
+ private void compileFlowDependencies(Transaction transaction, SvcLogicContext localContext) throws Exception {
- dbService.populateModuleAndRPC(transaction, localContext.getAttribute(VNF_TYPE));
- ObjectMapper mapper = new ObjectMapper();
- log.debug("Individual Transaction Details :" + transaction.toString());
+ dbService.populateModuleAndRPC(transaction, localContext.getAttribute(VNF_TYPE));
+ ObjectMapper mapper = new ObjectMapper();
+ log.debug("Individual Transaction Details :" + transaction.toString());
- if ((localContext.getAttribute(SEQUENCE_TYPE) == null)
- || (localContext.getAttribute(SEQUENCE_TYPE) != null
- && !localContext.getAttribute(SEQUENCE_TYPE)
- .equalsIgnoreCase(DESINGTIME))) {
+ if ((localContext.getAttribute(SEQUENCE_TYPE) == null) || (localContext.getAttribute(SEQUENCE_TYPE) != null
+ && !localContext.getAttribute(SEQUENCE_TYPE).equalsIgnoreCase(DESINGTIME))) {
- localContext.setAttribute("artifact-content", mapper.writeValueAsString(transaction));
- dbService.loadSequenceIntoDB(localContext);
+ localContext.setAttribute("artifact-content", mapper.writeValueAsString(transaction));
+ dbService.loadSequenceIntoDB(localContext);
+ }
+ // get a field in transction class as transactionhandle interface and register
+ // the Handler here for each transactions
}
- //get a field in transction class as transactionhandle interface and register the Handler here for each trnactions
- }
}
Code Review / appc.git / blobdiff