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 java.util.concurrent.CountDownLatch;
  26 import java.util.concurrent.TimeUnit;
  27 import java.util.concurrent.atomic.AtomicReference;
  28 import org.camunda.bpm.engine.delegate.DelegateExecution;
  29 import org.onap.ccsdk.cds.controllerblueprints.common.api.ActionIdentifiers;
  30 import org.onap.ccsdk.cds.controllerblueprints.common.api.CommonHeader;
  31 import org.onap.ccsdk.cds.controllerblueprints.common.api.EventType;
  32 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput;
  33 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOutput;
  34 import org.onap.so.client.PreconditionFailedException;
  35 import org.onap.so.client.RestPropertiesLoader;
  36 import org.onap.so.client.cds.beans.AbstractCDSPropertiesBean;
  37 import org.onap.so.client.exception.BadResponseException;
  38 import org.onap.so.client.exception.ExceptionBuilder;
  39 import org.slf4j.Logger;
  40 import org.slf4j.LoggerFactory;
  41 import org.springframework.beans.factory.annotation.Autowired;
  42 import org.springframework.stereotype.Component;
  43 import com.google.protobuf.InvalidProtocolBufferException;
  44 import com.google.protobuf.Struct;
  45 import com.google.protobuf.Struct.Builder;
  46 import com.google.protobuf.util.JsonFormat;
  47 import io.grpc.Status;
  48
  49 /**
  50  * Util class to support Call to CDS client
  51  *
  52  */
  53 @Component
  54 public class AbstractCDSProcessingBBUtils implements CDSProcessingListener {
  55
  56     private static final Logger logger = LoggerFactory.getLogger(AbstractCDSProcessingBBUtils.class);
  57
  58     private static final String SUCCESS = "Success";
  59     private static final String FAILED = "Failed";
  60     private static final String PROCESSING = "Processing";
  61
  62     /**
  63      * indicate exception thrown.
  64      */
  65     private static final String EXCEPTION = "Exception";
  66
  67
  68     private final AtomicReference<String> cdsResponse = new AtomicReference<>();
  69
  70     @Autowired
  71     private ExceptionBuilder exceptionUtil;
  72
  73     /**
  74      * Extracting data from execution object and building the ExecutionServiceInput Object
  75      *
  76      * @param execution DelegateExecution object
  77      */
  78     public void constructExecutionServiceInputObject(DelegateExecution execution) {
  79         logger.trace("Start AbstractCDSProcessingBBUtils.preProcessRequest ");
  80
  81         try {
  82             AbstractCDSPropertiesBean executionObject =
  83                     (AbstractCDSPropertiesBean) execution.getVariable("executionObject");
  84
  85             String payload = executionObject.getRequestObject();
  86
  87             CommonHeader commonHeader = CommonHeader.newBuilder().setOriginatorId(executionObject.getOriginatorId())
  88                     .setRequestId(executionObject.getRequestId()).setSubRequestId(executionObject.getSubRequestId())
  89                     .build();
  90             ActionIdentifiers actionIdentifiers =
  91                     ActionIdentifiers.newBuilder().setBlueprintName(executionObject.getBlueprintName())
  92                             .setBlueprintVersion(executionObject.getBlueprintVersion())
  93                             .setActionName(executionObject.getActionName()).setMode(executionObject.getMode()).build();
  94
  95             Builder struct = Struct.newBuilder();
  96             try {
  97                 JsonFormat.parser().merge(payload, struct);
  98             } catch (InvalidProtocolBufferException e) {
  99                 logger.error("Failed to parse received message. blueprint({}:{}) for action({}). {}",
 100                         executionObject.getBlueprintVersion(), executionObject.getBlueprintName(),
 101                         executionObject.getActionName(), e);
 102             }
 103
 104             ExecutionServiceInput executionServiceInput =
 105                     ExecutionServiceInput.newBuilder().setCommonHeader(commonHeader)
 106                             .setActionIdentifiers(actionIdentifiers).setPayload(struct.build()).build();
 107
 108             execution.setVariable("executionServiceInput", executionServiceInput);
 109
 110         } catch (Exception ex) {
 111             exceptionUtil.buildAndThrowWorkflowException(execution, 7000, ex);
 112         }
 113     }
 114
 115     /**
 116      * get the executionServiceInput object from execution and send a request to CDS Client and wait for TIMEOUT period
 117      *
 118      * @param execution DelegateExecution object
 119      */
 120     public void sendRequestToCDSClient(DelegateExecution execution) {
 121
 122         logger.trace("Start AbstractCDSProcessingBBUtils.sendRequestToCDSClient ");
 123         try {
 124             CDSProperties props = RestPropertiesLoader.getInstance().getNewImpl(CDSProperties.class);
 125             if (props == null) {
 126                 throw new PreconditionFailedException(
 127                         "No RestProperty.CDSProperties implementation found on classpath, can't create client.");
 128             }
 129
 130             ExecutionServiceInput executionServiceInput =
 131                     (ExecutionServiceInput) execution.getVariable("executionServiceInput");
 132
 133             try (CDSProcessingClient cdsClient = new CDSProcessingClient(this)) {
 134                 CountDownLatch countDownLatch = cdsClient.sendRequest(executionServiceInput);
 135                 countDownLatch.await(props.getTimeout(), TimeUnit.SECONDS);
 136             } catch (InterruptedException ex) {
 137                 logger.error("Caught exception in sendRequestToCDSClient in AbstractCDSProcessingBBUtils : ", ex);
 138                 Thread.currentThread().interrupt();
 139             }
 140
 141             if (cdsResponse != null) {
 142                 String cdsResponseStatus = cdsResponse.get();
 143                 execution.setVariable("CDSStatus", cdsResponseStatus);
 144
 145                 /**
 146                  * throw CDS failed exception.
 147                  */
 148                 if (cdsResponseStatus != SUCCESS) {
 149                     throw new BadResponseException("CDS call failed with status: " + cdsResponseStatus);
 150                 }
 151             }
 152
 153         } catch (Exception ex) {
 154             exceptionUtil.buildAndThrowWorkflowException(execution, 7000, ex);
 155         }
 156     }
 157
 158     /**
 159      * Get Response from CDS Client
 160      *
 161      */
 162     @Override
 163     public void onMessage(ExecutionServiceOutput message) {
 164         logger.info("Received notification from CDS: {}", message);
 165         EventType eventType = message.getStatus().getEventType();
 166
 167         switch (eventType) {
 168
 169             case EVENT_COMPONENT_FAILURE:
 170                 // failed processing with failure
 171                 cdsResponse.set(FAILED);
 172                 break;
 173             case EVENT_COMPONENT_PROCESSING:
 174                 // still processing
 175                 cdsResponse.set(PROCESSING);
 176                 break;
 177             case EVENT_COMPONENT_EXECUTED:
 178                 // done with async processing
 179                 cdsResponse.set(SUCCESS);
 180                 break;
 181             default:
 182                 cdsResponse.set(FAILED);
 183                 break;
 184         }
 185
 186     }
 187
 188     /**
 189      * On error at CDS, log the error
 190      */
 191     @Override
 192     public void onError(Throwable t) {
 193         Status status = Status.fromThrowable(t);
 194         logger.error("Failed processing blueprint {}", status, t);
 195         cdsResponse.set(EXCEPTION);
 196     }
 197
 198 }