2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2017 - 2019 Bell Canada.
6 * ================================================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=========================================================
21 package org.onap.so.client.cds;
23 import io.grpc.ManagedChannel;
24 import io.grpc.stub.StreamObserver;
25 import java.util.concurrent.CountDownLatch;
26 import org.onap.ccsdk.cds.controllerblueprints.common.api.ActionIdentifiers;
27 import org.onap.ccsdk.cds.controllerblueprints.processing.api.BluePrintProcessingServiceGrpc;
28 import org.onap.ccsdk.cds.controllerblueprints.processing.api.BluePrintProcessingServiceGrpc.BluePrintProcessingServiceStub;
29 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput;
30 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOutput;
31 import org.slf4j.Logger;
32 import org.slf4j.LoggerFactory;
34 class CDSProcessingHandler {
36 private static final Logger log = LoggerFactory.getLogger(CDSProcessingHandler.class);
38 private CDSProcessingListener listener;
40 CDSProcessingHandler(final CDSProcessingListener listener) {
41 this.listener = listener;
44 CountDownLatch process(ExecutionServiceInput request, ManagedChannel channel) {
46 ActionIdentifiers header = request.getActionIdentifiers();
48 log.info("Processing blueprint({}:{}) for action({})", header.getBlueprintVersion(), header.getBlueprintName(),
49 header.getActionName());
51 final CountDownLatch finishLatch = new CountDownLatch(1);
53 final BluePrintProcessingServiceStub asyncStub = BluePrintProcessingServiceGrpc.newStub(channel);
55 final StreamObserver<ExecutionServiceOutput> responseObserver = new StreamObserver<ExecutionServiceOutput>() {
57 public void onNext(ExecutionServiceOutput output) {
58 listener.onMessage(output);
62 public void onError(Throwable t) {
64 finishLatch.countDown();
68 public void onCompleted() {
69 log.info("Completed blueprint({}:{}) for action({})", header.getBlueprintVersion(),
70 header.getBlueprintName(), header.getBlueprintVersion());
71 finishLatch.countDown();
75 final StreamObserver<ExecutionServiceInput> requestObserver = asyncStub.process(responseObserver);
78 // Send our message to CDS backend for processing
79 requestObserver.onNext(request);
80 // Mark the end of requests
81 requestObserver.onCompleted();
82 } catch (RuntimeException e) {
83 requestObserver.onError(e);