<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-bom</artifactId>
- <version>Bismuth-RELEASE</version>
+ <version>Bismuth-SR10</version>
<type>pom</type>
<scope>import</scope>
</dependency>
import static org.onap.dcaegen2.services.prh.model.logging.MDCVariables.SERVICE_NAME;
import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
+
import java.util.Map;
+import javax.net.ssl.SSLException;
+
import org.onap.dcaegen2.services.prh.config.AaiClientConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
+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;
public class AaiReactiveWebClient {
- private Logger logger = LoggerFactory.getLogger(this.getClass());
+ private static final Logger LOGGER = LoggerFactory.getLogger(AaiReactiveWebClient.class);
private String aaiUserName;
private String aaiUserPassword;
*
* @return WebClient
*/
- public WebClient build() {
+ public WebClient build() throws SSLException {
+ SslContext sslContext;
+ sslContext = SslContextBuilder
+ .forClient()
+ .trustManager(InsecureTrustManagerFactory.INSTANCE)
+ .build();
+ LOGGER.debug("Setting ssl context");
+
return WebClient.builder()
+ .clientConnector(new ReactorClientHttpConnector(clientOptions -> {
+ clientOptions.sslContext(sslContext);
+ clientOptions.disablePool();
+ }))
.defaultHeaders(httpHeaders -> httpHeaders.setAll(aaiHeaders))
.filter(basicAuthentication(aaiUserName, aaiUserPassword))
.filter(logRequest())
private ExchangeFilterFunction logRequest() {
return ExchangeFilterFunction.ofRequestProcessor(clientRequest -> {
MDC.put(SERVICE_NAME, String.valueOf(clientRequest.url()));
- logger.info("Request: {} {}", clientRequest.method(), clientRequest.url());
+ LOGGER.info("Request: {} {}", clientRequest.method(), clientRequest.url());
clientRequest.headers()
- .forEach((name, values) -> values.forEach(value -> logger.info("{}={}", name, value)));
+ .forEach((name, values) -> values.forEach(value -> LOGGER.info("{}={}", name, value)));
return Mono.just(clientRequest);
});
}
private ExchangeFilterFunction logResponse() {
return ExchangeFilterFunction.ofResponseProcessor(clientResponse -> {
MDC.put(RESPONSE_CODE, String.valueOf(clientResponse.statusCode()));
- logger.info("Response Status {}", clientResponse.statusCode());
+ LOGGER.info("Response Status {}", clientResponse.statusCode());
return Mono.just(clientResponse);
});
}
package org.onap.dcaegen2.services.prh.service.producer;
-import static org.onap.dcaegen2.services.prh.model.logging.MDCVariables.REQUEST_ID;
-import static org.onap.dcaegen2.services.prh.model.logging.MDCVariables.X_INVOCATION_ID;
-import static org.onap.dcaegen2.services.prh.model.logging.MDCVariables.X_ONAP_REQUEST_ID;
-
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.UUID;
import org.apache.http.client.utils.URIBuilder;
import org.onap.dcaegen2.services.prh.config.AaiClientConfiguration;
-import org.onap.dcaegen2.services.prh.exceptions.AaiRequestException;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
-import org.springframework.http.HttpStatus;
-import org.springframework.web.reactive.function.BodyInserters;
+import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.UUID;
+
+import static org.onap.dcaegen2.services.prh.model.CommonFunctions.createJsonBody;
+import static org.onap.dcaegen2.services.prh.model.logging.MDCVariables.*;
+
public class AaiProducerReactiveHttpClient {
+ private WebClient webClient;
private final String aaiHost;
private final String aaiProtocol;
private final Integer aaiHostPortNumber;
private final String aaiBasePath;
- private final Logger logger = LoggerFactory.getLogger(this.getClass());
- private WebClient webClient;
-
+ private final String aaiPnfPath;
/**
* Constructor of AaiProducerReactiveHttpClient.
this.aaiProtocol = configuration.aaiProtocol();
this.aaiHostPortNumber = configuration.aaiPort();
this.aaiBasePath = configuration.aaiBasePath();
+ this.aaiPnfPath = configuration.aaiPnfPath();
}
/**
* @param consumerDmaapModelMono - object which will be sent to AAI database
* @return status code of operation
*/
- public Mono<Integer> getAaiProducerResponse(Mono<ConsumerDmaapModel> consumerDmaapModelMono) {
- return consumerDmaapModelMono
- .doOnNext(consumerDmaapModel -> logger.info("Sending PNF model to AAI {}", consumerDmaapModel))
- .flatMap(this::patchAaiRequest);
+ public Mono<ClientResponse> getAaiProducerResponse(ConsumerDmaapModel consumerDmaapModelMono) {
+ try {
+ return patchAaiRequest(consumerDmaapModelMono);
+ } catch (URISyntaxException e) {
+ return Mono.error(e);
+ }
}
public AaiProducerReactiveHttpClient createAaiWebClient(WebClient webClient) {
return this;
}
- private Mono<Integer> patchAaiRequest(ConsumerDmaapModel dmaapModel) {
- try {
- return webClient.patch()
+ private Mono<ClientResponse> patchAaiRequest(ConsumerDmaapModel dmaapModel) throws URISyntaxException {
+ return
+ webClient.patch()
.uri(getUri(dmaapModel.getSourceName()))
.header(X_ONAP_REQUEST_ID, MDC.get(REQUEST_ID))
.header(X_INVOCATION_ID, UUID.randomUUID().toString())
- .body(BodyInserters.fromObject(dmaapModel))
- .retrieve()
- .onStatus(
- HttpStatus::is4xxClientError,
- clientResponse -> Mono
- .error(new AaiRequestException("AaiProducer HTTP " + clientResponse.statusCode()))
- )
- .onStatus(HttpStatus::is5xxServerError,
- clientResponse -> Mono
- .error(new AaiRequestException("AaiProducer HTTP " + clientResponse.statusCode())))
- .bodyToMono(Integer.class);
- } catch (URISyntaxException e) {
- return Mono.error(e);
- }
+ .body(Mono.just(createJsonBody(dmaapModel)), String.class)
+ .exchange();
}
URI getUri(String pnfName) throws URISyntaxException {
.setScheme(aaiProtocol)
.setHost(aaiHost)
.setPort(aaiHostPortNumber)
- .setPath(aaiBasePath + "/" + pnfName)
+ .setPath(aaiBasePath + aaiPnfPath + "/" + pnfName)
.build();
}
}
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;
+
import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.onap.dcaegen2.services.prh.config.AaiClientConfiguration;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModelForUnitTest;
+import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClient.ResponseSpec;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
-
class AaiProducerReactiveHttpClientTest {
private static final Integer SUCCESS_RESPONSE = 200;
private static AaiClientConfiguration aaiConfigurationMock = mock(AaiClientConfiguration.class);
private static WebClient webClient = mock(WebClient.class);
- private static ConsumerDmaapModel dmaapModel = new ConsumerDmaapModelForUnitTest();
- private static WebClient.RequestBodyUriSpec requestBodyUriSpec;
- private static ResponseSpec responseSpec;
- private static Map<String, String> aaiHeaders;
+ private ConsumerDmaapModel dmaapModel = spy(new ConsumerDmaapModelForUnitTest());
+ private WebClient.RequestBodyUriSpec requestBodyUriSpec;
+ private ResponseSpec responseSpec;
- @BeforeAll
- static void setUp() {
- setupHeaders();
+ private Map<String, String> aaiHeaders;
+ private ClientResponse clientResponse;
+ private Mono<ClientResponse> clientResponseMono;
+ @BeforeEach
+ void setUp() {
+ setupHeaders();
+ clientResponse = mock(ClientResponse.class);
+ clientResponseMono = Mono.just(clientResponse);
+ when(dmaapModel.getSourceName()).thenReturn("NOKnhfsadhff");
when(aaiConfigurationMock.aaiHost()).thenReturn("54.45.33.2");
when(aaiConfigurationMock.aaiProtocol()).thenReturn("https");
when(aaiConfigurationMock.aaiPort()).thenReturn(1234);
responseSpec = mock(ResponseSpec.class);
}
- private static void setupHeaders() {
- aaiHeaders = new HashMap<>();
- aaiHeaders.put("X-FromAppId", "PRH");
- aaiHeaders.put("X-TransactionId", "vv-temp");
- aaiHeaders.put("Accept", "application/json");
- aaiHeaders.put("Real-Time", "true");
- aaiHeaders.put("Content-Type", "application/merge-patch+json");
- }
-
@Test
void getAaiProducerResponse_shouldReturn200() {
//given
mockWebClientDependantObject();
doReturn(expectedResult).when(responseSpec).bodyToMono(Integer.class);
aaiProducerReactiveHttpClient.createAaiWebClient(webClient);
- Mono<Integer> response = aaiProducerReactiveHttpClient.getAaiProducerResponse(Mono.just(dmaapModel));
//then
- StepVerifier.create(response).expectSubscription()
+ StepVerifier.create(aaiProducerReactiveHttpClient.getAaiProducerResponse(dmaapModel)).expectSubscription()
.expectNextMatches(results -> {
- Assertions.assertEquals(results, expectedResult.block());
+ Assertions.assertEquals(results, clientResponse);
return true;
}).verifyComplete();
}
//when
when(webClient.patch()).thenReturn(requestBodyUriSpec);
aaiProducerReactiveHttpClient.createAaiWebClient(webClient);
- when(aaiProducerReactiveHttpClient.getUri("pnfName")).thenThrow(URISyntaxException.class);
-
+ doThrow(URISyntaxException.class).when(aaiProducerReactiveHttpClient).getUri(any());
//then
StepVerifier.create(
aaiProducerReactiveHttpClient.getAaiProducerResponse(
- Mono.just(dmaapModel)
+ dmaapModel
)).expectSubscription().expectError(Exception.class).verify();
}
+ @Test
+ void getAppropriateUri_whenPassingCorrectedPathForPnf() throws URISyntaxException {
+ Assertions.assertEquals(aaiProducerReactiveHttpClient.getUri("NOKnhfsadhff"),
+ URI.create("https://54.45.33.2:1234/aai/v11/network/pnfs/pnf/NOKnhfsadhff"));
+ }
+
+
+ private void setupHeaders() {
+ aaiHeaders = new HashMap<>();
+ aaiHeaders.put("X-FromAppId", "PRH");
+ aaiHeaders.put("X-TransactionId", "vv-temp");
+ aaiHeaders.put("Accept", "application/json");
+ aaiHeaders.put("Real-Time", "true");
+ aaiHeaders.put("Content-Type", "application/merge-patch+json");
+ }
+
private void mockWebClientDependantObject() {
WebClient.RequestHeadersSpec requestHeadersSpec = mock(WebClient.RequestHeadersSpec.class);
when(webClient.patch()).thenReturn(requestBodyUriSpec);
when(requestBodyUriSpec.uri((URI) any())).thenReturn(requestBodyUriSpec);
when(requestBodyUriSpec.header(any(), any())).thenReturn(requestBodyUriSpec);
- when(requestBodyUriSpec.body(any())).thenReturn(requestHeadersSpec);
- doReturn(responseSpec).when(requestHeadersSpec).retrieve();
- doReturn(responseSpec).when(responseSpec).onStatus(any(), any());
+ when(requestBodyUriSpec.body(any(), (Class<Object>) any())).thenReturn(requestHeadersSpec);
+ when(requestHeadersSpec.exchange()).thenReturn(clientResponseMono);
}
-
}
logging:
level:
ROOT: ERROR
+ org.onap.dcaegen2.services.prh: INFO
+ reactor.ipc.netty.http.client: WARN
org.springframework: ERROR
org.springframework.data: ERROR
- org.onap.dcaegen2.services.prh: INFO
+ org.springframework.web.reactive: WARN
app:
filepath: config/prh_endpoints.json
\ No newline at end of file
"aaiBasePath": "/aai/v12",
"aaiPnfPath": "/network/pnfs/pnf",
"aaiHeaders": {
+ "X-FromAppId": "prh",
+ "X-TransactionId": "9999",
"Accept": "application/json",
"Real-Time": "true",
"Content-Type": "application/merge-patch+json"
package org.onap.dcaegen2.services.prh;
import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.jackson.JacksonAutoConfiguration;
import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18
}
@Bean
- ConcurrentTaskScheduler concurrentTaskScheduler() {
+ TaskScheduler concurrentTaskScheduler() {
return new ConcurrentTaskScheduler();
}
-
- @Bean
- ThreadPoolTaskScheduler threadPoolTaskScheduler() {
- ThreadPoolTaskScheduler threadPoolTaskScheduler
- = new ThreadPoolTaskScheduler();
- threadPoolTaskScheduler.setPoolSize(5);
- threadPoolTaskScheduler.setThreadNamePrefix(
- "CloudThreadPoolTaskScheduler");
- return threadPoolTaskScheduler;
- }
}
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.scheduling.annotation.EnableScheduling;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;
@Primary
public class CloudConfiguration extends AppConfig {
- private Logger logger = LoggerFactory.getLogger(this.getClass());
+ private static final Logger LOGGER = LoggerFactory.getLogger(CloudConfiguration.class);
private PrhConfigurationProvider prhConfigurationProvider;
private AaiClientConfiguration aaiClientCloudConfiguration;
}
private void parsingConfigError(Throwable throwable) {
- logger.warn("Error in case of processing system environment, more details below: ", throwable);
+ LOGGER.warn("Error in case of processing system environment, more details below: ", throwable);
}
private void cloudConfigError(Throwable throwable) {
- logger.warn("Exception during getting configuration from CONSUL/CONFIG_BINDING_SERVICE ", throwable);
+ LOGGER.warn("Exception during getting configuration from CONSUL/CONFIG_BINDING_SERVICE ", throwable);
}
private void parsingConfigSuccess(EnvProperties envProperties) {
- logger.info("Fetching PRH configuration from ConfigBindingService/Consul");
+ LOGGER.info("Fetching PRH configuration from ConfigBindingService/Consul");
prhConfigurationProvider.callForPrhConfiguration(envProperties)
.subscribe(this::parseCloudConfig, this::cloudConfigError);
}
private void parseCloudConfig(JsonObject jsonObject) {
- logger.info("Received application configuration: {}", jsonObject);
+ LOGGER.info("Received application configuration: {}", jsonObject);
CloudConfigParser cloudConfigParser = new CloudConfigParser(jsonObject);
dmaapPublisherCloudConfiguration = cloudConfigParser.getDmaapPublisherConfig();
aaiClientCloudConfiguration = cloudConfigParser.getAaiClientConfig();
package org.onap.dcaegen2.services.prh.configuration;
-import java.util.Optional;
-import java.util.Properties;
import org.onap.dcaegen2.services.prh.exceptions.EnvironmentLoaderException;
import org.onap.dcaegen2.services.prh.model.EnvProperties;
import org.onap.dcaegen2.services.prh.model.ImmutableEnvProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.Optional;
+import java.util.Properties;
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 8/10/18
class EnvironmentProcessor {
private static final int DEFAULT_CONSUL_PORT = 8500;
- private static Logger logger = LoggerFactory.getLogger(EnvironmentProcessor.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(EnvironmentProcessor.class);
private EnvironmentProcessor() {
}
- static Flux<EnvProperties> evaluate(Properties systemEnvironment) {
- logger.info("Loading configuration from system environment variables {}", systemEnvironment);
+ static Mono<EnvProperties> evaluate(Properties systemEnvironment) {
+ LOGGER.info("Loading configuration from system environment variables");
EnvProperties envProperties;
try {
envProperties = ImmutableEnvProperties.builder().consulHost(getConsulHost(systemEnvironment))
.consulPort(getConsultPort(systemEnvironment)).cbsName(getConfigBindingService(systemEnvironment))
.appName(getService(systemEnvironment)).build();
} catch (EnvironmentLoaderException e) {
- return Flux.error(e);
+ return Mono.error(e);
}
- logger.info("Evaluated environment system variables {}", envProperties);
- return Flux.just(envProperties);
+ LOGGER.info("Evaluated environment system variables {}", envProperties);
+ return Mono.just(envProperties);
}
private static String getConsulHost(Properties systemEnvironments) throws EnvironmentLoaderException {
}
private static Integer getDefaultPortOfConsul() {
- logger.warn("$CONSUL_PORT environment has not been defined");
- logger.warn("$CONSUL_PORT variable will be set to default port {}", DEFAULT_CONSUL_PORT);
+ LOGGER.warn("$CONSUL_PORT environment has not been defined");
+ LOGGER.warn("$CONSUL_PORT variable will be set to default port {}", DEFAULT_CONSUL_PORT);
return DEFAULT_CONSUL_PORT;
}
}
private static final String DMAAP_PRODUCER = "dmaapProducerConfiguration";
private static final String DMAAP_CONSUMER = "dmaapConsumerConfiguration";
- private final Logger logger = LoggerFactory.getLogger(this.getClass());
+ private static final Logger LOGGER = LoggerFactory.getLogger(PrhAppConfig.class);
AaiClientConfiguration aaiClientConfiguration;
DmaapPublisherConfiguration.class);
}
} catch (IOException e) {
- logger.warn("Problem with file loading, file: {}", filepath, e);
+ LOGGER.warn("Problem with file loading, file: {}", filepath, e);
} catch (JsonSyntaxException e) {
- logger.warn("Problem with Json deserialization", e);
+ LOGGER.warn("Problem with Json deserialization", e);
}
}
package org.onap.dcaegen2.services.prh.configuration;
import io.swagger.annotations.ApiOperation;
-import java.time.Duration;
-import java.time.Instant;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.ScheduledFuture;
-import javax.annotation.PostConstruct;
import org.onap.dcaegen2.services.prh.tasks.ScheduledTasks;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.Marker;
import org.slf4j.MarkerFactory;
import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.annotation.EnableScheduling;
-import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import reactor.core.publisher.Mono;
+import javax.annotation.PostConstruct;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ScheduledFuture;
+
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 6/13/18
*/
@EnableScheduling
public class SchedulerConfig {
- private static final int SCHEDULING_DELAY_FOR_PRH_TASKS = 5;
+ private static final int SCHEDULING_DELAY_FOR_PRH_TASKS = 10;
private static final int SCHEDULING_REQUEST_FOR_CONFIGURATION_DELAY = 5;
+ private static final Logger LOGGER = LoggerFactory.getLogger(SchedulerConfig.class);
+ private static final Marker ENTRY = MarkerFactory.getMarker("ENTRY");
private static volatile List<ScheduledFuture> scheduledPrhTaskFutureList = new ArrayList<>();
- private final Logger logger = LoggerFactory.getLogger(this.getClass());
- private final Marker ENTRY = MarkerFactory.getMarker("ENRTY");
- private final ConcurrentTaskScheduler taskScheduler;
+ private final TaskScheduler taskScheduler;
private final ScheduledTasks scheduledTask;
- private final TaskScheduler cloudTaskScheduler;
private final CloudConfiguration cloudConfiguration;
@Autowired
- public SchedulerConfig(@Qualifier("concurrentTaskScheduler") ConcurrentTaskScheduler concurrentTaskScheduler,
- ScheduledTasks scheduledTask, ThreadPoolTaskScheduler cloudTaskScheduler,
- CloudConfiguration cloudConfiguration) {
- this.taskScheduler = concurrentTaskScheduler;
+ public SchedulerConfig(TaskScheduler taskScheduler,
+ ScheduledTasks scheduledTask,
+ CloudConfiguration cloudConfiguration) {
+ this.taskScheduler = taskScheduler;
this.scheduledTask = scheduledTask;
- this.cloudTaskScheduler = cloudTaskScheduler;
this.cloudConfiguration = cloudConfiguration;
}
@PostConstruct
@ApiOperation(value = "Start task if possible")
public synchronized boolean tryToStartTask() {
- logger.info(ENTRY,"Start scheduling PRH workflow");
+ LOGGER.info(ENTRY, "Start scheduling PRH workflow");
if (scheduledPrhTaskFutureList.isEmpty()) {
- scheduledPrhTaskFutureList.add(cloudTaskScheduler
+ scheduledPrhTaskFutureList.add(taskScheduler
.scheduleAtFixedRate(cloudConfiguration::runTask, Instant.now(),
Duration.ofMinutes(SCHEDULING_REQUEST_FOR_CONFIGURATION_DELAY)));
scheduledPrhTaskFutureList.add(taskScheduler
@Api(value = "HeartbeatController", description = "Check liveness of PRH service")
public class HeartbeatController {
- private final Logger logger = LoggerFactory.getLogger(this.getClass());
+ private static final Logger LOGGER = LoggerFactory.getLogger(HeartbeatController.class);
/**
* Endpoint for checking that PRH is alive.
}
)
public Mono<ResponseEntity<String>> heartbeat() {
- logger.trace("Receiving heartbeat request");
+ LOGGER.trace("Receiving heartbeat request");
return Mono.defer(() ->
Mono.just(new ResponseEntity<>("alive", HttpStatus.OK))
);
@Api(value = "ScheduleController", description = "Schedule Controller")
public class ScheduleController {
- private final Logger logger = LoggerFactory.getLogger(this.getClass());
+ private static final Logger LOGGER = LoggerFactory.getLogger(ScheduleController.class);
private final SchedulerConfig schedulerConfig;
@RequestMapping(value = "start", method = RequestMethod.GET)
@ApiOperation(value = "Start scheduling worker request")
public Mono<ResponseEntity<String>> startTasks() {
- logger.trace("Receiving start scheduling worker request");
+ LOGGER.trace("Receiving start scheduling worker request");
return Mono.fromSupplier(schedulerConfig::tryToStartTask).map(this::createStartTaskResponse);
}
@RequestMapping(value = "stopPrh", method = RequestMethod.GET)
@ApiOperation(value = "Receiving stop scheduling worker request")
public Mono<ResponseEntity<String>> stopTask() {
- logger.trace("Receiving stop scheduling worker request");
+ LOGGER.trace("Receiving stop scheduling worker request");
return schedulerConfig.getResponseFromCancellationOfTasks();
}
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
-import java.util.Optional;
-import java.util.stream.StreamSupport;
import org.onap.dcaegen2.services.prh.exceptions.DmaapEmptyResponseException;
import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
import org.onap.dcaegen2.services.prh.model.ImmutableConsumerDmaapModel;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Mono;
+import java.util.Optional;
+import java.util.stream.StreamSupport;
+
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/8/18
*/
private static final String OAM_IPV_6_ADDRESS = "oamV6IpAddress";
private static final String SOURCE_NAME = "sourceName";
- private final Logger logger = LoggerFactory.getLogger(this.getClass());
-
/**
* Extract info from string and create @see {@link org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel}.
*
*/
public Mono<ConsumerDmaapModel> getJsonObject(Mono<String> monoMessage) {
return monoMessage
- .doOnNext(message -> logger.info("Consumed message from DmaaP: {}", message))
.flatMap(this::getJsonParserMessage)
.flatMap(this::createJsonConsumerModel);
}
private Mono<JsonElement> getJsonParserMessage(String message) {
return StringUtils.isEmpty(message) ? Mono.error(new DmaapEmptyResponseException())
- : Mono.fromSupplier(() -> new JsonParser().parse(message));
+ : Mono.fromCallable(() -> new JsonParser().parse(message));
}
private Mono<ConsumerDmaapModel> createJsonConsumerModel(JsonElement jsonElement) {
return jsonElement.isJsonObject()
- ? create(Mono.fromSupplier(jsonElement::getAsJsonObject))
+ ? create(Mono.fromCallable(jsonElement::getAsJsonObject))
: getConsumerDmaapModelFromJsonArray(jsonElement);
}
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
-public class HttpGetClient {
+class HttpGetClient {
- private static final Logger logger = LoggerFactory.getLogger(HttpGetClient.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(HttpGetClient.class);
private final WebClient webClient;
private final Gson gson;
this(WebClient.builder().filter(logRequest()).filter(logResponse()).build());
}
- HttpGetClient(WebClient webClient){
+ HttpGetClient(WebClient webClient) {
this.webClient = webClient;
this.gson = new Gson();
}
- public <T> Mono<T> callHttpGet(String url, Class<T> tClass) {
+ <T> Mono<T> callHttpGet(String url, Class<T> tClass) {
return webClient
.get()
.uri(url)
.onStatus(HttpStatus::is4xxClientError, response -> Mono.error(getException(response)))
.onStatus(HttpStatus::is5xxServerError, response -> Mono.error(getException(response)))
.bodyToMono(String.class)
- .flatMap(body->getJsonFromRequest(body,tClass));
+ .flatMap(body -> getJsonFromRequest(body, tClass));
}
private RuntimeException getException(ClientResponse response) {
try {
return Mono.just(parseJson(body, tClass));
} catch (JsonSyntaxException | IllegalStateException e) {
- logger.warn("Converting string to json threw error ", e);
return Mono.error(e);
}
}
- private <T> T parseJson(String body, Class<T> tClass){
+ private <T> T parseJson(String body, Class<T> tClass) {
return gson.fromJson(body, tClass);
}
private static ExchangeFilterFunction logResponse() {
return ExchangeFilterFunction.ofResponseProcessor(clientResponse -> {
- logger.info("Response status {}", clientResponse.statusCode());
+ LOGGER.info("Response status {}", clientResponse.statusCode());
return Mono.just(clientResponse);
});
}
private static ExchangeFilterFunction logRequest() {
return ExchangeFilterFunction.ofRequestProcessor(clientRequest -> {
- logger.info("Request: {} {}", clientRequest.method(), clientRequest.url());
+ LOGGER.info("Request: {} {}", clientRequest.method(), clientRequest.url());
clientRequest.headers()
- .forEach((name, values) -> values.forEach(value -> logger.info("{}={}", name, value)));
+ .forEach((name, values) -> values.forEach(value -> LOGGER.info("{}={}", name, value)));
return Mono.just(clientRequest);
});
}
@Service
public class PrhConfigurationProvider {
- private final Logger logger = LoggerFactory.getLogger(this.getClass());
+ private static final Logger LOGGER = LoggerFactory.getLogger(PrhConfigurationProvider.class);
private final HttpGetClient httpGetClient;
}
private Mono<String> callConsulForConfigBindingServiceEndpoint(EnvProperties envProperties) {
- logger.info("Retrieving Config Binding Service endpoint from Consul");
+ LOGGER.info("Retrieving Config Binding Service endpoint from Consul");
try {
return httpGetClient.callHttpGet(getConsulUrl(envProperties), JsonArray.class)
.flatMap(jsonArray -> this.createConfigBindingServiceURL(jsonArray, envProperties.appName()));
} catch (URISyntaxException e) {
- logger.warn("Malformed Consul uri", e);
+ LOGGER.warn("Malformed Consul uri", e);
return Mono.error(e);
}
}
}
private Mono<JsonObject> callConfigBindingServiceForPrhConfiguration(String configBindingServiceUri) {
- logger.info("Retrieving PRH configuration");
+ LOGGER.info("Retrieving PRH configuration");
return httpGetClient.callHttpGet(configBindingServiceUri, JsonObject.class);
}
return Mono.just(getUri(jsonObject.get("ServiceAddress").getAsString(),
jsonObject.get("ServicePort").getAsInt(), "/service_component", appName));
} catch (URISyntaxException e) {
- logger.warn("Malformed Config Binding Service uri", e);
+ LOGGER.warn("Malformed Config Binding Service uri", e);
return Mono.error(e);
}
}
throw new IllegalStateException("JSON Array was empty");
}
} catch (IllegalStateException e) {
- logger.warn("Failed to retrieve JSON Object from array", e);
+ LOGGER.warn("Failed to retrieve JSON Object from array", e);
return Mono.error(e);
}
}
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
+import javax.net.ssl.SSLException;
+
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/13/18
*/
public abstract class AaiProducerTask {
- abstract Mono<ConsumerDmaapModel> publish(Mono<ConsumerDmaapModel> message) throws AaiNotFoundException;
+ abstract Mono<ConsumerDmaapModel> publish(ConsumerDmaapModel message) throws AaiNotFoundException;
- abstract AaiProducerReactiveHttpClient resolveClient();
+ abstract AaiProducerReactiveHttpClient resolveClient() throws SSLException;
protected abstract AaiClientConfiguration resolveConfiguration();
- protected abstract Mono<ConsumerDmaapModel> execute(Mono<ConsumerDmaapModel> consumerDmaapModel)
- throws PrhTaskException;
+ protected abstract Mono<ConsumerDmaapModel> execute(ConsumerDmaapModel consumerDmaapModel)
+ throws PrhTaskException, SSLException;
- WebClient buildWebClient() {
+ WebClient buildWebClient() throws SSLException {
return new AaiReactiveWebClient().fromConfiguration(resolveConfiguration()).build();
}
}
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;
+import javax.net.ssl.SSLException;
+
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/13/18
*/
public class AaiProducerTaskImpl extends
AaiProducerTask {
- private final Logger logger = LoggerFactory.getLogger(this.getClass());
- private final Marker INVOKE = MarkerFactory.getMarker("INVOKE");
-
+ private static final Marker INVOKE = MarkerFactory.getMarker("INVOKE");
+ private static final Logger LOGGER = LoggerFactory.getLogger(AaiProducerTaskImpl.class);
private final Config config;
private AaiProducerReactiveHttpClient aaiProducerReactiveHttpClient;
}
@Override
- Mono<ConsumerDmaapModel> publish(Mono<ConsumerDmaapModel> consumerDmaapModel) {
-
+ Mono<ConsumerDmaapModel> publish(ConsumerDmaapModel consumerDmaapModel) {
+ LOGGER.info("Publish to AAI DmaapModel");
return aaiProducerReactiveHttpClient.getAaiProducerResponse(consumerDmaapModel)
.flatMap(response -> {
- if (HttpUtils.isSuccessfulResponseCode(response)) {
- return consumerDmaapModel;
+ if (HttpUtils.isSuccessfulResponseCode(response.statusCode().value())) {
+ return Mono.just(consumerDmaapModel);
}
return Mono
.error(new AaiNotFoundException("Incorrect response code for continuation of tasks workflow"));
}
@Override
- AaiProducerReactiveHttpClient resolveClient() {
- return new AaiProducerReactiveHttpClient(resolveConfiguration());
+ AaiProducerReactiveHttpClient resolveClient() throws SSLException {
+ return new AaiProducerReactiveHttpClient(resolveConfiguration()).createAaiWebClient(buildWebClient());
}
@Override
}
@Override
- protected Mono<ConsumerDmaapModel> execute(Mono<ConsumerDmaapModel> consumerDmaapModel) throws PrhTaskException {
+ protected Mono<ConsumerDmaapModel> execute(ConsumerDmaapModel consumerDmaapModel)
+ throws PrhTaskException, SSLException {
if (consumerDmaapModel == null) {
throw new DmaapNotFoundException("Invoked null object to DMaaP task");
}
aaiProducerReactiveHttpClient = resolveClient();
- logger.info(INVOKE, "Method called with arg {}", consumerDmaapModel);
+ LOGGER.debug(INVOKE, "Method called with arg {}", consumerDmaapModel);
return publish(consumerDmaapModel);
}
package org.onap.dcaegen2.services.prh.tasks;
import org.onap.dcaegen2.services.prh.config.DmaapConsumerConfiguration;
-import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
import org.onap.dcaegen2.services.prh.service.DMaaPReactiveWebClient;
import org.onap.dcaegen2.services.prh.service.consumer.DMaaPConsumerReactiveHttpClient;
*/
abstract class DmaapConsumerTask {
- abstract Mono<ConsumerDmaapModel> consume(Mono<String> message) throws PrhTaskException;
+ abstract Mono<ConsumerDmaapModel> consume(Mono<String> message);
abstract DMaaPConsumerReactiveHttpClient resolveClient();
protected abstract DmaapConsumerConfiguration resolveConfiguration();
- protected abstract Mono<ConsumerDmaapModel> execute(String object) throws PrhTaskException;
+ protected abstract Mono<ConsumerDmaapModel> execute(String object);
WebClient buildWebClient() {
return new DMaaPReactiveWebClient().fromConfiguration(resolveConfiguration()).build();
package org.onap.dcaegen2.services.prh.tasks;
-import java.util.Map;
import org.onap.dcaegen2.services.prh.config.DmaapConsumerConfiguration;
import org.onap.dcaegen2.services.prh.configuration.AppConfig;
import org.onap.dcaegen2.services.prh.configuration.Config;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.model.logging.MDCVariables;
import org.onap.dcaegen2.services.prh.service.DmaapConsumerJsonParser;
import org.onap.dcaegen2.services.prh.service.consumer.DMaaPConsumerReactiveHttpClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.slf4j.MDC;
import org.slf4j.Marker;
import org.slf4j.MarkerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@Component
public class DmaapConsumerTaskImpl extends DmaapConsumerTask {
- private final Logger logger = LoggerFactory.getLogger(this.getClass());
- private final Marker INVOKE = MarkerFactory.getMarker("INVOKE");
+ private static final Marker INVOKE = MarkerFactory.getMarker("INVOKE");
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(DmaapConsumerTaskImpl.class);
private final Config config;
private DmaapConsumerJsonParser dmaapConsumerJsonParser;
- private DMaaPConsumerReactiveHttpClient dmaaPConsumerReactiveHttpClient;
@Autowired
public DmaapConsumerTaskImpl(Config config) {
@Override
public Mono<ConsumerDmaapModel> execute(String object) {
- dmaaPConsumerReactiveHttpClient = resolveClient();
- logger.info(INVOKE, "Method called with arg {}", object);
+ DMaaPConsumerReactiveHttpClient dmaaPConsumerReactiveHttpClient = resolveClient();
+ LOGGER.debug(INVOKE, "Method called with arg {}", object);
return consume(dmaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse());
}
import org.onap.dcaegen2.services.prh.config.DmaapPublisherConfiguration;
import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.service.DMaaPReactiveWebClient;
import org.onap.dcaegen2.services.prh.service.producer.DMaaPProducerReactiveHttpClient;
-import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.client.RestTemplate;
import reactor.core.publisher.Mono;
/**
*/
abstract class DmaapPublisherTask {
- abstract Mono<String> publish(Mono<ConsumerDmaapModel> consumerDmaapModel) throws PrhTaskException;
+ abstract Mono<ResponseEntity<String>> publish(ConsumerDmaapModel consumerDmaapModel) throws PrhTaskException;
abstract DMaaPProducerReactiveHttpClient resolveClient();
protected abstract DmaapPublisherConfiguration resolveConfiguration();
- protected abstract Mono<String> execute(Mono<ConsumerDmaapModel> consumerDmaapModel) throws PrhTaskException;
+ protected abstract Mono<ResponseEntity<String>> execute(ConsumerDmaapModel consumerDmaapModel)
+ throws PrhTaskException;
- WebClient buildWebClient() {
- return new DMaaPReactiveWebClient().fromConfiguration(resolveConfiguration()).build();
- }
+ abstract RestTemplate buildWebClient();
}
import org.slf4j.Marker;
import org.slf4j.MarkerFactory;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
+import org.springframework.web.client.RestTemplate;
import reactor.core.publisher.Mono;
/**
@Component
public class DmaapPublisherTaskImpl extends DmaapPublisherTask {
- private final Logger logger = LoggerFactory.getLogger(this.getClass());
- private final Marker INVOKE = MarkerFactory.getMarker("INVOKE");
+ private static final Logger LOGGER = LoggerFactory.getLogger(DmaapPublisherTaskImpl.class);
+ private static final Marker INVOKE = MarkerFactory.getMarker("INVOKE");
private final Config config;
private DMaaPProducerReactiveHttpClient dmaapProducerReactiveHttpClient;
}
@Override
- Mono<String> publish(Mono<ConsumerDmaapModel> consumerDmaapModel) {
- return consumerDmaapModel.flatMap(dmaapModel -> {
- logger.info("Publishing on DMaaP topic {} object {}", resolveConfiguration().dmaapTopicName(),
- dmaapModel);
- return dmaapProducerReactiveHttpClient.getDMaaPProducerResponse(dmaapModel);
- });
+ Mono<ResponseEntity<String>> publish(ConsumerDmaapModel consumerDmaapModel) {
+ return dmaapProducerReactiveHttpClient.getDMaaPProducerResponse(consumerDmaapModel);
}
@Override
- public Mono<String> execute(Mono<ConsumerDmaapModel> consumerDmaapModel) throws DmaapNotFoundException {
+ public Mono<ResponseEntity<String>> execute(ConsumerDmaapModel consumerDmaapModel) throws DmaapNotFoundException {
if (consumerDmaapModel == null) {
throw new DmaapNotFoundException("Invoked null object to DMaaP task");
}
dmaapProducerReactiveHttpClient = resolveClient();
- logger.info(INVOKE, "Method called with arg {}", consumerDmaapModel);
+ LOGGER.info(INVOKE, "Method called with arg {}", consumerDmaapModel);
return publish(consumerDmaapModel);
}
+ @Override
+ RestTemplate buildWebClient() {
+ return new RestTemplate();
+ }
+
@Override
protected DmaapPublisherConfiguration resolveConfiguration() {
return config.getDmaapPublisherConfiguration();
import java.util.Map;
import java.util.UUID;
-import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import javax.net.ssl.SSLException;
import org.onap.dcaegen2.services.prh.exceptions.DmaapEmptyResponseException;
import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
-import org.slf4j.Marker;
-import org.slf4j.MarkerFactory;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18
@Component
public class ScheduledTasks {
- private final Logger logger = LoggerFactory.getLogger(this.getClass());
+ private static final Logger logger = LoggerFactory.getLogger(ScheduledTasks.class);
+
private final DmaapConsumerTask dmaapConsumerTask;
private final DmaapPublisherTask dmaapProducerTask;
private final AaiProducerTask aaiProducerTask;
*/
public void scheduleMainPrhEventTask() {
MDCVariables.setMdcContextMap(contextMap);
- logger.trace("Execution of tasks was registered");
-
- Mono<String> dmaapProducerResponse = Mono.fromCallable(consumeFromDMaaPMessage())
- .doOnError(DmaapEmptyResponseException.class, error -> logger.warn("Nothing to consume from DMaaP"))
- .map(this::publishToAaiConfiguration)
- .flatMap(this::publishToDmaapConfiguration)
- .subscribeOn(Schedulers.elastic());
+ try {
+ logger.trace("Execution of tasks was registered");
+ CountDownLatch mainCountDownLatch = new CountDownLatch(1);
+ consumeFromDMaaPMessage()
+ .doOnError(DmaapEmptyResponseException.class, error ->
+ logger.warn("Nothing to consume from DMaaP")
+ )
+ .flatMap(this::publishToAaiConfiguration)
+ .flatMap(this::publishToDmaapConfiguration)
+ .doOnTerminate(mainCountDownLatch::countDown)
+ .subscribe(this::onSuccess, this::onError, this::onComplete);
- dmaapProducerResponse.subscribe(this::onSuccess, this::onError, this::onComplete);
+ mainCountDownLatch.await();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
}
+
private void onComplete() {
logger.info("PRH tasks have been completed");
}
- private void onSuccess(String responseCode) {
- MDC.put(RESPONSE_CODE, responseCode);
- logger.info("Prh consumed tasks. HTTP Response code {}", responseCode);
+ private void onSuccess(ResponseEntity<String> responseCode) {
+ MDC.put(RESPONSE_CODE, responseCode.getStatusCode().toString());
+ logger.info("Prh consumed tasks successfully. HTTP Response code from DMaaPProducer {}",
+ responseCode.getStatusCode().value());
}
private void onError(Throwable throwable) {
}
}
- private Callable<Mono<ConsumerDmaapModel>> consumeFromDMaaPMessage() {
- return () -> {
+
+ private Mono<ConsumerDmaapModel> consumeFromDMaaPMessage() {
+ return Mono.defer(() -> {
MDCVariables.setMdcContextMap(contextMap);
MDC.put(INSTANCE_UUID, UUID.randomUUID().toString());
+ logger.info("Init configs");
dmaapConsumerTask.initConfigs();
return dmaapConsumerTask.execute("");
- };
+ });
}
- private Mono<ConsumerDmaapModel> publishToAaiConfiguration(Mono<ConsumerDmaapModel> monoDMaaPModel) {
+ private Mono<ConsumerDmaapModel> publishToAaiConfiguration(ConsumerDmaapModel monoDMaaPModel) {
try {
return aaiProducerTask.execute(monoDMaaPModel);
- } catch (PrhTaskException e) {
+ } catch (PrhTaskException | SSLException e) {
return Mono.error(e);
}
}
- private Mono<String> publishToDmaapConfiguration(Mono<ConsumerDmaapModel> monoAaiModel) {
+ private Mono<ResponseEntity<String>> publishToDmaapConfiguration(ConsumerDmaapModel monoAaiModel) {
try {
return dmaapProducerTask.execute(monoAaiModel);
} catch (PrhTaskException e) {
logging.level.org.springframework=ERROR
logging.level.org.springframework.data=ERROR
logging.level.org.onap.dcaegen2.services.prh=INFO
+logging.level.org.springframework.web.reactive=WARN
+logging.level.reactor.ipc.netty.http.client=WARN
app.filepath=config/prh_endpoints.json
app.xonaprequestid=requestID
app.xinvocationid=invocationID
+
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
+import javax.net.ssl.SSLException;
import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.function.Executable;
import org.onap.dcaegen2.services.prh.config.AaiClientConfiguration;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
import org.onap.dcaegen2.services.prh.model.ImmutableConsumerDmaapModel;
import org.onap.dcaegen2.services.prh.service.producer.AaiProducerReactiveHttpClient;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.reactive.function.client.ClientResponse;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
private static final String BASE_PATH = "/aai/v11";
private static final String PNF_PATH = "/network/pnfs/pnf";
- private static ConsumerDmaapModel consumerDmaapModel;
- private static AaiProducerTaskImpl aaiProducerTask;
- private static AaiClientConfiguration aaiClientConfiguration;
- private static AaiProducerReactiveHttpClient aaiProducerReactiveHttpClient;
- private static AppConfig appConfig;
+ private ConsumerDmaapModel consumerDmaapModel;
+ private AaiProducerTaskImpl aaiProducerTask;
+ private AaiClientConfiguration aaiClientConfiguration;
+ private AaiProducerReactiveHttpClient aaiProducerReactiveHttpClient;
+ private AppConfig appConfig;
+ private ClientResponse clientResponse;
- @BeforeAll
- static void setUp() {
+ @BeforeEach
+ void setUp() {
+ clientResponse = mock(ClientResponse.class);
aaiClientConfiguration = new ImmutableAaiClientConfiguration.Builder()
.aaiHost(AAI_HOST)
.aaiPort(PORT)
}
- private static void getAaiProducerTask_whenMockingResponseObject(Integer statusCode) {
- //given
- aaiProducerReactiveHttpClient = mock(AaiProducerReactiveHttpClient.class);
- when(aaiProducerReactiveHttpClient.getAaiProducerResponse(any()))
- .thenReturn(Mono.just(statusCode));
- when(appConfig.getAaiClientConfiguration()).thenReturn(aaiClientConfiguration);
- aaiProducerTask = spy(new AaiProducerTaskImpl(appConfig));
- when(aaiProducerTask.resolveConfiguration()).thenReturn(aaiClientConfiguration);
- doReturn(aaiProducerReactiveHttpClient).when(aaiProducerTask).resolveClient();
- }
-
@Test
void whenPassedObjectDoesntFit_ThrowsPrhTaskException() {
//given/when/
}
@Test
- void whenPassedObjectFits_ReturnsCorrectStatus() throws PrhTaskException {
+ void whenPassedObjectFits_ReturnsCorrectStatus() throws PrhTaskException, SSLException {
//given/when
getAaiProducerTask_whenMockingResponseObject(200);
- Mono<ConsumerDmaapModel> response = aaiProducerTask.execute(Mono.just(consumerDmaapModel));
+ Mono<ConsumerDmaapModel> response = aaiProducerTask.execute(consumerDmaapModel);
//then
verify(aaiProducerReactiveHttpClient, times(1)).getAaiProducerResponse(any());
}
@Test
- void whenPassedObjectFits_butIncorrectResponseReturns() throws PrhTaskException {
+ void whenPassedObjectFits_butIncorrectResponseReturns() throws PrhTaskException, SSLException {
//given/when
getAaiProducerTask_whenMockingResponseObject(400);
- StepVerifier.create(aaiProducerTask.execute(Mono.just(consumerDmaapModel))).expectSubscription()
+ StepVerifier.create(aaiProducerTask.execute(consumerDmaapModel)).expectSubscription()
.expectError(PrhTaskException.class).verify();
//then
verify(aaiProducerReactiveHttpClient, times(1)).getAaiProducerResponse(any());
verifyNoMoreInteractions(aaiProducerReactiveHttpClient);
}
+
+ private void getAaiProducerTask_whenMockingResponseObject(int statusCode) throws SSLException {
+ //given
+ doReturn(HttpStatus.valueOf(statusCode)).when(clientResponse).statusCode();
+ Mono<ClientResponse> clientResponseMono = Mono.just(clientResponse);
+ aaiProducerReactiveHttpClient = mock(AaiProducerReactiveHttpClient.class);
+ when(aaiProducerReactiveHttpClient.getAaiProducerResponse(any()))
+ .thenReturn(clientResponseMono);
+ when(appConfig.getAaiClientConfiguration()).thenReturn(aaiClientConfiguration);
+ aaiProducerTask = spy(new AaiProducerTaskImpl(appConfig));
+ when(aaiProducerTask.resolveConfiguration()).thenReturn(aaiClientConfiguration);
+ doReturn(aaiProducerReactiveHttpClient).when(aaiProducerTask).resolveClient();
+ }
}
\ No newline at end of file
package org.onap.dcaegen2.services.prh.tasks;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
-
import org.onap.dcaegen2.services.prh.config.AaiClientConfiguration;
import org.onap.dcaegen2.services.prh.configuration.AppConfig;
import org.onap.dcaegen2.services.prh.service.producer.AaiProducerReactiveHttpClient;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
+import javax.net.ssl.SSLException;
+
+import static org.mockito.Mockito.*;
+
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/13/18
*/
*/
@Bean
@Primary
- public AaiProducerTask registerSimpleAaiPublisherTask() {
+ public AaiProducerTask registerSimpleAaiPublisherTask() throws SSLException {
AppConfig appConfig = spy(AppConfig.class);
doReturn(mock(AaiClientConfiguration.class)).when(appConfig).getAaiClientConfiguration();
AaiProducerTaskImpl aaiProducerTask = spy(new AaiProducerTaskImpl(appConfig));
package org.onap.dcaegen2.services.prh.tasks;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-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.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
-import static org.mockito.Mockito.when;
-
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.function.Executable;
import org.onap.dcaegen2.services.prh.model.ImmutableConsumerDmaapModel;
import org.onap.dcaegen2.services.prh.service.producer.DMaaPProducerReactiveHttpClient;
import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/17/18
*/
@Test
void whenPassedObjectFits_ReturnsCorrectStatus() throws PrhTaskException {
//given
- prepareMocksForTests(HttpStatus.OK.value());
+ ResponseEntity<String> responseEntity = prepareMocksForTests(HttpStatus.OK.value());
//when
- StepVerifier.create(dmaapPublisherTask.execute(Mono.just(consumerDmaapModel))).expectSubscription()
- .expectNext(HttpStatus.OK.toString()).verifyComplete();
+ when(responseEntity.getStatusCode()).thenReturn(HttpStatus.OK);
+ StepVerifier.create(dmaapPublisherTask.execute(consumerDmaapModel)).expectSubscription()
+ .expectNext(responseEntity).verifyComplete();
//then
verify(dMaaPProducerReactiveHttpClient, times(1))
- .getDMaaPProducerResponse(any());
+ .getDMaaPProducerResponse(consumerDmaapModel);
verifyNoMoreInteractions(dMaaPProducerReactiveHttpClient);
}
@Test
void whenPassedObjectFits_butIncorrectResponseReturns() throws DmaapNotFoundException {
//given
- prepareMocksForTests(HttpStatus.UNAUTHORIZED.value());
+ ResponseEntity<String> responseEntity = prepareMocksForTests(HttpStatus.UNAUTHORIZED.value());
//when
- StepVerifier.create(dmaapPublisherTask.execute(Mono.just(consumerDmaapModel))).expectSubscription()
- .expectNext(String.valueOf(HttpStatus.UNAUTHORIZED.value())).verifyComplete();
+ when(responseEntity.getStatusCode()).thenReturn(HttpStatus.UNAUTHORIZED);
+ StepVerifier.create(dmaapPublisherTask.execute(consumerDmaapModel)).expectSubscription()
+ .expectNext(responseEntity).verifyComplete();
//then
- verify(dMaaPProducerReactiveHttpClient, times(1)).getDMaaPProducerResponse(any());
+ verify(dMaaPProducerReactiveHttpClient, times(1))
+ .getDMaaPProducerResponse(consumerDmaapModel);
verifyNoMoreInteractions(dMaaPProducerReactiveHttpClient);
}
- private void prepareMocksForTests(Integer httpResponseCode) {
+ private ResponseEntity<String> prepareMocksForTests(Integer httpResponseCode) {
+ ResponseEntity<String> responseEntity = mock(ResponseEntity.class);
+ //when
+ when(responseEntity.getStatusCode()).thenReturn(HttpStatus.valueOf(httpResponseCode));
dMaaPProducerReactiveHttpClient = mock(DMaaPProducerReactiveHttpClient.class);
when(dMaaPProducerReactiveHttpClient.getDMaaPProducerResponse(any()))
- .thenReturn(Mono.just(httpResponseCode.toString()));
+ .thenReturn(Mono.just(responseEntity));
dmaapPublisherTask = spy(new DmaapPublisherTaskImpl(appConfig));
when(dmaapPublisherTask.resolveConfiguration()).thenReturn(dmaapPublisherConfiguration);
doReturn(dMaaPProducerReactiveHttpClient).when(dmaapPublisherTask).resolveClient();
+ return responseEntity;
}
}
\ No newline at end of file
return gsonBuilder.create().toJson(ImmutableConsumerDmaapModel.builder().ipv4(consumerDmaapModel.getIpv4())
.ipv6(consumerDmaapModel.getIpv6()).sourceName(consumerDmaapModel.getSourceName()).build());
}
-}
+}
\ No newline at end of file
package org.onap.dcaegen2.services.prh.service;
-import java.util.HashMap;
-import java.util.Map;
import org.onap.dcaegen2.services.prh.config.DmaapCustomConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
-import org.springframework.http.HttpHeaders;
import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
private String dmaaPUserName;
private String dmaaPUserPassword;
- private String dmaaPContentType;
/**
* Creating DMaaPReactiveWebClient passing to them basic DMaaPConfig.
public DMaaPReactiveWebClient fromConfiguration(DmaapCustomConfig dmaapCustomConfig) {
this.dmaaPUserName = dmaapCustomConfig.dmaapUserName();
this.dmaaPUserPassword = dmaapCustomConfig.dmaapUserPassword();
- this.dmaaPContentType = dmaapCustomConfig.dmaapContentType();
-
return this;
}
*/
public WebClient build() {
return WebClient.builder()
- .defaultHeader(HttpHeaders.CONTENT_TYPE, dmaaPContentType)
.filter(logRequest())
.filter(logResponse())
.build();
import java.util.UUID;
import org.apache.http.client.utils.URIBuilder;
import org.onap.dcaegen2.services.prh.config.DmaapConsumerConfiguration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
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 Logger logger = LoggerFactory.getLogger(this.getClass());
private final String dmaapHostName;
private final String dmaapProtocol;
private final Integer dmaapPortNumber;
private final String dmaapTopicName;
private final String consumerGroup;
private final String consumerId;
+ private final String contentType;
private WebClient webClient;
/**
this.dmaapTopicName = consumerConfiguration.dmaapTopicName();
this.consumerGroup = consumerConfiguration.consumerGroup();
this.consumerId = consumerConfiguration.consumerId();
+ this.contentType = consumerConfiguration.dmaapContentType();
}
/**
.uri(getUri())
.header(X_ONAP_REQUEST_ID, MDC.get(REQUEST_ID))
.header(X_INVOCATION_ID, UUID.randomUUID().toString())
+ .header(HttpHeaders.CONTENT_TYPE, contentType)
.retrieve()
.onStatus(HttpStatus::is4xxClientError, clientResponse ->
- Mono.error(new Exception("DmaaPConsumer HTTP " + clientResponse.statusCode()))
+ Mono.error(new RuntimeException("DmaaPConsumer HTTP " + clientResponse.statusCode()))
)
.onStatus(HttpStatus::is5xxServerError, clientResponse ->
- Mono.error(new Exception("DmaaPConsumer HTTP " + clientResponse.statusCode())))
+ Mono.error(new RuntimeException("DmaaPConsumer HTTP " + clientResponse.statusCode())))
.bodyToMono(String.class);
} catch (URISyntaxException e) {
- logger.warn("Exception while evaluating URI ");
return Mono.error(e);
}
}
package org.onap.dcaegen2.services.prh.service.producer;
+import static org.onap.dcaegen2.services.prh.model.CommonFunctions.createJsonBody;
import static org.onap.dcaegen2.services.prh.model.logging.MDCVariables.REQUEST_ID;
import static org.onap.dcaegen2.services.prh.model.logging.MDCVariables.X_INVOCATION_ID;
import static org.onap.dcaegen2.services.prh.model.logging.MDCVariables.X_ONAP_REQUEST_ID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
-import org.springframework.http.HttpStatus;
-import org.springframework.web.reactive.function.BodyInserters;
-import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.client.RestTemplate;
import reactor.core.publisher.Mono;
/**
public class DMaaPProducerReactiveHttpClient {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
private final String dmaapHostName;
private final Integer dmaapPortNumber;
private final String dmaapProtocol;
private final String dmaapTopicName;
- private WebClient webClient;
+ private final String dmaapContentType;
+ private RestTemplate restTemplate;
/**
* Constructor DMaaPProducerReactiveHttpClient.
this.dmaapProtocol = dmaapPublisherConfiguration.dmaapProtocol();
this.dmaapPortNumber = dmaapPublisherConfiguration.dmaapPortNumber();
this.dmaapTopicName = dmaapPublisherConfiguration.dmaapTopicName();
+ this.dmaapContentType = dmaapPublisherConfiguration.dmaapContentType();
}
/**
* @param consumerDmaapModelMono - object which will be sent to DMaaP
* @return status code of operation
*/
- public Mono<String> getDMaaPProducerResponse(ConsumerDmaapModel consumerDmaapModelMono) {
- try {
- return webClient
- .post()
- .uri(getUri())
- .header(X_ONAP_REQUEST_ID, MDC.get(REQUEST_ID))
- .header(X_INVOCATION_ID, UUID.randomUUID().toString())
- .body(BodyInserters.fromObject(consumerDmaapModelMono))
- .retrieve()
- .onStatus(HttpStatus::is4xxClientError, clientResponse ->
- Mono.error(new Exception("DmaapProducer HTTP" + clientResponse.statusCode()))
- )
- .onStatus(HttpStatus::is5xxServerError, clientResponse ->
- Mono.error(new Exception("DmaapProducer HTTP " + clientResponse.statusCode())))
- .bodyToMono(String.class);
- } catch (URISyntaxException e) {
- logger.warn("Exception while evaluating URI");
- return Mono.error(e);
- }
+
+ public Mono<ResponseEntity<String>> getDMaaPProducerResponse(ConsumerDmaapModel consumerDmaapModelMono) {
+ return Mono.defer(() -> {
+ try {
+ HttpEntity<String> request = new HttpEntity<>(createJsonBody(consumerDmaapModelMono), getAllHeaders());
+ return Mono.just(restTemplate.exchange(getUri(), HttpMethod.POST, request, String.class));
+ } catch (URISyntaxException e) {
+ logger.warn("Exception while evaluating URI");
+ return Mono.error(e);
+ }
+ });
+ }
+
+ private HttpHeaders getAllHeaders() {
+ HttpHeaders headers = new HttpHeaders();
+ headers.set(X_ONAP_REQUEST_ID, MDC.get(REQUEST_ID));
+ headers.set(X_INVOCATION_ID, UUID.randomUUID().toString());
+ headers.set(HttpHeaders.CONTENT_TYPE, dmaapContentType);
+ return headers;
+
}
- public DMaaPProducerReactiveHttpClient createDMaaPWebClient(WebClient webClient) {
- this.webClient = webClient;
+ public DMaaPProducerReactiveHttpClient createDMaaPWebClient(RestTemplate restTemplate) {
+ this.restTemplate = restTemplate;
return this;
}
import java.net.URI;
import java.net.URISyntaxException;
+
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
.expectError(Exception.class).verify();
}
+ @Test
+ void getAppropriateUri_whenPassingCorrectedPathForPnf() throws URISyntaxException {
+ Assertions.assertEquals(dmaapConsumerReactiveHttpClient.getUri(),
+ URI.create("https://54.45.33.2:1234/unauthenticated.SEC_OTHER_OUTPUT/OpenDCAE-c12/c12"));
+ }
+
private void mockDependantObjects() {
when(webClient.get()).thenReturn(requestHeadersSpec);
when(requestHeadersSpec.uri((URI) any())).thenReturn(requestHeadersSpec);
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.net.URISyntaxException;
+
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.onap.dcaegen2.services.prh.config.DmaapPublisherConfiguration;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModelForUnitTest;
-import org.springframework.http.HttpHeaders;
-import org.springframework.web.reactive.function.client.WebClient;
-import org.springframework.web.reactive.function.client.WebClient.RequestBodyUriSpec;
-import org.springframework.web.reactive.function.client.WebClient.RequestHeadersSpec;
-import org.springframework.web.reactive.function.client.WebClient.ResponseSpec;
-import reactor.core.publisher.Mono;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.client.RestTemplate;
import reactor.test.StepVerifier;
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 7/4/18
*/
+
class DMaaPProducerReactiveHttpClientTest {
private DMaaPProducerReactiveHttpClient dmaapProducerReactiveHttpClient;
private DmaapPublisherConfiguration dmaapPublisherConfigurationMock = mock(
DmaapPublisherConfiguration.class);
private ConsumerDmaapModel consumerDmaapModel = new ConsumerDmaapModelForUnitTest();
- private WebClient webClient = mock(WebClient.class);
- private RequestBodyUriSpec requestBodyUriSpec;
- private ResponseSpec responseSpec;
@BeforeEach
when(dmaapPublisherConfigurationMock.dmaapUserName()).thenReturn("PRH");
when(dmaapPublisherConfigurationMock.dmaapUserPassword()).thenReturn("PRH");
when(dmaapPublisherConfigurationMock.dmaapContentType()).thenReturn("application/json");
- when(dmaapPublisherConfigurationMock.dmaapTopicName()).thenReturn("pnfReady");
-
+ when(dmaapPublisherConfigurationMock.dmaapTopicName()).thenReturn("unauthenticated.PNF_READY");
dmaapProducerReactiveHttpClient = new DMaaPProducerReactiveHttpClient(dmaapPublisherConfigurationMock);
- webClient = spy(WebClient.builder()
- .defaultHeader(HttpHeaders.CONTENT_TYPE, dmaapPublisherConfigurationMock.dmaapContentType())
- .filter(basicAuthentication(dmaapPublisherConfigurationMock.dmaapUserName(),
- dmaapPublisherConfigurationMock.dmaapUserPassword()))
- .build());
- requestBodyUriSpec = mock(RequestBodyUriSpec.class);
- responseSpec = mock(ResponseSpec.class);
}
@Test
void getHttpResponse_Success() {
//given
- Integer responseSuccess = 200;
- Mono<Integer> expectedResult = Mono.just(responseSuccess);
-
+ int responseSuccess = 200;
+ ResponseEntity<String> mockedResponseEntity = mock(ResponseEntity.class);
+ RestTemplate restTemplate = mock(RestTemplate.class);
//when
- mockWebClientDependantObject();
- doReturn(expectedResult).when(responseSpec).bodyToMono(String.class);
- dmaapProducerReactiveHttpClient.createDMaaPWebClient(webClient);
- Mono<String> response = dmaapProducerReactiveHttpClient.getDMaaPProducerResponse(consumerDmaapModel);
+ when(mockedResponseEntity.getStatusCode()).thenReturn(HttpStatus.valueOf(responseSuccess));
+ doReturn(mockedResponseEntity).when(restTemplate)
+ .exchange(any(URI.class), any(HttpMethod.class), any(HttpEntity.class), (Class<Object>) any());
+ dmaapProducerReactiveHttpClient.createDMaaPWebClient(restTemplate);
//then
- Assertions.assertEquals(response.block(), expectedResult.block());
+ StepVerifier.create(dmaapProducerReactiveHttpClient.getDMaaPProducerResponse(consumerDmaapModel))
+ .expectSubscription().expectNext(mockedResponseEntity).verifyComplete();
}
@Test
//given
dmaapProducerReactiveHttpClient = spy(dmaapProducerReactiveHttpClient);
//when
- when(webClient.post()).thenReturn(requestBodyUriSpec);
- dmaapProducerReactiveHttpClient.createDMaaPWebClient(webClient);
when(dmaapProducerReactiveHttpClient.getUri()).thenThrow(URISyntaxException.class);
//then
.expectError(Exception.class).verify();
}
- private void mockWebClientDependantObject() {
- RequestHeadersSpec requestHeadersSpec = mock(RequestHeadersSpec.class);
- when(webClient.post()).thenReturn(requestBodyUriSpec);
- when(requestBodyUriSpec.uri((URI) any())).thenReturn(requestBodyUriSpec);
- when(requestBodyUriSpec.header(any(), any())).thenReturn(requestBodyUriSpec);
- when(requestBodyUriSpec.body(any())).thenReturn(requestHeadersSpec);
- doReturn(responseSpec).when(requestHeadersSpec).retrieve();
- doReturn(responseSpec).when(responseSpec).onStatus(any(), any());
+ @Test
+ void getAppropriateUri_whenPassingCorrectedPathForPnf() throws URISyntaxException {
+ Assertions.assertEquals(dmaapProducerReactiveHttpClient.getUri(),
+ URI.create("https://54.45.33.2:1234/unauthenticated.PNF_READY"));
}
}
\ No newline at end of file
Code Review / dcaegen2 / services / prh.git / commitdiff