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

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * Copyright (C) 2019 Bell Canada.
   4  * Modifications Copyright (C) 2020 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.internal.DnsNameResolverProvider;
  25 import io.grpc.internal.PickFirstLoadBalancerProvider;
  26 import io.grpc.netty.NettyChannelBuilder;
  27 import java.util.concurrent.CountDownLatch;
  28 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput;
  29 import org.onap.policy.cds.api.CdsProcessorListener;
  30 import org.onap.policy.cds.properties.CdsServerProperties;
  31 import org.onap.policy.common.parameters.GroupValidationResult;
  32 import org.slf4j.Logger;
  33 import org.slf4j.LoggerFactory;
  34
  35 /**
  36  * <p>
  37  * The CDS processor client uses gRPC for communication between Policy and CDS. This communication is configured to use
  38  * a streaming approach, which means the client sends an event to which the server can reply with multiple
  39  * sub-responses, until full completion of the processing.
  40  * </p>
  41  */
  42 public class CdsProcessorGrpcClient implements AutoCloseable {
  43
  44     private static final Logger LOGGER = LoggerFactory.getLogger(CdsProcessorGrpcClient.class);
  45
  46     private ManagedChannel channel;
  47     private CdsProcessorHandler handler;
  48
  49     /**
  50      * Constructor, create a CDS processor gRPC client.
  51      *
  52      * @param listener the listener to listen on
  53      */
  54     public CdsProcessorGrpcClient(final CdsProcessorListener listener, CdsServerProperties props) {
  55         final GroupValidationResult validationResult = props.validate();
  56         Preconditions.checkState(validationResult.getStatus().isValid(), "Error validating CDS server "
  57             + "properties: " + validationResult.getResult());
  58
  59         StringBuilder bldr = new StringBuilder("gRPC://");
  60         bldr.append(props.getHost());
  61         bldr.append(":");
  62         bldr.append(props.getPort());
  63         bldr.append('/');
  64
  65         String url = bldr.toString();
  66
  67         this.channel = NettyChannelBuilder.forAddress(props.getHost(), props.getPort())
  68             .nameResolverFactory(new DnsNameResolverProvider())
  69             .loadBalancerFactory(new PickFirstLoadBalancerProvider())
  70             .intercept(new BasicAuthClientHeaderInterceptor(props)).usePlaintext().build();
  71         this.handler = new CdsProcessorHandler(listener, url);
  72         LOGGER.info("CdsProcessorListener started");
  73     }
  74
  75     CdsProcessorGrpcClient(final ManagedChannel channel, final CdsProcessorHandler handler) {
  76         this.channel = channel;
  77         this.handler = handler;
  78     }
  79
  80     /**
  81      * Sends a request to the CDS backend micro-service.
  82      *
  83      * <p>The caller will be returned a CountDownLatch that can be used to define how long the processing can wait. The
  84      * CountDownLatch is initiated with just 1 count. When the client receives an #onCompleted callback, the counter
  85      * will decrement.</p>
  86      *
  87      * <p>It is the user responsibility to close the client.</p>
  88      *
  89      * @param input request to send
  90      * @return CountDownLatch instance that can be use to #await for completeness of processing
  91      */
  92     public CountDownLatch sendRequest(ExecutionServiceInput input) {
  93         return handler.process(input, channel);
  94     }
  95
  96     @Override
  97     public void close() {
  98         if (channel != null) {
  99             channel.shutdown();
 100         }
 101         LOGGER.info("CdsProcessorListener stopped");
 102     }
 103 }