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
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 LOGGER.info(LOG_MSG, EventType.IN, CommInfrastructure.REST, url, NetLoggerUtil.SYSTEM_LS,
61 NetLoggerUtil.log(EventType.IN, CommInfrastructure.REST, url, output.toString());
63 listener.onMessage(output);
67 public void onError(Throwable throwable) {
68 LOGGER.info(LOG_MSG, EventType.IN, CommInfrastructure.REST, url, NetLoggerUtil.SYSTEM_LS,
70 NetLoggerUtil.log(EventType.IN, CommInfrastructure.REST, url, throwable.toString());
71 listener.onError(throwable);
72 finishLatch.countDown();
76 public void onCompleted() {
77 LOGGER.info("Completed blueprint({}:{}) for action({})", header.getBlueprintVersion(),
78 header.getBlueprintName(), header.getBlueprintVersion());
79 finishLatch.countDown();
83 final StreamObserver<ExecutionServiceInput> requestObserver = asyncStub.process(responseObserver);
85 LOGGER.info(LOG_MSG, EventType.OUT, CommInfrastructure.REST, url, NetLoggerUtil.SYSTEM_LS,
87 NetLoggerUtil.log(EventType.OUT, CommInfrastructure.REST, url, request.toString());
89 // Send the message to CDS backend for processing
90 requestObserver.onNext(request);
91 // Mark the end of requests
92 requestObserver.onCompleted();
93 } catch (RuntimeException e) {
94 requestObserver.onError(e);