summaryrefslogtreecommitdiffstats
path: root/prh-app-server/src
diff options
context:
space:
mode:
authorwasala <przemyslaw.wasala@nokia.com>2018-07-04 14:12:23 +0200
committerwasala <przemyslaw.wasala@nokia.com>2018-08-07 09:45:22 +0200
commit3d1f84c127e2244a42d5e02d4c50f8e9f06000d1 (patch)
treea9a661e395404c2d0f8203041a7ab6c5f54acd50 /prh-app-server/src
parent79984d737c71d3c92f3cd283eaf2b9b6157c2ce2 (diff)
Added reactive DMaaPClient
Extracted WebCLientBuilder for Producer and Consumer. Added unit test for ReactiveProducerClient. Change-Id: I632e6928813ed9feb48982900c173f741e4483e3 Issue-ID: DCAEGEN2-563 Signed-off-by: wasala <przemyslaw.wasala@nokia.com>
Diffstat (limited to 'prh-app-server/src')
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java4
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java28
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTask.java9
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java34
-rw-r--r--prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java14
-rw-r--r--prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImplTest.java19
-rw-r--r--prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskSpy.java8
-rw-r--r--prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapProducerTaskSpy.java7
-rw-r--r--prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java38
9 files changed, 84 insertions, 77 deletions
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java
index 5cd30f8b..6db36a8a 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java
@@ -22,7 +22,7 @@ package org.onap.dcaegen2.services.prh.tasks;
import java.util.Optional;
import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.service.consumer.DmaapConsumerReactiveHttpClient;
+import org.onap.dcaegen2.services.prh.service.consumer.DMaaPConsumerReactiveHttpClient;
import reactor.core.publisher.Mono;
/**
@@ -32,7 +32,7 @@ abstract class DmaapConsumerTask {
abstract Mono<ConsumerDmaapModel> consume(Mono<String> message) throws PrhTaskException;
- abstract DmaapConsumerReactiveHttpClient resolveClient();
+ abstract DMaaPConsumerReactiveHttpClient resolveClient();
abstract void initConfigs();
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java
index 08008f0a..90382e51 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java
@@ -23,11 +23,10 @@ import java.util.Optional;
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.exceptions.DmaapEmptyResponseException;
-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.DmaapConsumerJsonParser;
-import org.onap.dcaegen2.services.prh.service.consumer.DmaapConsumerReactiveHttpClient;
+import org.onap.dcaegen2.services.prh.service.consumer.DMaaPConsumerReactiveHttpClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -43,7 +42,7 @@ public class DmaapConsumerTaskImpl extends DmaapConsumerTask {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
private final Config prhAppConfig;
private DmaapConsumerJsonParser dmaapConsumerJsonParser;
- private DmaapConsumerReactiveHttpClient dmaapConsumerReactiveHttpClient;
+ private DMaaPConsumerReactiveHttpClient dMaaPConsumerReactiveHttpClient;
@Autowired
public DmaapConsumerTaskImpl(AppConfig prhAppConfig) {
@@ -58,16 +57,15 @@ public class DmaapConsumerTaskImpl extends DmaapConsumerTask {
@Override
Mono<ConsumerDmaapModel> consume(Mono<String> message) {
- logger.info("Consumed model from DmaaP: {}", message);
+ logger.info("Consumed model from DMaaP: {}", message);
return dmaapConsumerJsonParser.getJsonObject(message);
}
@Override
public Mono<ConsumerDmaapModel> execute(String object) {
- dmaapConsumerReactiveHttpClient = resolveClient();
- dmaapConsumerReactiveHttpClient.initWebClient();
+ dMaaPConsumerReactiveHttpClient = resolveClient();
logger.trace("Method called with arg {}", object);
- return consume((dmaapConsumerReactiveHttpClient.getDmaaPConsumerResponse()));
+ return consume((dMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse()));
}
@Override
@@ -80,8 +78,16 @@ public class DmaapConsumerTaskImpl extends DmaapConsumerTask {
}
@Override
- DmaapConsumerReactiveHttpClient resolveClient() {
- return Optional.ofNullable(dmaapConsumerReactiveHttpClient)
- .orElseGet(() -> new DmaapConsumerReactiveHttpClient(resolveConfiguration()));
+ DMaaPConsumerReactiveHttpClient resolveClient() {
+
+ return Optional.ofNullable(dMaaPConsumerReactiveHttpClient)
+ .orElseGet(() -> {
+ DmaapConsumerConfiguration dmaapConsumerConfiguration = resolveConfiguration();
+ return new DMaaPConsumerReactiveHttpClient(dmaapConsumerConfiguration).createDMaaPWebClient(
+ new DMaaPReactiveWebClient.WebClientBuilder()
+ .dmaapContentType(dmaapConsumerConfiguration.dmaapContentType())
+ .dmaapUserName(dmaapConsumerConfiguration.dmaapUserName())
+ .dmaapUserPassword(dmaapConsumerConfiguration.dmaapUserPassword()).build());
+ });
}
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTask.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTask.java
index 3520d134..27670e29 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTask.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTask.java
@@ -21,16 +21,17 @@ package org.onap.dcaegen2.services.prh.tasks;
import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.service.producer.ExtendedDmaapProducerHttpClientImpl;
+import org.onap.dcaegen2.services.prh.service.producer.DMaaPProducerReactiveHttpClient;
+import reactor.core.publisher.Mono;
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18
*/
abstract class DmaapPublisherTask {
- abstract Integer publish(ConsumerDmaapModel consumerDmaapModel) throws PrhTaskException;
+ abstract Mono<Integer> publish(Mono<ConsumerDmaapModel> consumerDmaapModel) throws PrhTaskException;
- abstract ExtendedDmaapProducerHttpClientImpl resolveClient();
+ abstract DMaaPProducerReactiveHttpClient resolveClient();
- protected abstract Integer execute(ConsumerDmaapModel consumerDmaapModel) throws PrhTaskException;
+ protected abstract Mono<Integer> execute(Mono<ConsumerDmaapModel> consumerDmaapModel) throws PrhTaskException;
}
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java
index 7cbeb3b3..faf43bc7 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java
@@ -25,12 +25,13 @@ import org.onap.dcaegen2.services.prh.configuration.AppConfig;
import org.onap.dcaegen2.services.prh.configuration.Config;
import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.service.producer.ExtendedDmaapProducerHttpClientImpl;
+import org.onap.dcaegen2.services.prh.service.DMaaPReactiveWebClient;
+import org.onap.dcaegen2.services.prh.service.producer.DMaaPProducerReactiveHttpClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;
+import reactor.core.publisher.Mono;
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/13/18
@@ -40,7 +41,7 @@ public class DmaapPublisherTaskImpl extends DmaapPublisherTask {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
private final Config prhAppConfig;
- private ExtendedDmaapProducerHttpClientImpl extendedDmaapProducerHttpClient;
+ private DMaaPProducerReactiveHttpClient dMaaPProducerReactiveHttpClient;
@Autowired
public DmaapPublisherTaskImpl(AppConfig prhAppConfig) {
@@ -48,19 +49,17 @@ public class DmaapPublisherTaskImpl extends DmaapPublisherTask {
}
@Override
- Integer publish(ConsumerDmaapModel consumerDmaapModel) throws DmaapNotFoundException {
- logger.info("Publishing on DmaaP topic {} object {}", resolveConfiguration().dmaapTopicName(),
+ Mono<Integer> publish(Mono<ConsumerDmaapModel> consumerDmaapModel) {
+ logger.info("Publishing on DMaaP topic {} object {}", resolveConfiguration().dmaapTopicName(),
consumerDmaapModel);
- return extendedDmaapProducerHttpClient.getHttpProducerResponse(consumerDmaapModel)
- .filter(response -> response == HttpStatus.OK.value())
- .orElseThrow(() -> new DmaapNotFoundException("Incorrect response from Dmaap"));
+ return dMaaPProducerReactiveHttpClient.getDMaaPProducerResponse(consumerDmaapModel).map(Integer::parseInt);
}
@Override
- public Integer execute(ConsumerDmaapModel consumerDmaapModel) throws DmaapNotFoundException {
+ public Mono<Integer> execute(Mono<ConsumerDmaapModel> consumerDmaapModel) throws DmaapNotFoundException {
consumerDmaapModel = Optional.ofNullable(consumerDmaapModel)
- .orElseThrow(() -> new DmaapNotFoundException("Invoked null object to Dmaap task"));
- extendedDmaapProducerHttpClient = resolveClient();
+ .orElseThrow(() -> new DmaapNotFoundException("Invoked null object to DMaaP task"));
+ dMaaPProducerReactiveHttpClient = resolveClient();
logger.trace("Method called with arg {}", consumerDmaapModel);
return publish(consumerDmaapModel);
}
@@ -70,8 +69,15 @@ public class DmaapPublisherTaskImpl extends DmaapPublisherTask {
}
@Override
- ExtendedDmaapProducerHttpClientImpl resolveClient() {
- return Optional.ofNullable(extendedDmaapProducerHttpClient)
- .orElseGet(() -> new ExtendedDmaapProducerHttpClientImpl(resolveConfiguration()));
+ DMaaPProducerReactiveHttpClient resolveClient() {
+ return Optional.ofNullable(dMaaPProducerReactiveHttpClient)
+ .orElseGet(() -> {
+ DmaapPublisherConfiguration dmaapPublisherConfiguration = resolveConfiguration();
+ return new DMaaPProducerReactiveHttpClient(dmaapPublisherConfiguration).createDMaaPWebClient(
+ new DMaaPReactiveWebClient.WebClientBuilder()
+ .dmaapContentType(dmaapPublisherConfiguration.dmaapContentType())
+ .dmaapUserName(dmaapPublisherConfiguration.dmaapUserName())
+ .dmaapUserPassword(dmaapPublisherConfiguration.dmaapUserPassword()).build());
+ });
}
} \ No newline at end of file
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java
index e161e3c5..2787e64b 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java
@@ -97,13 +97,11 @@ public class ScheduledTasks {
}
private Mono<Integer> publishToDMaaPConfiguration(Mono<ConsumerDmaapModel> monoAAIModel) {
- return monoAAIModel.flatMap(aaiModel -> {
- try {
- return Mono.just(dmaapProducerTask.execute(aaiModel));
- } catch (PrhTaskException e) {
- logger.warn("Exception in DMaaPProducer task ", e);
- return Mono.error(e);
- }
- });
+ try {
+ return dmaapProducerTask.execute(monoAAIModel);
+ } catch (PrhTaskException e) {
+ logger.warn("Exception in DMaaPProducer task ", e);
+ return Mono.error(e);
+ }
}
}
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImplTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImplTest.java
index 71e132c4..c5a9d85a 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImplTest.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImplTest.java
@@ -24,7 +24,6 @@ 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 com.google.gson.JsonElement;
@@ -32,7 +31,6 @@ import com.google.gson.JsonParser;
import java.util.Optional;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.onap.dcaegen2.services.prh.config.DmaapConsumerConfiguration;
@@ -43,7 +41,7 @@ import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
import org.onap.dcaegen2.services.prh.model.ImmutableConsumerDmaapModel;
import org.onap.dcaegen2.services.prh.service.DmaapConsumerJsonParser;
-import org.onap.dcaegen2.services.prh.service.consumer.DmaapConsumerReactiveHttpClient;
+import org.onap.dcaegen2.services.prh.service.consumer.DMaaPConsumerReactiveHttpClient;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
@@ -54,7 +52,7 @@ class DmaapConsumerTaskImplTest {
private static ConsumerDmaapModel consumerDmaapModel;
private static DmaapConsumerTaskImpl dmaapConsumerTask;
- private static DmaapConsumerReactiveHttpClient dmaapConsumerReactiveHttpClient;
+ private static DMaaPConsumerReactiveHttpClient dMaaPConsumerReactiveHttpClient;
private static AppConfig appConfig;
private static DmaapConsumerConfiguration dmaapConsumerConfiguration;
private static String message;
@@ -102,22 +100,21 @@ class DmaapConsumerTaskImplTest {
StepVerifier.create(dmaapConsumerTask.execute("Sample input")).expectSubscription()
.expectError(DmaapEmptyResponseException.class);
- verify(dmaapConsumerReactiveHttpClient, times(1)).getDmaaPConsumerResponse();
+ verify(dMaaPConsumerReactiveHttpClient, times(1)).getDMaaPConsumerResponse();
}
@Test
public void whenPassedObjectFits_ReturnsCorrectResponse() throws PrhTaskException {
//given
prepareMocksForDmaapConsumer(Optional.of(message));
-
//when
Mono<ConsumerDmaapModel> response = dmaapConsumerTask.execute("Sample input");
//then
- verify(dmaapConsumerReactiveHttpClient, times(1)).getDmaaPConsumerResponse();
- Assertions.assertNotNull(response);
+ verify(dMaaPConsumerReactiveHttpClient, times(1)).getDMaaPConsumerResponse();
Assertions.assertEquals(consumerDmaapModel, response.block());
+
}
private void prepareMocksForDmaapConsumer(Optional<String> message) {
@@ -125,11 +122,11 @@ class DmaapConsumerTaskImplTest {
JsonElement jsonElement = new JsonParser().parse(parsed);
Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
.when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
- dmaapConsumerReactiveHttpClient = mock(DmaapConsumerReactiveHttpClient.class);
- when(dmaapConsumerReactiveHttpClient.getDmaaPConsumerResponse()).thenReturn(Mono.just(message.orElse("")));
+ dMaaPConsumerReactiveHttpClient = mock(DMaaPConsumerReactiveHttpClient.class);
+ when(dMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse()).thenReturn(Mono.just(message.orElse("")));
when(appConfig.getDmaapConsumerConfiguration()).thenReturn(dmaapConsumerConfiguration);
dmaapConsumerTask = spy(new DmaapConsumerTaskImpl(appConfig, dmaapConsumerJsonParser));
when(dmaapConsumerTask.resolveConfiguration()).thenReturn(dmaapConsumerConfiguration);
- doReturn(dmaapConsumerReactiveHttpClient).when(dmaapConsumerTask).resolveClient();
+ doReturn(dMaaPConsumerReactiveHttpClient).when(dmaapConsumerTask).resolveClient();
}
} \ No newline at end of file
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskSpy.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskSpy.java
index 225dd3f5..97c75ce0 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskSpy.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskSpy.java
@@ -25,7 +25,7 @@ import static org.mockito.Mockito.spy;
import org.onap.dcaegen2.services.prh.config.DmaapConsumerConfiguration;
import org.onap.dcaegen2.services.prh.configuration.AppConfig;
-import org.onap.dcaegen2.services.prh.service.consumer.DmaapConsumerReactiveHttpClient;
+import org.onap.dcaegen2.services.prh.service.consumer.DMaaPConsumerReactiveHttpClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
@@ -42,10 +42,10 @@ public class DmaapConsumerTaskSpy {
AppConfig appConfig = spy(AppConfig.class);
doReturn(mock(DmaapConsumerConfiguration.class)).when(appConfig).getDmaapConsumerConfiguration();
DmaapConsumerTaskImpl dmaapConsumerTask = spy(new DmaapConsumerTaskImpl(appConfig));
- DmaapConsumerReactiveHttpClient dmaapConsumerReactiveHttpClient = mock(
- DmaapConsumerReactiveHttpClient.class);
+ DMaaPConsumerReactiveHttpClient DMaaPConsumerReactiveHttpClient = mock(
+ DMaaPConsumerReactiveHttpClient.class);
doReturn(mock(DmaapConsumerConfiguration.class)).when(dmaapConsumerTask).resolveConfiguration();
- doReturn(dmaapConsumerReactiveHttpClient).when(dmaapConsumerTask).resolveClient();
+ doReturn(DMaaPConsumerReactiveHttpClient).when(dmaapConsumerTask).resolveClient();
return dmaapConsumerTask;
}
}
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapProducerTaskSpy.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapProducerTaskSpy.java
index 01056606..ceb0ed81 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapProducerTaskSpy.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapProducerTaskSpy.java
@@ -22,11 +22,10 @@ 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 static org.mockito.Mockito.when;
import org.onap.dcaegen2.services.prh.config.DmaapPublisherConfiguration;
import org.onap.dcaegen2.services.prh.configuration.AppConfig;
-import org.onap.dcaegen2.services.prh.service.producer.ExtendedDmaapProducerHttpClientImpl;
+import org.onap.dcaegen2.services.prh.service.producer.DMaaPProducerReactiveHttpClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
@@ -43,8 +42,8 @@ public class DmaapProducerTaskSpy {
AppConfig appConfig = spy(AppConfig.class);
doReturn(mock(DmaapPublisherConfiguration.class)).when(appConfig).getDmaapPublisherConfiguration();
DmaapPublisherTaskImpl dmaapPublisherTask = spy(new DmaapPublisherTaskImpl(appConfig));
- ExtendedDmaapProducerHttpClientImpl extendedDmaapProducerHttpClient = mock(
- ExtendedDmaapProducerHttpClientImpl.class);
+ DMaaPProducerReactiveHttpClient extendedDmaapProducerHttpClient = mock(
+ DMaaPProducerReactiveHttpClient.class);
doReturn(mock(DmaapPublisherConfiguration.class)).when(dmaapPublisherTask).resolveConfiguration();
doReturn(extendedDmaapProducerHttpClient).when(dmaapPublisherTask).resolveClient();
return dmaapPublisherTask;
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java
index 13534ce8..6b08be55 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java
@@ -28,7 +28,6 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
-import java.util.Optional;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -36,11 +35,14 @@ import org.junit.jupiter.api.function.Executable;
import org.onap.dcaegen2.services.prh.config.DmaapPublisherConfiguration;
import org.onap.dcaegen2.services.prh.config.ImmutableDmaapPublisherConfiguration;
import org.onap.dcaegen2.services.prh.configuration.AppConfig;
+import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException;
import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
import org.onap.dcaegen2.services.prh.model.ImmutableConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.service.producer.ExtendedDmaapProducerHttpClientImpl;
+import org.onap.dcaegen2.services.prh.service.producer.DMaaPProducerReactiveHttpClient;
import org.springframework.http.HttpStatus;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
/**
* @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/17/18
@@ -49,7 +51,7 @@ class DmaapPublisherTaskImplTest {
private static ConsumerDmaapModel consumerDmaapModel;
private static DmaapPublisherTaskImpl dmaapPublisherTask;
- private static ExtendedDmaapProducerHttpClientImpl extendedDmaapProducerHttpClient;
+ private static DMaaPProducerReactiveHttpClient dMaaPProducerReactiveHttpClient;
private static AppConfig appConfig;
private static DmaapPublisherConfiguration dmaapPublisherConfiguration;
@@ -85,38 +87,36 @@ class DmaapPublisherTaskImplTest {
prepareMocksForTests(HttpStatus.OK.value());
//when
- Integer response = dmaapPublisherTask.execute(consumerDmaapModel);
+ StepVerifier.create(dmaapPublisherTask.execute(Mono.just(consumerDmaapModel))).expectSubscription()
+ .expectNext(HttpStatus.OK.value());
//then
- verify(extendedDmaapProducerHttpClient, times(1))
- .getHttpProducerResponse(any(ConsumerDmaapModel.class));
- verifyNoMoreInteractions(extendedDmaapProducerHttpClient);
- Assertions.assertEquals((Integer) HttpStatus.OK.value(), response);
+ verify(dMaaPProducerReactiveHttpClient, times(1))
+ .getDMaaPProducerResponse(any(Mono.class));
+ verifyNoMoreInteractions(dMaaPProducerReactiveHttpClient);
}
@Test
- public void whenPassedObjectFits_butIncorrectResponseReturns() {
+ public void whenPassedObjectFits_butIncorrectResponseReturns() throws DmaapNotFoundException {
//given
prepareMocksForTests(HttpStatus.UNAUTHORIZED.value());
//when
- Executable executableFunction = () -> dmaapPublisherTask.execute(consumerDmaapModel);
+ StepVerifier.create(dmaapPublisherTask.execute(Mono.just(consumerDmaapModel))).expectSubscription()
+ .expectError(PrhTaskException.class);
//then
- Assertions
- .assertThrows(PrhTaskException.class, executableFunction, "Incorrect response from DMAAP");
- verify(extendedDmaapProducerHttpClient, times(1)).getHttpProducerResponse(any(ConsumerDmaapModel.class));
- verifyNoMoreInteractions(extendedDmaapProducerHttpClient);
+ verify(dMaaPProducerReactiveHttpClient, times(1)).getDMaaPProducerResponse(any(Mono.class));
+ verifyNoMoreInteractions(dMaaPProducerReactiveHttpClient);
}
private void prepareMocksForTests(Integer httpResponseCode) {
- extendedDmaapProducerHttpClient = mock(ExtendedDmaapProducerHttpClientImpl.class);
- when(extendedDmaapProducerHttpClient.getHttpProducerResponse(consumerDmaapModel))
- .thenReturn(Optional.of(httpResponseCode));
- when(appConfig.getDmaapPublisherConfiguration()).thenReturn(dmaapPublisherConfiguration);
+ dMaaPProducerReactiveHttpClient = mock(DMaaPProducerReactiveHttpClient.class);
+ when(dMaaPProducerReactiveHttpClient.getDMaaPProducerResponse(any(Mono.class)))
+ .thenReturn(Mono.just(httpResponseCode));
dmaapPublisherTask = spy(new DmaapPublisherTaskImpl(appConfig));
when(dmaapPublisherTask.resolveConfiguration()).thenReturn(dmaapPublisherConfiguration);
- doReturn(extendedDmaapProducerHttpClient).when(dmaapPublisherTask).resolveClient();
+ doReturn(dMaaPProducerReactiveHttpClient).when(dmaapPublisherTask).resolveClient();
}
} \ No newline at end of file