2 * ============LICENSE_START=======================================================
3 * DCAEGEN2-SERVICES-SDK
4 * ================================================================================
5 * Copyright (C) 2018-2019 NOKIA Intellectual Property. All rights reserved.
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=========================================================
22 package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers;
24 import com.google.gson.JsonArray;
25 import com.google.gson.JsonObject;
26 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpMethod;
27 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpRequest;
28 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.ImmutableHttpRequest;
29 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
30 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClientFactory;
31 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
32 import org.onap.dcaegen2.services.sdk.rest.services.uri.URI;
33 import org.slf4j.Logger;
34 import org.slf4j.LoggerFactory;
35 import reactor.core.publisher.Mono;
38 * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 11/15/18
40 public final class ReactiveCloudConfigurationProvider implements CloudConfigurationProvider {
42 private static final Logger LOGGER = LoggerFactory.getLogger(ReactiveCloudConfigurationProvider.class);
43 private static final String EXCEPTION_MESSAGE = "Unsupported method call: ";
45 private final RxHttpClient rxHttpClient;
47 public ReactiveCloudConfigurationProvider() {
48 this(RxHttpClientFactory.create());
51 ReactiveCloudConfigurationProvider(RxHttpClient rxHttpClient) {
52 this.rxHttpClient = rxHttpClient;
56 public Mono<JsonObject> callForServiceConfigurationReactive(CbsClientConfiguration configuration) {
57 return callConsulForConfigBindingServiceEndpoint(configuration)
58 .flatMap(this::callConfigBindingServiceForConfiguration);
62 public Mono<JsonObject> callForServiceConfigurationReactive(String consulHost, int consulPort, String cbsName,
64 throw new UnsupportedOperationException(EXCEPTION_MESSAGE + this);
68 public JsonObject callForServiceConfiguration(String consulHost, int consulPort, String cbsName, String appName) {
69 throw new UnsupportedOperationException(EXCEPTION_MESSAGE + this);
73 public JsonObject callForServiceConfiguration(CbsClientConfiguration configuration) {
74 throw new UnsupportedOperationException(EXCEPTION_MESSAGE + this);
77 private Mono<String> callConsulForConfigBindingServiceEndpoint(CbsClientConfiguration configuration) {
78 LOGGER.info("Retrieving Config Binding Service endpoint from Consul");
80 HttpRequest httpRequest = ImmutableHttpRequest.builder()
81 .url(getConsulUrl(configuration)).method(HttpMethod.GET).build();
83 return rxHttpClient.call(httpRequest)
84 .map(resp -> resp.bodyAsJson(JsonArray.class))
86 this.createConfigBindingServiceUrl(
88 configuration.appName())
92 private String getConsulUrl(CbsClientConfiguration configuration) {
93 return getUri(configuration.consulHost(), configuration.consulPort(), "/v1/catalog/service",
94 configuration.cbsName());
97 private Mono<JsonObject> callConfigBindingServiceForConfiguration(String configBindingServiceUri) {
98 LOGGER.info("Retrieving configuration");
99 HttpRequest httpRequest = ImmutableHttpRequest.builder()
100 .url(configBindingServiceUri).method(HttpMethod.GET).build();
102 return rxHttpClient.call(httpRequest)
103 .map(httpResponse -> httpResponse.bodyAsJson(JsonObject.class));
107 private Mono<String> createConfigBindingServiceUrl(JsonArray jsonArray, String appName) {
108 return getConfigBindingObject(jsonArray)
109 .flatMap(jsonObject -> buildConfigBindingServiceUrl(jsonObject, appName));
112 private Mono<String> buildConfigBindingServiceUrl(JsonObject jsonObject, String appName) {
113 return Mono.just(getUri(jsonObject.get("ServiceAddress").getAsString(),
114 jsonObject.get("ServicePort").getAsInt(), "/service_component", appName));
117 private Mono<JsonObject> getConfigBindingObject(JsonArray jsonArray) {
119 if (jsonArray.size() > 0) {
120 return Mono.just(jsonArray.get(0).getAsJsonObject());
122 throw new IllegalStateException("JSON Array was empty");
124 } catch (IllegalStateException e) {
125 LOGGER.warn("Failed to retrieve JSON Object from array", e);
126 return Mono.error(e);
130 private String getUri(String host, Integer port, String... paths) {
131 return new URI.URIBuilder()
135 .path(String.join("/", paths))