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