bpmn/MSOCommonBPMN/src/main/java/org/onap/so/client/cds/AbstractCDSProcessingBBUtils.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP - SO
   4  * ================================================================================
   5  * Copyright (C) 2019 TechMahindra
   6  * ================================================================================
   7  * Modifications Copyright (c) 2019 Samsung
   8  * ================================================================================
   9  * Licensed under the Apache License, Version 2.0 (the "License");
  10  * you may not use this file except in compliance with the License.
  11  * You may obtain a copy of the License at
  12  *
  13  *      http://www.apache.org/licenses/LICENSE-2.0
  14  *
  15  * Unless required by applicable law or agreed to in writing, software
  16  * distributed under the License is distributed on an "AS IS" BASIS,
  17  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18  * See the License for the specific language governing permissions and
  19  * limitations under the License.
  20  * ============LICENSE_END=========================================================
  21  */
  22
  23 package org.onap.so.client.cds;
  24
  25 import com.google.protobuf.InvalidProtocolBufferException;
  26 import com.google.protobuf.Struct;
  27 import com.google.protobuf.Struct.Builder;
  28 import com.google.protobuf.util.JsonFormat;
  29 import io.grpc.Status;
  30 import org.camunda.bpm.engine.delegate.DelegateExecution;
  31 import org.onap.ccsdk.cds.controllerblueprints.common.api.ActionIdentifiers;
  32 import org.onap.ccsdk.cds.controllerblueprints.common.api.CommonHeader;
  33 import org.onap.ccsdk.cds.controllerblueprints.common.api.EventType;
  34 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput;
  35 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOutput;
  36 import org.onap.so.bpmn.common.BuildingBlockExecution;
  37 import org.onap.so.client.PreconditionFailedException;
  38 import org.onap.so.client.RestPropertiesLoader;
  39 import org.onap.so.client.cds.beans.AbstractCDSPropertiesBean;
  40 import org.onap.so.client.exception.BadResponseException;
  41 import org.onap.so.client.exception.ExceptionBuilder;
  42 import org.slf4j.Logger;
  43 import org.slf4j.LoggerFactory;
  44 import org.springframework.beans.factory.annotation.Autowired;
  45 import org.springframework.stereotype.Component;
  46 import java.util.concurrent.CountDownLatch;
  47 import java.util.concurrent.TimeUnit;
  48
  49 /**
  50  * Util class to support Call to CDS client
  51  */
  52 @Component
  53 public class AbstractCDSProcessingBBUtils {
  54
  55     private static final Logger logger = LoggerFactory.getLogger(AbstractCDSProcessingBBUtils.class);
  56
  57     private static final String SUCCESS = "Success";
  58     private static final String FAILED = "Failed";
  59     private static final String PROCESSING = "Processing";
  60     private static final String RESPONSE_PAYLOAD = "CDSResponsePayload";
  61     private static final String CDS_STATUS = "ControllerStatus";
  62     private static final String EXEC_INPUT = "executionServiceInput";
  63     private static final String EXECUTION_OBJECT = "executionObject";
  64     private static final String EXCEPTION = "Exception";
  65
  66     @Autowired
  67     protected ExceptionBuilder exceptionUtil;
  68
  69     /**
  70      * Extracting data from execution object and building the ExecutionServiceInput Object
  71      *
  72      * @param execution DelegateExecution object
  73      */
  74     public void constructExecutionServiceInputObject(DelegateExecution execution) {
  75         logger.trace("Start AbstractCDSProcessingBBUtils.preProcessRequest for DelegateExecution object.");
  76
  77         try {
  78             AbstractCDSPropertiesBean executionObject =
  79                     (AbstractCDSPropertiesBean) execution.getVariable(EXECUTION_OBJECT);
  80
  81             ExecutionServiceInput executionServiceInput = prepareExecutionServiceInput(executionObject);
  82
  83             execution.setVariable(EXEC_INPUT, executionServiceInput);
  84
  85         } catch (Exception ex) {
  86             exceptionUtil.buildAndThrowWorkflowException(execution, 7000, ex);
  87         }
  88     }
  89
  90     /**
  91      * Extracting data from execution object and building the ExecutionServiceInput Object
  92      *
  93      * @param execution BuildingBlockExecution object
  94      */
  95     public void constructExecutionServiceInputObject(BuildingBlockExecution execution) {
  96         logger.trace("Start AbstractCDSProcessingBBUtils.preProcessRequest for BuildingBlockExecution object.");
  97
  98         try {
  99             AbstractCDSPropertiesBean executionObject = execution.getVariable(EXECUTION_OBJECT);
 100
 101             ExecutionServiceInput executionServiceInput = prepareExecutionServiceInput(executionObject);
 102
 103             execution.setVariable(EXEC_INPUT, executionServiceInput);
 104
 105         } catch (Exception ex) {
 106             exceptionUtil.buildAndThrowWorkflowException(execution, 7000, ex);
 107         }
 108     }
 109
 110     /**
 111      * get the executionServiceInput object from execution and send a request to CDS Client and wait for TIMEOUT period
 112      *
 113      * @param execution DelegateExecution object
 114      */
 115     public void sendRequestToCDSClient(DelegateExecution execution) {
 116
 117         logger.trace("Start AbstractCDSProcessingBBUtils.sendRequestToCDSClient for DelegateExecution object.");
 118         try {
 119             ExecutionServiceInput executionServiceInput = (ExecutionServiceInput) execution.getVariable(EXEC_INPUT);
 120             CDSResponse cdsResponse = getCdsResponse(executionServiceInput);
 121             execution.setVariable(CDS_STATUS, cdsResponse.status);
 122
 123             if (cdsResponse.payload != null) {
 124                 String payload = JsonFormat.printer().print(cdsResponse.payload);
 125                 execution.setVariable(RESPONSE_PAYLOAD, payload);
 126             }
 127
 128         } catch (Exception ex) {
 129             exceptionUtil.buildAndThrowWorkflowException(execution, 7000, ex);
 130         }
 131     }
 132
 133     /**
 134      * get the executionServiceInput object from execution and send a request to CDS Client and wait for TIMEOUT period
 135      *
 136      * @param execution BuildingBlockExecution object
 137      */
 138     public void sendRequestToCDSClient(BuildingBlockExecution execution) {
 139
 140         logger.trace("Start AbstractCDSProcessingBBUtils.sendRequestToCDSClient for BuildingBlockExecution object.");
 141         try {
 142             ExecutionServiceInput executionServiceInput = execution.getVariable(EXEC_INPUT);
 143             CDSResponse cdsResponse = getCdsResponse(executionServiceInput);
 144             execution.setVariable(CDS_STATUS, cdsResponse.status);
 145
 146             if (cdsResponse.payload != null) {
 147                 String payload = JsonFormat.printer().print(cdsResponse.payload);
 148                 execution.setVariable(RESPONSE_PAYLOAD, payload);
 149             }
 150
 151         } catch (Exception ex) {
 152             exceptionUtil.buildAndThrowWorkflowException(execution, 7000, ex);
 153         }
 154     }
 155
 156     private CDSResponse getCdsResponse(ExecutionServiceInput executionServiceInput) throws BadResponseException {
 157         CDSProperties props = RestPropertiesLoader.getInstance().getNewImpl(CDSProperties.class);
 158         if (props == null) {
 159             throw new PreconditionFailedException(
 160                     "No RestProperty.CDSProperties implementation found on classpath, can't create client.");
 161         }
 162
 163         CDSResponse cdsResponse = new CDSResponse();
 164
 165         try (CDSProcessingClient cdsClient = new CDSProcessingClient(new ResponseHandler(cdsResponse))) {
 166             CountDownLatch countDownLatch = cdsClient.sendRequest(executionServiceInput);
 167             countDownLatch.await(props.getTimeout(), TimeUnit.SECONDS);
 168         } catch (InterruptedException ex) {
 169             logger.error("Caught exception in sendRequestToCDSClient in AbstractCDSProcessingBBUtils : ", ex);
 170             Thread.currentThread().interrupt();
 171         }
 172
 173         String cdsResponseStatus = cdsResponse.status;
 174
 175         /**
 176          * throw CDS failed exception.
 177          */
 178         if (!cdsResponseStatus.equals(SUCCESS)) {
 179             throw new BadResponseException("CDS call failed with status: " + cdsResponse.status + " and errorMessage: "
 180                     + cdsResponse.errorMessage);
 181         }
 182         return cdsResponse;
 183     }
 184
 185     private ExecutionServiceInput prepareExecutionServiceInput(AbstractCDSPropertiesBean executionObject) {
 186         String payload = executionObject.getRequestObject();
 187
 188         CommonHeader commonHeader = CommonHeader.newBuilder().setOriginatorId(executionObject.getOriginatorId())
 189                 .setRequestId(executionObject.getRequestId()).setSubRequestId(executionObject.getSubRequestId())
 190                 .build();
 191         ActionIdentifiers actionIdentifiers =
 192                 ActionIdentifiers.newBuilder().setBlueprintName(executionObject.getBlueprintName())
 193                         .setBlueprintVersion(executionObject.getBlueprintVersion())
 194                         .setActionName(executionObject.getActionName()).setMode(executionObject.getMode()).build();
 195
 196         Builder struct = Struct.newBuilder();
 197         try {
 198             JsonFormat.parser().merge(payload, struct);
 199         } catch (InvalidProtocolBufferException e) {
 200             logger.error("Failed to parse received message. blueprint({}:{}) for action({}). {}",
 201                     executionObject.getBlueprintVersion(), executionObject.getBlueprintName(),
 202                     executionObject.getActionName(), e);
 203         }
 204
 205         return ExecutionServiceInput.newBuilder().setCommonHeader(commonHeader).setActionIdentifiers(actionIdentifiers)
 206                 .setPayload(struct.build()).build();
 207     }
 208
 209     private class ResponseHandler implements CDSProcessingListener {
 210
 211         private CDSResponse cdsResponse;
 212
 213         ResponseHandler(CDSResponse cdsResponse) {
 214             this.cdsResponse = cdsResponse;
 215         }
 216
 217         /**
 218          * Get Response from CDS Client
 219          */
 220         @Override
 221         public void onMessage(ExecutionServiceOutput message) {
 222             logger.info("Received notification from CDS: {}", message);
 223             EventType eventType = message.getStatus().getEventType();
 224
 225             switch (eventType) {
 226                 case EVENT_COMPONENT_PROCESSING:
 227                     cdsResponse.status = PROCESSING;
 228                     break;
 229                 case EVENT_COMPONENT_EXECUTED:
 230                     cdsResponse.status = SUCCESS;
 231                     break;
 232                 default:
 233                     cdsResponse.status = FAILED;
 234                     cdsResponse.errorMessage = message.getStatus().getErrorMessage();
 235                     break;
 236             }
 237             cdsResponse.payload = message.getPayload();
 238         }
 239
 240         /**
 241          * On error at CDS, log the error
 242          */
 243         @Override
 244         public void onError(Throwable t) {
 245             Status status = Status.fromThrowable(t);
 246             logger.error("Failed processing blueprint {}", status, t);
 247             cdsResponse.status = EXCEPTION;
 248         }
 249     }
 250
 251     private class CDSResponse {
 252
 253         String status;
 254         String errorMessage;
 255         Struct payload;
 256
 257         @Override
 258         public String toString() {
 259             return "CDSResponse{" + "status='" + status + '\'' + ", errorMessage='" + errorMessage + '\'' + ", payload="
 260                     + payload + '}';
 261         }
 262     }
 263 }