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 SSL error while re-configuring WebClient");
  99                     LOGGER.debug("SSL Exception\n", e);
 100                 }
 101             }
 102         }
 103     }
 104
 105     private void setupWebClient() throws SSLException {
 106         SslContext sslContext = createSslContext();
 107
 108         ClientHttpConnector reactorClientHttpConnector = new ReactorClientHttpConnector(
 109                 HttpClient.create().secure(sslContextSpec -> sslContextSpec.sslContext(sslContext)));
 110
 111         this.webClient = WebClient.builder()
 112                 .baseUrl(aaiClientConfiguration.aaiProtocol() + "://" + aaiClientConfiguration.aaiHost()
 113                         + ":" + aaiClientConfiguration.aaiPort())
 114                 .clientConnector(reactorClientHttpConnector)
 115                 .defaultHeaders(httpHeaders -> httpHeaders.setAll(aaiClientConfiguration.aaiHeaders()))
 116                 .filter(basicAuthentication(aaiClientConfiguration.aaiUserName(),
 117                         aaiClientConfiguration.aaiUserPassword()))
 118                 .filter(logRequest())
 119                 .filter(logResponse())
 120                 .build();
 121     }
 122
 123     public Mono<PnfAaiObject> getPnfObjectDataFor(String url) {
 124
 125         return performReactiveHttpGet(url, PnfAaiObject.class);
 126     }
 127
 128     public Mono<ServiceInstanceAaiObject> getServiceInstanceObjectDataFor(String url) {
 129
 130         return performReactiveHttpGet(url, ServiceInstanceAaiObject.class);
 131     }
 132
 133     private <T> Mono<T> performReactiveHttpGet(String url, Class<T> responseType) {
 134         LOGGER.debug("Will issue Reactive GET request to URL ({}) for object ({})", url, responseType.getName());
 135         WebClient webClient = getWebClient();
 136         return webClient
 137                 .get()
 138                 .uri(url)
 139                 .retrieve()
 140                 .onStatus(HttpStatus::is4xxClientError,
 141                     response -> Mono.error(createExceptionObject(url, response)))
 142                 .onStatus(HttpStatus::is5xxServerError,
 143                     response -> Mono.error(createExceptionObject(url, response)))
 144                 .bodyToMono(String.class)
 145                 .flatMap(body -> extractMono(body, responseType));
 146     }
 147
 148     private AaiTaskException createExceptionObject(String url, ClientResponse response) {
 149         return new AaiTaskException(String.format("A&AI Request for (%s) failed with HTTP status code %d", url,
 150                 response.statusCode().value()));
 151     }
 152
 153     private <T> Mono<T> extractMono(String body, Class<T> responseType) {
 154         LOGGER.debug("Response body \n{}", body);
 155         try {
 156             return Mono.just(parseFromJsonReply(body, responseType));
 157         } catch (JsonSyntaxException | IllegalStateException e) {
 158             return Mono.error(e);
 159         }
 160     }
 161
 162     private <T> T parseFromJsonReply(String body, Class<T> responseType) {
 163         return gson.fromJson(body, responseType);
 164     }
 165
 166     private static ExchangeFilterFunction logRequest() {
 167         return ExchangeFilterFunction.ofRequestProcessor(clientRequest -> {
 168             LOGGER.debug("Request: {} {}", clientRequest.method(), clientRequest.url());
 169             clientRequest.headers()
 170                     .forEach((name, values) -> values.forEach(value -> LOGGER.debug("{}={}", name, value)));
 171             return Mono.just(clientRequest);
 172         });
 173     }
 174
 175     private static ExchangeFilterFunction logResponse() {
 176         return ExchangeFilterFunction.ofResponseProcessor(clientResponse -> {
 177             LOGGER.debug("Response status {}", clientResponse.statusCode());
 178             return Mono.just(clientResponse);
 179         });
 180     }
 181
 182     private SslContext createSslContext() throws SSLException {
 183         if (aaiClientConfiguration.enableAaiCertAuth()) {
 184             return sslFactory.createSecureContext(
 185                     aaiClientConfiguration.keyStorePath(),
 186                     aaiClientConfiguration.keyStorePasswordPath(),
 187                     aaiClientConfiguration.trustStorePath(),
 188                     aaiClientConfiguration.trustStorePasswordPath()
 189             );
 190         }
 191         return sslFactory.createInsecureContext();
 192     }
 193
 194     private synchronized WebClient getWebClient() {
 195         return webClient;
 196     }
 197 }