import org.jetbrains.annotations.NotNull;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.CbsClientImpl;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.CbsLookup;
-import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.adapters.CloudHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.CloudHttpClient;
import reactor.core.publisher.Mono;
/**
import org.jetbrains.annotations.NotNull;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
-import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.adapters.CloudHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.CloudHttpClient;
import reactor.core.publisher.Mono;
public class CbsClientImpl implements CbsClient {
import com.google.gson.JsonObject;
import java.net.InetSocketAddress;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.EnvProperties;
-import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.adapters.CloudHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.CloudHttpClient;
import reactor.core.publisher.Mono;
/**
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
-import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.adapters.CloudHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.CloudHttpClient;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.EnvProperties;
import org.onap.dcaegen2.services.sdk.rest.services.uri.URI;
import org.slf4j.Logger;
import java.net.InetSocketAddress;
import org.junit.jupiter.api.Test;
import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
-import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.adapters.CloudHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.CloudHttpClient;
import reactor.core.publisher.Mono;
/**
import org.junit.jupiter.api.Test;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.EnvProperties;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.ImmutableEnvProperties;
-import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.adapters.CloudHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.CloudHttpClient;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import com.google.gson.JsonObject;
import org.junit.jupiter.api.Test;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.ImmutableEnvProperties;
-import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.adapters.CloudHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.CloudHttpClient;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.EnvProperties;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
* ============LICENSE_END=====================================
*/
-package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.adapters;
+package org.onap.dcaegen2.services.sdk.rest.services.adapters.http;
import com.google.gson.Gson;
import io.netty.handler.codec.http.HttpStatusClass;
+import io.netty.handler.ssl.SslContext;
import io.vavr.collection.Stream;
import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;
import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;
-import reactor.netty.Connection;
import reactor.netty.http.client.HttpClient;
import reactor.netty.http.client.HttpClientRequest;
import reactor.netty.http.client.HttpClientResponse;
public class CloudHttpClient {
private static final Logger LOGGER = LoggerFactory.getLogger(CloudHttpClient.class);
-
private final Gson gson = new Gson();
private final HttpClient httpClient;
this(HttpClient.create());
}
+ public CloudHttpClient(SslContext sslContext) {
+ this(HttpClient.create().secure(sslContextSpec -> sslContextSpec.sslContext(sslContext)));
+ }
- CloudHttpClient(HttpClient httpClient) {
+ private CloudHttpClient(HttpClient httpClient) {
this.httpClient = httpClient;
+
}
public <T> Mono<T> get(String url, RequestDiagnosticContext context, Class<T> bodyClass) {
+ return get(url, context, Collections.EMPTY_MAP, bodyClass);
+ }
+
+ public <T> Mono<T> get(String url, RequestDiagnosticContext context, Map<String, String> customHeaders, Class<T> bodyClass) {
final HttpClient clientWithHeaders = httpClient
- .doOnRequest((req, conn) -> logRequest(context, req))
- .doOnResponse((rsp, conn) -> logResponse(context, rsp))
- .headers(hdrs -> context.remoteCallHttpHeaders().forEach((BiConsumer<String, String>) hdrs::set));
+ .doOnRequest((req, conn) -> logRequest(context, req))
+ .doOnResponse((rsp, conn) -> logResponse(context, rsp))
+ .headers(hdrs -> context.remoteCallHttpHeaders().forEach((BiConsumer<String, String>) hdrs::set))
+ .headers(hdrs -> customHeaders.forEach(hdrs::set));
return callHttpGet(clientWithHeaders, url, bodyClass);
}
private <T> Mono<T> callHttpGet(HttpClient client, String url, Class<T> bodyClass) {
return client.get()
- .uri(url)
- .responseSingle((resp, content) -> HttpStatusClass.SUCCESS.contains(resp.status().code())
- ? content.asString()
- : Mono.error(createException(url, resp)))
- .map(body -> parseJson(body, bodyClass));
+ .uri(url)
+ .responseSingle((resp, content) -> HttpStatusClass.SUCCESS.contains(resp.status().code())
+ ? content.asString()
+ : Mono.error(createException(url, resp)))
+ .map(body -> parseJson(body, bodyClass));
}
private Exception createException(String url, HttpClientResponse response) {
return new IOException(String.format("Request failed for URL '%s'. Response code: %s",
- url,
- response.status()));
+ url,
+ response.status()));
}
private <T> T parseJson(String body, Class<T> bodyClass) {
LOGGER.debug("Request: {} {}", httpClientRequest.method(), httpClientRequest.uri());
if (LOGGER.isTraceEnabled()) {
final String headers = Stream.ofAll(httpClientRequest.requestHeaders())
- .map(entry -> entry.getKey() + "=" + entry.getValue())
- .collect(Collectors.joining("\n"));
+ .map(entry -> entry.getKey() + "=" + entry.getValue())
+ .collect(Collectors.joining("\n"));
LOGGER.trace(headers);
}
});
<dependency>
<groupId>org.onap.dcaegen2.services.sdk.rest.services</groupId>
<artifactId>aai-client</artifactId>
- <version>1.1.2-SNAPSHOT</version>
+ <version>${project.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer;
-import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.REQUEST_ID;
-import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.X_INVOCATION_ID;
-import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.X_ONAP_REQUEST_ID;
-
import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
import java.util.UUID;
-import java.util.function.Consumer;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.CloudHttpClient;
import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.ImmutableRequestDiagnosticContext;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
import org.onap.dcaegen2.services.sdk.rest.services.uri.URI.URIBuilder;
-import org.slf4j.MDC;
-import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpStatus;
-import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
*/
public class DMaaPConsumerReactiveHttpClient {
+ private final static String SLASH = "/";
+ private final static String CONTENT_TYPE = "Content-Type";
private final String dmaapHostName;
private final String dmaapProtocol;
private final Integer dmaapPortNumber;
private final String consumerGroup;
private final String consumerId;
private final String contentType;
- private final WebClient webClient;
- private final static String SLASH="/";
+ private final CloudHttpClient cloudHttpClient;
/**
* Constructor of DMaaPConsumerReactiveHttpClient.
*
* @param consumerConfiguration - DMaaP consumer configuration object
*/
- public DMaaPConsumerReactiveHttpClient(DmaapConsumerConfiguration consumerConfiguration, WebClient webClient) {
+
+ public DMaaPConsumerReactiveHttpClient(DmaapConsumerConfiguration consumerConfiguration,
+ CloudHttpClient cloudHttpClient) {
this.dmaapHostName = consumerConfiguration.dmaapHostName();
this.dmaapProtocol = consumerConfiguration.dmaapProtocol();
this.dmaapPortNumber = consumerConfiguration.dmaapPortNumber();
this.consumerGroup = consumerConfiguration.consumerGroup();
this.consumerId = consumerConfiguration.consumerId();
this.contentType = consumerConfiguration.dmaapContentType();
- this.webClient = webClient;
+ this.cloudHttpClient = cloudHttpClient;
}
/**
*
* @return reactive response from DMaaP in string format
*/
- public Mono<String> getDMaaPConsumerResponse() {
- return webClient
- .get()
- .uri(getUri())
- .headers(getHeaders())
- .retrieve()
- .onStatus(HttpStatus::is4xxClientError, clientResponse ->
- Mono.error(new RuntimeException("DmaaPConsumer HTTP " + clientResponse.statusCode()))
- )
- .onStatus(HttpStatus::is5xxServerError, clientResponse ->
- Mono.error(new RuntimeException("DmaaPConsumer HTTP " + clientResponse.statusCode())))
- .bodyToMono(String.class);
- }
-
- private Consumer<HttpHeaders> getHeaders() {
- return httpHeaders -> {
- httpHeaders.set(X_ONAP_REQUEST_ID, MDC.get(REQUEST_ID));
- httpHeaders.set(X_INVOCATION_ID, UUID.randomUUID().toString());
- httpHeaders.set(HttpHeaders.CONTENT_TYPE, contentType);
- };
- }
-
- private String createRequestPath() {
- return new StringBuilder().append(SLASH).append(dmaapTopicName).append(SLASH).append(consumerGroup).append(SLASH).append(consumerId).toString();
+ public Mono<String> getDMaaPConsumerResponse(Optional<RequestDiagnosticContext> requestDiagnosticContextOptional) {
+ Map<String,String> headers = new HashMap<>();
+ headers.put(CONTENT_TYPE,contentType);
+ if (requestDiagnosticContextOptional.isPresent()) {
+ return cloudHttpClient.get(getUri().toString(), requestDiagnosticContextOptional.get(),headers, String.class);
+ }
+ RequestDiagnosticContext requestDiagnosticContext = ImmutableRequestDiagnosticContext.builder()
+ .invocationId(UUID.randomUUID()).requestId(UUID.randomUUID()).build();
+ return cloudHttpClient.get(getUri().toString(), requestDiagnosticContext, headers, String.class);
}
URI getUri() {
new URIBuilder().scheme(dmaapProtocol).host(dmaapHostName).port(dmaapPortNumber).path(createRequestPath())
.build().toString());
}
+
+ private String createRequestPath() {
+ return new StringBuilder().append(SLASH).append(dmaapTopicName).append(SLASH).append(consumerGroup)
+ .append(SLASH).append(consumerId).toString();
+ }
}
package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer;
-import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.RESPONSE_CODE;
-import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.SERVICE_NAME;
-
import io.netty.handler.ssl.SslContext;
import javax.net.ssl.SSLException;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.CloudHttpClient;
import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
import org.onap.dcaegen2.services.sdk.rest.services.ssl.SslFactory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.slf4j.MDC;
-import org.springframework.http.client.reactive.ClientHttpConnector;
-import org.springframework.http.client.reactive.ReactorClientHttpConnector;
-import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
-import org.springframework.web.reactive.function.client.WebClient;
-import reactor.core.publisher.Mono;
-import reactor.netty.http.client.HttpClient;
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 7/4/18
*/
public class DMaaPReactiveWebClientFactory {
- private final Logger logger = LoggerFactory.getLogger(this.getClass());
-
private final SslFactory sslFactory;
public DMaaPReactiveWebClientFactory() {
}
/**
- * Construct Reactive WebClient with appropriate settings.
+ * Construct CloudHttpClient with appropriate settings.
*
- * @return WebClient
+ * @return CloudHttpClient
*/
- public WebClient build(DmaapConsumerConfiguration consumerConfiguration) throws SSLException {
+
+ public CloudHttpClient build(DmaapConsumerConfiguration consumerConfiguration) throws SSLException {
SslContext sslContext = createSslContext(consumerConfiguration);
- ClientHttpConnector reactorClientHttpConnector = new ReactorClientHttpConnector(
- HttpClient.create().secure(sslContextSpec -> sslContextSpec.sslContext(sslContext)));
- return WebClient.builder()
- .clientConnector(reactorClientHttpConnector)
- .filter(logRequest())
- .filter(logResponse())
- .build();
+ return new CloudHttpClient(sslContext);
}
private SslContext createSslContext(DmaapConsumerConfiguration consumerConfiguration) throws SSLException {
if (consumerConfiguration.enableDmaapCertAuth()) {
return sslFactory.createSecureContext(
- consumerConfiguration.keyStorePath(), consumerConfiguration.keyStorePasswordPath(),
- consumerConfiguration.trustStorePath(), consumerConfiguration.trustStorePasswordPath()
+ consumerConfiguration.keyStorePath(), consumerConfiguration.keyStorePasswordPath(),
+ consumerConfiguration.trustStorePath(), consumerConfiguration.trustStorePasswordPath()
);
}
return sslFactory.createInsecureContext();
}
-
- private ExchangeFilterFunction logResponse() {
- return ExchangeFilterFunction.ofResponseProcessor(clientResponse -> {
- MDC.put(RESPONSE_CODE, String.valueOf(clientResponse.statusCode()));
- logger.info("Response Status {}", clientResponse.statusCode());
- MDC.remove(RESPONSE_CODE);
- return Mono.just(clientResponse);
- });
- }
-
- private ExchangeFilterFunction logRequest() {
- return ExchangeFilterFunction.ofRequestProcessor(clientRequest -> {
- MDC.put(SERVICE_NAME, String.valueOf(clientRequest.url()));
- logger.info("Request: {} {}", clientRequest.method(), clientRequest.url());
- clientRequest.headers()
- .forEach((name, values) -> values.forEach(value -> logger.info("{}={}", name, value)));
- MDC.remove(SERVICE_NAME);
- return Mono.just(clientRequest);
- });
- }
-
}
package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
-import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication;
import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.http.entity.ContentType;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.CloudHttpClient;
import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
-import org.springframework.http.HttpHeaders;
-import org.springframework.web.reactive.function.client.WebClient;
-import org.springframework.web.reactive.function.client.WebClient.RequestHeadersUriSpec;
-import org.springframework.web.reactive.function.client.WebClient.ResponseSpec;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
private static final String JSON_MESSAGE = "{ \"responseFromDmaap\": \"Success\"}";
private DMaaPConsumerReactiveHttpClient dmaapConsumerReactiveHttpClient;
private DmaapConsumerConfiguration consumerConfigurationMock = mock(DmaapConsumerConfiguration.class);
- private Mono<String> expectedResult = Mono.empty();
- private WebClient webClient;
- private RequestHeadersUriSpec requestHeadersSpec;
- private ResponseSpec responseSpec;
-
+ private Mono<String> expectedResult;
+ private CloudHttpClient httpClient = mock(CloudHttpClient.class);
+ private URI exampleTestUri = URI
+ .create("https://54.45.33.2:1234/unauthenticated.SEC_OTHER_OUTPUT/OpenDCAE-c12/c12");
+ private RequestDiagnosticContext requestDiagnosticContext = mock(RequestDiagnosticContext.class);
@BeforeEach
void setUp() {
when(consumerConfigurationMock.dmaapPortNumber()).thenReturn(1234);
when(consumerConfigurationMock.dmaapUserName()).thenReturn("PRH");
when(consumerConfigurationMock.dmaapUserPassword()).thenReturn("PRH");
- when(consumerConfigurationMock.dmaapContentType()).thenReturn("application/json");
+ when(consumerConfigurationMock.dmaapContentType()).thenReturn(ContentType.APPLICATION_JSON.getMimeType());
when(consumerConfigurationMock.dmaapTopicName()).thenReturn("unauthenticated.SEC_OTHER_OUTPUT");
when(consumerConfigurationMock.consumerGroup()).thenReturn("OpenDCAE-c12");
when(consumerConfigurationMock.consumerId()).thenReturn("c12");
-
- webClient = spy(WebClient.builder()
- .defaultHeader(HttpHeaders.CONTENT_TYPE, consumerConfigurationMock.dmaapContentType())
- .filter(basicAuthentication(consumerConfigurationMock.dmaapUserName(),
- consumerConfigurationMock.dmaapUserPassword()))
- .build());
- dmaapConsumerReactiveHttpClient = new DMaaPConsumerReactiveHttpClient(consumerConfigurationMock, webClient);
- requestHeadersSpec = mock(RequestHeadersUriSpec.class);
- responseSpec = mock(ResponseSpec.class);
+ dmaapConsumerReactiveHttpClient = new DMaaPConsumerReactiveHttpClient(consumerConfigurationMock, httpClient);
}
-
@Test
void getHttpResponse_Success() {
//given
expectedResult = Mono.just(JSON_MESSAGE);
-
+ when(httpClient.get(exampleTestUri.toString(), requestDiagnosticContext, getCustomHeaders(), String.class))
+ .thenReturn(expectedResult);
//when
- mockDependantObjects();
- doReturn(expectedResult).when(responseSpec).bodyToMono(String.class);
- Mono<String> response = dmaapConsumerReactiveHttpClient.getDMaaPConsumerResponse();
-
+ Mono<String> response = dmaapConsumerReactiveHttpClient
+ .getDMaaPConsumerResponse(Optional.of(requestDiagnosticContext));
//then
StepVerifier.create(response).expectSubscription()
.expectNextMatches(results -> {
@Test
void getAppropriateUri_whenPassingCorrectedPathForPnf() {
- Assertions.assertEquals(dmaapConsumerReactiveHttpClient.getUri(),
- URI.create("https://54.45.33.2:1234/unauthenticated.SEC_OTHER_OUTPUT/OpenDCAE-c12/c12"));
+ Assertions.assertEquals(dmaapConsumerReactiveHttpClient.getUri(), exampleTestUri);
}
- private void mockDependantObjects() {
- when(webClient.get()).thenReturn(requestHeadersSpec);
- when(requestHeadersSpec.uri((URI) any())).thenReturn(requestHeadersSpec);
- when(requestHeadersSpec.headers(any())).thenReturn(requestHeadersSpec);
- when(requestHeadersSpec.retrieve()).thenReturn(responseSpec);
- doReturn(responseSpec).when(responseSpec).onStatus(any(), any());
+ private Map<String, String> getCustomHeaders() {
+ Map<String, String> customHeaders = new HashMap<>();
+ customHeaders.put("Content-Type", ContentType.APPLICATION_JSON.getMimeType());
+ return customHeaders;
}
}
\ No newline at end of file
import org.junit.jupiter.api.Test;
import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
import org.onap.dcaegen2.services.sdk.rest.services.ssl.SslFactory;
-import org.springframework.web.reactive.function.client.WebClient;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.CloudHttpClient;
+
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 7/5/18
DmaapConsumerConfiguration dmaapConsumerConfiguration = givenDmaapConfigurationWithSslDisabled();
//when
- WebClient dmaapReactiveWebClient = webClientFactory.build(dmaapConsumerConfiguration);
+ CloudHttpClient dmaapReactiveWebClient = webClientFactory.build(dmaapConsumerConfiguration);
//then
Assertions.assertNotNull(dmaapReactiveWebClient);
DmaapConsumerConfiguration dmaapConsumerConfiguration = givenDmaapConfigurationWithSslEnabled();
//when
- WebClient dmaapReactiveWebClient = webClientFactory.build(dmaapConsumerConfiguration);
+ CloudHttpClient dmaapReactiveWebClient = webClientFactory.build(dmaapConsumerConfiguration);
//then
Assertions.assertNotNull(dmaapReactiveWebClient);
Code Review / dcaegen2 / services / sdk.git / commitdiff