2 * ============LICENSE_START=======================================================
3 * Copyright (C) 2019-2021 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
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.common.api.ActionIdentifiers;
26 import org.onap.ccsdk.cds.controllerblueprints.processing.api.BluePrintProcessingServiceGrpc;
27 import org.onap.ccsdk.cds.controllerblueprints.processing.api.BluePrintProcessingServiceGrpc.BluePrintProcessingServiceStub;
28 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput;
29 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOutput;
30 import org.onap.policy.cds.api.CdsProcessorListener;
31 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
32 import org.onap.policy.common.endpoints.utils.NetLoggerUtil;
33 import org.onap.policy.common.endpoints.utils.NetLoggerUtil.EventType;
34 import org.slf4j.Logger;
35 import org.slf4j.LoggerFactory;
37 public class CdsProcessorHandler {
38 private static final Logger LOGGER = LoggerFactory.getLogger(CdsProcessorHandler.class);
39 private static final String LOG_MSG = "[{}|{}|{}|]{}{}";
41 private CdsProcessorListener listener;
44 CdsProcessorHandler(final CdsProcessorListener listener, String url) {
45 this.listener = listener;
49 CountDownLatch process(ExecutionServiceInput request, ManagedChannel channel) {
50 final ActionIdentifiers header = request.getActionIdentifiers();
51 LOGGER.info("Processing blueprint({}:{}) for action({})", header.getBlueprintVersion(),
52 header.getBlueprintName(), header.getBlueprintVersion());
54 final CountDownLatch finishLatch = new CountDownLatch(1);
55 final BluePrintProcessingServiceStub asyncStub = BluePrintProcessingServiceGrpc.newStub(channel);
56 final StreamObserver<ExecutionServiceOutput> responseObserver = new StreamObserver<ExecutionServiceOutput>() {
58 public void onNext(ExecutionServiceOutput output) {
59 NetLoggerUtil.log(EventType.IN, CommInfrastructure.REST, url, output.toString());
60 listener.onMessage(output);
64 public void onError(Throwable throwable) {
65 LOGGER.info(LOG_MSG, EventType.IN, CommInfrastructure.REST, url, NetLoggerUtil.SYSTEM_LS,
67 listener.onError(throwable);
68 finishLatch.countDown();
72 public void onCompleted() {
73 LOGGER.info("Completed blueprint({}:{}) for action({})", header.getBlueprintVersion(),
74 header.getBlueprintName(), header.getBlueprintVersion());
75 finishLatch.countDown();
79 final StreamObserver<ExecutionServiceInput> requestObserver = asyncStub.process(responseObserver);
81 NetLoggerUtil.log(EventType.OUT, CommInfrastructure.REST, url, request.toString());
83 // Send the message to CDS backend for processing
84 requestObserver.onNext(request);
85 // Mark the end of requests
86 requestObserver.onCompleted();
87 } catch (RuntimeException e) {
88 requestObserver.onError(e);