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  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.so.client.cds;
  22
  23 import java.util.concurrent.CountDownLatch;
  24 import java.util.concurrent.TimeUnit;
  25 import java.util.concurrent.atomic.AtomicReference;
  26
  27 import org.camunda.bpm.engine.delegate.DelegateExecution;
  28 import org.onap.ccsdk.apps.controllerblueprints.common.api.ActionIdentifiers;
  29 import org.onap.ccsdk.apps.controllerblueprints.common.api.CommonHeader;
  30 import org.onap.ccsdk.apps.controllerblueprints.common.api.EventType;
  31 import org.onap.ccsdk.apps.controllerblueprints.processing.api.ExecutionServiceInput;
  32 import org.onap.ccsdk.apps.controllerblueprints.processing.api.ExecutionServiceOutput;
  33 import org.onap.so.client.PreconditionFailedException;
  34 import org.onap.so.client.RestPropertiesLoader;
  35 import org.onap.so.client.cds.beans.AbstractCDSPropertiesBean;
  36 import org.onap.so.client.exception.ExceptionBuilder;
  37 import org.slf4j.Logger;
  38 import org.slf4j.LoggerFactory;
  39 import org.springframework.beans.factory.annotation.Autowired;
  40 import org.springframework.stereotype.Component;
  41
  42 import com.google.protobuf.InvalidProtocolBufferException;
  43 import com.google.protobuf.Struct;
  44 import com.google.protobuf.Struct.Builder;
  45 import com.google.protobuf.util.JsonFormat;
  46
  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     private final AtomicReference<String> cdsResponse = new AtomicReference<>();
  63
  64     @Autowired
  65     private ExceptionBuilder exceptionUtil;
  66
  67     /**
  68      * Extracting data from execution object and building the ExecutionServiceInput
  69      * Object
  70      *
  71      * @param execution
  72      *            DelegateExecution object
  73      */
  74     public void constructExecutionServiceInputObject(DelegateExecution execution) {
  75         logger.trace("Start AbstractCDSProcessingBBUtils.preProcessRequest ");
  76
  77         try {
  78             AbstractCDSPropertiesBean executionObject = (AbstractCDSPropertiesBean) execution
  79                     .getVariable("executionObject");
  80
  81             String payload = executionObject.getRequestObject();
  82
  83             CommonHeader commonHeader = CommonHeader.newBuilder().setOriginatorId(executionObject.getOriginatorId())
  84                     .setRequestId(executionObject.getRequestId()).setSubRequestId(executionObject.getSubRequestId())
  85                     .build();
  86             ActionIdentifiers actionIdentifiers = ActionIdentifiers.newBuilder()
  87                     .setBlueprintName(executionObject.getBlueprintName())
  88                     .setBlueprintVersion(executionObject.getBlueprintVersion())
  89                     .setActionName(executionObject.getActionName()).setMode(executionObject.getMode()).build();
  90
  91             Builder struct = Struct.newBuilder();
  92             try {
  93                 JsonFormat.parser().merge(payload, struct);
  94             } catch (InvalidProtocolBufferException e) {
  95                 logger.error("Failed to parse received message. blueprint({}:{}) for action({}). {}",
  96                         executionObject.getBlueprintVersion(), executionObject.getBlueprintName(),
  97                         executionObject.getActionName(), e);
  98             }
  99
 100             ExecutionServiceInput executionServiceInput = ExecutionServiceInput.newBuilder()
 101                     .setCommonHeader(commonHeader).setActionIdentifiers(actionIdentifiers).setPayload(struct.build())
 102                     .build();
 103
 104             execution.setVariable("executionServiceInput", executionServiceInput);
 105
 106         } catch (Exception ex) {
 107             exceptionUtil.buildAndThrowWorkflowException(execution, 7000, ex);
 108         }
 109     }
 110
 111     /**
 112      * get the executionServiceInput object from execution and send a request to CDS
 113      * Client and wait for TIMEOUT period
 114      *
 115      * @param execution
 116      *            DelegateExecution object
 117      */
 118     public void sendRequestToCDSClient(DelegateExecution execution) {
 119
 120         logger.trace("Start AbstractCDSProcessingBBUtils.sendRequestToCDSClient ");
 121         try {
 122             CDSProperties props = RestPropertiesLoader.getInstance().getNewImpl(CDSProperties.class);
 123             if (props == null) {
 124                 throw new PreconditionFailedException(
 125                         "No RestProperty.CDSProperties implementation found on classpath, can't create client.");
 126             }
 127
 128             ExecutionServiceInput executionServiceInput = (ExecutionServiceInput) execution
 129                     .getVariable("executionServiceInput");
 130
 131             CDSProcessingListener cdsProcessingListener = new AbstractCDSProcessingBBUtils();
 132
 133             CDSProcessingClient cdsClient = new CDSProcessingClient(cdsProcessingListener);
 134             CountDownLatch countDownLatch = cdsClient.sendRequest(executionServiceInput);
 135
 136             try {
 137                 countDownLatch.await(props.getTimeout(), TimeUnit.SECONDS);
 138             } catch (InterruptedException ex) {
 139                 logger.error("Caught exception in sendRequestToCDSClient in AbstractCDSProcessingBBUtils : ", ex);
 140             } finally {
 141                 cdsClient.close();
 142             }
 143
 144             if (cdsResponse != null) {
 145                 execution.setVariable("CDSStatus", cdsResponse.get());
 146             }
 147
 148         } catch (Exception ex) {
 149             exceptionUtil.buildAndThrowWorkflowException(execution, 7000, ex);
 150         }
 151     }
 152
 153     /**
 154      * Get Response from CDS Client
 155      *
 156      */
 157     @Override
 158     public void onMessage(ExecutionServiceOutput message) {
 159         logger.info("Received notification from CDS: {}", message);
 160         EventType eventType = message.getStatus().getEventType();
 161
 162         switch (eventType) {
 163
 164         case EVENT_COMPONENT_FAILURE:
 165             // failed processing with failure
 166             cdsResponse.set(FAILED);
 167             break;
 168         case EVENT_COMPONENT_PROCESSING:
 169             // still processing
 170             cdsResponse.set(PROCESSING);
 171             break;
 172         case EVENT_COMPONENT_EXECUTED:
 173             // done with async processing
 174             cdsResponse.set(SUCCESS);
 175             break;
 176         default:
 177             cdsResponse.set(FAILED);
 178             break;
 179         }
 180
 181     }
 182
 183     /**
 184      * On error at CDS, log the error
 185      */
 186     @Override
 187     public void onError(Throwable t) {
 188         Status status = Status.fromThrowable(t);
 189         logger.error("Failed processing blueprint {}", status, t);
 190     }
 191
 192 }