gerrit.onap Code Review - dcaegen2/services.git/blob

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
   4  * ================================================================================
   5  * Copyright (C) 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
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  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=========================================================
  19  */
  20
  21 package org.onap.bbs.event.processor.utilities;
  22
  23 import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication;
  24
  25 import com.google.gson.Gson;
  26 import com.google.gson.JsonSyntaxException;
  27
  28 import io.netty.handler.ssl.SslContext;
  29
  30 import javax.annotation.PostConstruct;
  31 import javax.annotation.PreDestroy;
  32 import javax.net.ssl.SSLException;
  33
  34 import org.onap.bbs.event.processor.config.AaiClientConfiguration;
  35 import org.onap.bbs.event.processor.config.ApplicationConfiguration;
  36 import org.onap.bbs.event.processor.config.ConfigurationChangeObserver;
  37 import org.onap.bbs.event.processor.exceptions.AaiTaskException;
  38 import org.onap.bbs.event.processor.model.PnfAaiObject;
  39 import org.onap.bbs.event.processor.model.ServiceInstanceAaiObject;
  40 import org.onap.dcaegen2.services.sdk.rest.services.ssl.SslFactory;
  41 import org.slf4j.Logger;
  42 import org.slf4j.LoggerFactory;
  43 import org.springframework.beans.factory.annotation.Autowired;
  44 import org.springframework.http.HttpStatus;
  45 import org.springframework.http.client.reactive.ClientHttpConnector;
  46 import org.springframework.http.client.reactive.ReactorClientHttpConnector;
  47 import org.springframework.stereotype.Component;
  48 import org.springframework.web.reactive.function.client.ClientResponse;
  49 import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
  50 import org.springframework.web.reactive.function.client.WebClient;
  51
  52 import reactor.core.publisher.Mono;
  53 import reactor.netty.http.client.HttpClient;
  54
  55 @Component
  56 public class AaiReactiveClient implements ConfigurationChangeObserver {
  57
  58     private static final Logger LOGGER = LoggerFactory.getLogger(AaiReactiveClient.class);
  59
  60     private final Gson gson;
  61     private WebClient webClient;
  62     private SslFactory sslFactory;
  63     private final ApplicationConfiguration configuration;
  64     private AaiClientConfiguration aaiClientConfiguration;
  65
  66     @Autowired
  67     AaiReactiveClient(ApplicationConfiguration configuration, Gson gson) throws SSLException {
  68         this.configuration = configuration;
  69         this.gson = gson;
  70         this.sslFactory = new SslFactory();
  71
  72         aaiClientConfiguration = this.configuration.getAaiClientConfiguration();
  73         setupWebClient();
  74     }
  75
  76     @PostConstruct
  77     public void registerForConfigChanges() {
  78         configuration.register(this);
  79     }
  80
  81     @PreDestroy
  82     public void unRegisterForConfigChanges() {
  83         configuration.unRegister(this);
  84     }
  85
  86     @Override
  87     public void updateConfiguration() {
  88         AaiClientConfiguration newConfiguration = configuration.getAaiClientConfiguration();
  89         if (aaiClientConfiguration.equals(newConfiguration)) {
  90             LOGGER.info("No Configuration changes necessary for AAI Reactive client");
  91         } else {
  92             synchronized (this) {
  93                 LOGGER.info("AAI Reactive client must be re-configured");
  94                 aaiClientConfiguration = newConfiguration;
  95                 try {
  96                     setupWebClient();
  97                 } catch (SSLException e) {
  98                     LOGGER.error("AAI Reactive client error while re-configuring WebClient");
  99                 }
 100             }
 101         }
 102     }
 103
 104     private void setupWebClient() throws SSLException {
 105         SslContext sslContext = createSslContext();
 106
 107         ClientHttpConnector reactorClientHttpConnector = new ReactorClientHttpConnector(
 108                 HttpClient.create().secure(sslContextSpec -> sslContextSpec.sslContext(sslContext)));
 109
 110         this.webClient = WebClient.builder()
 111                 .baseUrl(aaiClientConfiguration.aaiProtocol() + "://" + aaiClientConfiguration.aaiHost()
 112                         + ":" + aaiClientConfiguration.aaiPort())
 113                 .clientConnector(reactorClientHttpConnector)
 114                 .defaultHeaders(httpHeaders -> httpHeaders.setAll(aaiClientConfiguration.aaiHeaders()))
 115                 .filter(basicAuthentication(aaiClientConfiguration.aaiUserName(),
 116                         aaiClientConfiguration.aaiUserPassword()))
 117                 .filter(logRequest())
 118                 .filter(logResponse())
 119                 .build();
 120     }
 121
 122     public Mono<PnfAaiObject> getPnfObjectDataFor(String url) {
 123
 124         return performReactiveHttpGet(url, PnfAaiObject.class);
 125     }
 126
 127     public Mono<ServiceInstanceAaiObject> getServiceInstanceObjectDataFor(String url) {
 128
 129         return performReactiveHttpGet(url, ServiceInstanceAaiObject.class);
 130     }
 131
 132     private <T> Mono<T> performReactiveHttpGet(String url, Class<T> responseType) {
 133         LOGGER.debug("Will issue Reactive GET request to URL ({}) for object ({})", url, responseType.getName());
 134         WebClient webClient = getWebClient();
 135         return webClient
 136                 .get()
 137                 .uri(url)
 138                 .retrieve()
 139                 .onStatus(HttpStatus::is4xxClientError,
 140                     response -> Mono.error(createExceptionObject(url, response)))
 141                 .onStatus(HttpStatus::is5xxServerError,
 142                     response -> Mono.error(createExceptionObject(url, response)))
 143                 .bodyToMono(String.class)
 144                 .flatMap(body -> extractMono(body, responseType));
 145     }
 146
 147     private AaiTaskException createExceptionObject(String url, ClientResponse response) {
 148         return new AaiTaskException(String.format("A&AI Request for (%s) failed with HTTP status code %d", url,
 149                 response.statusCode().value()));
 150     }
 151
 152     private <T> Mono<T> extractMono(String body, Class<T> responseType) {
 153         LOGGER.debug("Response body \n{}", body);
 154         try {
 155             return Mono.just(parseFromJsonReply(body, responseType));
 156         } catch (JsonSyntaxException | IllegalStateException e) {
 157             return Mono.error(e);
 158         }
 159     }
 160
 161     private <T> T parseFromJsonReply(String body, Class<T> responseType) {
 162         return gson.fromJson(body, responseType);
 163     }
 164
 165     private static ExchangeFilterFunction logRequest() {
 166         return ExchangeFilterFunction.ofRequestProcessor(clientRequest -> {
 167             LOGGER.debug("Request: {} {}", clientRequest.method(), clientRequest.url());
 168             clientRequest.headers()
 169                     .forEach((name, values) -> values.forEach(value -> LOGGER.debug("{}={}", name, value)));
 170             return Mono.just(clientRequest);
 171         });
 172     }
 173
 174     private static ExchangeFilterFunction logResponse() {
 175         return ExchangeFilterFunction.ofResponseProcessor(clientResponse -> {
 176             LOGGER.debug("Response status {}", clientResponse.statusCode());
 177             return Mono.just(clientResponse);
 178         });
 179     }
 180
 181     private SslContext createSslContext() throws SSLException {
 182         if (aaiClientConfiguration.enableAaiCertAuth()) {
 183             return sslFactory.createSecureContext(
 184                     aaiClientConfiguration.keyStorePath(),
 185                     aaiClientConfiguration.keyStorePasswordPath(),
 186                     aaiClientConfiguration.trustStorePath(),
 187                     aaiClientConfiguration.trustStorePasswordPath()
 188             );
 189         }
 190         return sslFactory.createInsecureContext();
 191     }
 192
 193     private synchronized WebClient getWebClient() {
 194         return webClient;
 195     }
 196 }