summaryrefslogtreecommitdiffstats
path: root/common/src/main
diff options
context:
space:
mode:
authorSteve Smokowski <ss835w@att.com>2019-02-26 23:16:35 +0000
committerGerrit Code Review <gerrit@onap.org>2019-02-26 23:16:35 +0000
commitdee52db16fa7d9ead3f98bfd18bfd1d747f3553b (patch)
treecfe6269648d8d3ea0da8424aab6f3aca12055ffc /common/src/main
parenta083e7066c8eaf17de2d322751c98e6e30419947 (diff)
parentf07c74c83343da4a8751d7f45fb1edcd111e3647 (diff)
Merge "Add CDS client"
Diffstat (limited to 'common/src/main')
-rw-r--r--common/src/main/java/org/onap/so/client/cds/CDSProcessingClient.java112
-rw-r--r--common/src/main/java/org/onap/so/client/cds/CDSProcessingHandler.java82
-rw-r--r--common/src/main/java/org/onap/so/client/cds/CDSProcessingListener.java25
-rw-r--r--common/src/main/java/org/onap/so/client/cds/CDSProperties.java25
4 files changed, 244 insertions, 0 deletions
diff --git a/common/src/main/java/org/onap/so/client/cds/CDSProcessingClient.java b/common/src/main/java/org/onap/so/client/cds/CDSProcessingClient.java
new file mode 100644
index 0000000000..0901cf589b
--- /dev/null
+++ b/common/src/main/java/org/onap/so/client/cds/CDSProcessingClient.java
@@ -0,0 +1,112 @@
+/*
+ * Copyright (C) 2019 Bell Canada.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onap.so.client.cds;
+
+import io.grpc.ManagedChannel;
+import io.grpc.internal.DnsNameResolverProvider;
+import io.grpc.internal.PickFirstLoadBalancerProvider;
+import io.grpc.netty.NettyChannelBuilder;
+import java.util.concurrent.CountDownLatch;
+import org.onap.ccsdk.apps.controllerblueprints.processing.api.ExecutionServiceInput;
+import org.onap.so.client.PreconditionFailedException;
+import org.onap.so.client.RestPropertiesLoader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * <p>
+ * The CDS processing client is using gRPC for communication between SO and CDS.
+ * That communication is configured to use a streaming approach, meaning that
+ * client can send an event to which server can reply will multiple sub-responses,
+ * until full completion of the processing.
+ * </p>
+ * <p>
+ * In order for the caller to manage the callback, it is the responsibility of the
+ * caller to implement and provide a {@link CDSProcessingListener} so received messages
+ * can be handled appropriately.
+ * </p>
+ *
+ * Here is an example of implementation of such listener:
+ * <pre>
+ * new CDSProcessingListener {
+ *
+ * &#64;Override
+ * public void onMessage(ExecutionServiceOutput message) {
+ * log.info("Received notification from CDS: {}", message);
+ * }
+ *
+ * &#64;Override
+ * public void onError(Throwable t) {
+ * Status status = Status.fromThrowable(t);
+ * log.error("Failed processing blueprint {}", status, t);
+ * }
+ * }
+ * </pre>
+ */
+public class CDSProcessingClient implements AutoCloseable {
+
+ private static final Logger log = LoggerFactory.getLogger(CDSProcessingClient.class);
+
+ private ManagedChannel channel;
+ private CDSProcessingHandler handler;
+
+ public CDSProcessingClient(final CDSProcessingListener listener) {
+ CDSProperties props = RestPropertiesLoader.getInstance().getNewImpl(CDSProperties.class);
+ if (props == null) {
+ throw new PreconditionFailedException(
+ "No RestProperty.CDSProperties implementation found on classpath, can't create client.");
+ }
+ this.channel = NettyChannelBuilder
+ .forAddress(props.getHost(), props.getPort())
+ .nameResolverFactory(new DnsNameResolverProvider())
+ .loadBalancerFactory(new PickFirstLoadBalancerProvider())
+ .usePlaintext()
+ .build();
+ this.handler = new CDSProcessingHandler(listener);
+ log.info("CDSProcessingClient started");
+ }
+
+ CDSProcessingClient(final ManagedChannel channel, final CDSProcessingHandler handler) {
+ this.channel = channel;
+ this.handler = handler;
+ }
+
+ /**
+ * Sends a request to the CDS backend micro-service.
+ *
+ * The caller will be returned a CountDownLatch that can be used
+ * to define how long the processing can wait. The CountDownLatch is
+ * initiated with just 1 count. When the client receives an #onCompleted callback,
+ * the counter will decrement.
+ *
+ * It is the user responsibility to close the client.
+ *
+ * @param input request to send
+ * @return CountDownLatch instance that can be use to #await for
+ * completeness of processing
+ */
+ public CountDownLatch sendRequest(ExecutionServiceInput input) {
+ return handler.process(input, channel);
+ }
+
+ @Override
+ public void close() {
+ if (channel != null) {
+ channel.shutdown();
+ }
+ log.info("CDSProcessingClient stopped");
+ }
+} \ No newline at end of file
diff --git a/common/src/main/java/org/onap/so/client/cds/CDSProcessingHandler.java b/common/src/main/java/org/onap/so/client/cds/CDSProcessingHandler.java
new file mode 100644
index 0000000000..244b89a6f5
--- /dev/null
+++ b/common/src/main/java/org/onap/so/client/cds/CDSProcessingHandler.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright (C) 2019 Bell Canada.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onap.so.client.cds;
+
+import io.grpc.ManagedChannel;
+import io.grpc.stub.StreamObserver;
+import java.util.concurrent.CountDownLatch;
+import org.onap.ccsdk.apps.controllerblueprints.common.api.ActionIdentifiers;
+import org.onap.ccsdk.apps.controllerblueprints.processing.api.BluePrintProcessingServiceGrpc;
+import org.onap.ccsdk.apps.controllerblueprints.processing.api.BluePrintProcessingServiceGrpc.BluePrintProcessingServiceStub;
+import org.onap.ccsdk.apps.controllerblueprints.processing.api.ExecutionServiceInput;
+import org.onap.ccsdk.apps.controllerblueprints.processing.api.ExecutionServiceOutput;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class CDSProcessingHandler {
+
+ private static final Logger log = LoggerFactory.getLogger(CDSProcessingHandler.class);
+
+ private CDSProcessingListener listener;
+
+ CDSProcessingHandler(final CDSProcessingListener listener) {
+ this.listener = listener;
+ }
+
+ CountDownLatch process(ExecutionServiceInput request, ManagedChannel channel) {
+
+ ActionIdentifiers header = request.getActionIdentifiers();
+
+ log.info("Processing blueprint({}:{}) for action({})", header.getBlueprintVersion(), header.getBlueprintName(),
+ header.getBlueprintVersion());
+
+ final CountDownLatch finishLatch = new CountDownLatch(1);
+
+ final BluePrintProcessingServiceStub asyncStub = BluePrintProcessingServiceGrpc.newStub(channel);
+
+ final StreamObserver<ExecutionServiceOutput> responseObserver = new StreamObserver<ExecutionServiceOutput>() {
+ @Override
+ public void onNext(ExecutionServiceOutput output) {
+ listener.onMessage(output);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ listener.onError(t);
+ finishLatch.countDown();
+ }
+
+ @Override
+ public void onCompleted() {
+ log.info("Completed blueprint({}:{}) for action({})", header.getBlueprintVersion(),
+ header.getBlueprintName(), header.getBlueprintVersion());
+ finishLatch.countDown();
+ }
+ };
+
+ final StreamObserver<ExecutionServiceInput> requestObserver = asyncStub.process(responseObserver);
+
+ try {
+ // Send our message to CDS backend for processing
+ requestObserver.onNext(request);
+ // Mark the end of requests
+ requestObserver.onCompleted();
+ } catch (RuntimeException e) {
+ requestObserver.onError(e);
+ }
+ return finishLatch;
+ }
+} \ No newline at end of file
diff --git a/common/src/main/java/org/onap/so/client/cds/CDSProcessingListener.java b/common/src/main/java/org/onap/so/client/cds/CDSProcessingListener.java
new file mode 100644
index 0000000000..2eae4ef5a7
--- /dev/null
+++ b/common/src/main/java/org/onap/so/client/cds/CDSProcessingListener.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright (C) 2019 Bell Canada.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onap.so.client.cds;
+
+import org.onap.ccsdk.apps.controllerblueprints.processing.api.ExecutionServiceOutput;
+
+public interface CDSProcessingListener {
+
+ void onMessage(ExecutionServiceOutput message);
+
+ void onError(Throwable t);
+}
diff --git a/common/src/main/java/org/onap/so/client/cds/CDSProperties.java b/common/src/main/java/org/onap/so/client/cds/CDSProperties.java
new file mode 100644
index 0000000000..bb2a54ec98
--- /dev/null
+++ b/common/src/main/java/org/onap/so/client/cds/CDSProperties.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright (C) 2019 Bell Canada.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onap.so.client.cds;
+
+import org.onap.so.client.RestProperties;
+
+
+public interface CDSProperties extends RestProperties {
+
+ String getHost();
+ int getPort();
+}