models-interactions/model-impl/cds/src/main/java/org/onap/policy/cds/client/CdsProcessorGrpcClient.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * Copyright (C) 2019-2021 Bell Canada.
   4  * Modifications Copyright (C) 2020-2021 AT&T Intellectual Property. All rights reserved.
   5  * ================================================================================
   6  * Licensed under the Apache License, Version 2.0 (the "License");
   7  * you may not use this file except in compliance with the License.
   8  * You may obtain a copy of the License at
   9  *
  10  *      http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  * ============LICENSE_END=========================================================
  18  */
  19
  20 package org.onap.policy.cds.client;
  21
  22 import com.google.common.base.Preconditions;
  23 import io.grpc.ManagedChannel;
  24 import io.grpc.netty.NettyChannelBuilder;
  25 import java.util.concurrent.CountDownLatch;
  26 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput;
  27 import org.onap.policy.cds.api.CdsProcessorListener;
  28 import org.onap.policy.cds.properties.CdsServerProperties;
  29 import org.onap.policy.common.parameters.ValidationResult;
  30 import org.slf4j.Logger;
  31 import org.slf4j.LoggerFactory;
  32
  33 /**
  34  * <p>
  35  * The CDS processor client uses gRPC for communication between Policy and CDS. This communication is configured to use
  36  * a streaming approach, which means the client sends an event to which the server can reply with multiple
  37  * sub-responses, until full completion of the processing.
  38  * </p>
  39  */
  40 public class CdsProcessorGrpcClient implements AutoCloseable {
  41
  42     private static final Logger LOGGER = LoggerFactory.getLogger(CdsProcessorGrpcClient.class);
  43
  44     private ManagedChannel channel;
  45     private CdsProcessorHandler handler;
  46
  47     /**
  48      * Constructor, create a CDS processor gRPC client.
  49      *
  50      * @param listener the listener to listen on
  51      */
  52     public CdsProcessorGrpcClient(final CdsProcessorListener listener, CdsServerProperties props) {
  53         final ValidationResult validationResult = props.validate();
  54         Preconditions.checkState(validationResult.getStatus().isValid(), "Error validating CDS server "
  55             + "properties: " + validationResult.getResult());
  56
  57         StringBuilder bldr = new StringBuilder("gRPC://");
  58         bldr.append(props.getHost());
  59         bldr.append(":");
  60         bldr.append(props.getPort());
  61         bldr.append('/');
  62
  63         String url = bldr.toString();
  64
  65         this.channel = NettyChannelBuilder.forAddress(props.getHost(), props.getPort())
  66             .intercept(new BasicAuthClientHeaderInterceptor(props)).usePlaintext().build();
  67         this.handler = new CdsProcessorHandler(listener, url);
  68         LOGGER.info("CdsProcessorListener started");
  69     }
  70
  71     CdsProcessorGrpcClient(final ManagedChannel channel, final CdsProcessorHandler handler) {
  72         this.channel = channel;
  73         this.handler = handler;
  74     }
  75
  76     /**
  77      * Sends a request to the CDS backend micro-service.
  78      *
  79      * <p>The caller will be returned a CountDownLatch that can be used to define how long the processing can wait. The
  80      * CountDownLatch is initiated with just 1 count. When the client receives an #onCompleted callback, the counter
  81      * will decrement.</p>
  82      *
  83      * <p>It is the user responsibility to close the client.</p>
  84      *
  85      * @param input request to send
  86      * @return CountDownLatch instance that can be use to #await for completeness of processing
  87      */
  88     public CountDownLatch sendRequest(ExecutionServiceInput input) {
  89         return handler.process(input, channel);
  90     }
  91
  92     @Override
  93     public void close() {
  94         if (channel != null) {
  95             channel.shutdown();
  96         }
  97         LOGGER.info("CdsProcessorListener stopped");
  98     }
  99 }