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
10 * http://www.apache.org/licenses/LICENSE-2.0
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=========================================================
20 package org.onap.policy.cds.client;
22 import io.grpc.ManagedChannel;
23 import io.grpc.stub.StreamObserver;
24 import java.util.concurrent.CountDownLatch;
25 import org.onap.ccsdk.cds.controllerblueprints.processing.api.BluePrintProcessingServiceGrpc;
26 import org.onap.ccsdk.cds.controllerblueprints.processing.api.BluePrintProcessingServiceGrpc.BluePrintProcessingServiceStub;
27 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput;
28 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOutput;
29 import org.onap.policy.cds.api.CdsProcessorListener;
30 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
31 import org.onap.policy.common.endpoints.utils.NetLoggerUtil;
32 import org.onap.policy.common.endpoints.utils.NetLoggerUtil.EventType;
33 import org.slf4j.Logger;
34 import org.slf4j.LoggerFactory;
36 public class CdsProcessorHandler {
37 private static final Logger LOGGER = LoggerFactory.getLogger(CdsProcessorHandler.class);
38 private static final String LOG_MSG = "[{}|{}|{}|]{}{}";
40 private CdsProcessorListener listener;
43 CdsProcessorHandler(final CdsProcessorListener listener, String url) {
44 this.listener = listener;
48 CountDownLatch process(ExecutionServiceInput request, ManagedChannel channel) {
49 final var header = request.getActionIdentifiers();
50 LOGGER.info("Processing blueprint({}:{}) for action({})", header.getBlueprintVersion(),
51 header.getBlueprintName(), header.getBlueprintVersion());
53 final var finishLatch = new CountDownLatch(1);
54 final BluePrintProcessingServiceStub asyncStub = BluePrintProcessingServiceGrpc.newStub(channel);
55 final StreamObserver<ExecutionServiceOutput> responseObserver = new StreamObserver<>() {
57 public void onNext(ExecutionServiceOutput output) {
58 NetLoggerUtil.log(EventType.IN, CommInfrastructure.REST, url, output.toString());
59 listener.onMessage(output);
63 public void onError(Throwable throwable) {
64 LOGGER.info(LOG_MSG, EventType.IN, CommInfrastructure.REST, url, NetLoggerUtil.SYSTEM_LS,
66 listener.onError(throwable);
67 finishLatch.countDown();
71 public void onCompleted() {
72 LOGGER.info("Completed blueprint({}:{}) for action({})", header.getBlueprintVersion(),
73 header.getBlueprintName(), header.getBlueprintVersion());
74 finishLatch.countDown();
78 final StreamObserver<ExecutionServiceInput> requestObserver = asyncStub.process(responseObserver);
80 NetLoggerUtil.log(EventType.OUT, CommInfrastructure.REST, url, request.toString());
82 // Send the message to CDS backend for processing
83 requestObserver.onNext(request);
84 // Mark the end of requests
85 requestObserver.onCompleted();
86 } catch (RuntimeException e) {
87 requestObserver.onError(e);