diff options
Diffstat (limited to 'rest-services/dmaap-client')
7 files changed, 522 insertions, 125 deletions
diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/ContentType.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/ContentType.java new file mode 100644 index 00000000..80a28d6c --- /dev/null +++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/ContentType.java @@ -0,0 +1,40 @@ +/* + * ============LICENSE_START==================================== + * DCAEGEN2-SERVICES-SDK + * ========================================================= + * Copyright (C) 2019 Nokia. All rights reserved. + * ========================================================= + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END===================================== + */ + +package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client; + +import io.netty.handler.codec.http.HttpHeaderValues; +import io.netty.util.AsciiString; + +public enum ContentType { + APPLICATION_JSON(HttpHeaderValues.APPLICATION_JSON), + TEXT_PLAIN(HttpHeaderValues.TEXT_PLAIN); + + private AsciiString contentType; + + ContentType(AsciiString contentType) { + this.contentType = contentType; + } + + @Override + public String toString(){ + return contentType.toString(); + } +} diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImpl.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImpl.java index aa88b9ee..191ec64f 100644 --- a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImpl.java +++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImpl.java @@ -27,6 +27,7 @@ import com.google.gson.JsonElement; import io.vavr.collection.HashMap; import io.vavr.collection.List; import java.time.Duration; +import java.util.stream.Collectors; import org.jetbrains.annotations.NotNull; import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpHeaders; import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpMethod; @@ -35,6 +36,7 @@ import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse; import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.ImmutableHttpRequest; import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RequestBody; import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.ContentType; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterPublisher; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishResponse; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest; @@ -74,14 +76,20 @@ public class MessageRouterPublisherImpl implements MessageRouterPublisher { List<JsonElement> batch) { LOGGER.debug("Sending a batch of {} items to DMaaP MR", batch.size()); LOGGER.trace("The items to be sent: {}", batch); - return httpClient.call(buildHttpRequest(request, asJsonBody(batch))) + return httpClient.call(buildHttpRequest(request, createBody(batch, request.contentType()))) .map(httpResponse -> buildResponse(httpResponse, batch)); } - private @NotNull RequestBody asJsonBody(List<? extends JsonElement> subItems) { - final JsonArray elements = new JsonArray(subItems.size()); - subItems.forEach(elements::add); - return RequestBody.fromJson(elements); + private @NotNull RequestBody createBody(List<? extends JsonElement> subItems, ContentType contentType) { + if(contentType == ContentType.APPLICATION_JSON) { + final JsonArray elements = new JsonArray(subItems.size()); + subItems.forEach(elements::add); + return RequestBody.fromJson(elements); + }else if(contentType == ContentType.TEXT_PLAIN){ + String messages = subItems.map(JsonElement::toString) + .collect(Collectors.joining("\n")); + return RequestBody.fromString(messages); + }else throw new IllegalArgumentException("Unsupported content type: " + contentType); } private @NotNull HttpRequest buildHttpRequest(MessageRouterPublishRequest request, RequestBody body) { @@ -89,7 +97,7 @@ public class MessageRouterPublisherImpl implements MessageRouterPublisher { .method(HttpMethod.POST) .url(request.sinkDefinition().topicUrl()) .diagnosticContext(request.diagnosticContext().withNewInvocationId()) - .customHeaders(HashMap.of(HttpHeaders.CONTENT_TYPE, request.contentType())) + .customHeaders(HashMap.of(HttpHeaders.CONTENT_TYPE, request.contentType().toString())) .body(body) .build(); } @@ -98,6 +106,7 @@ public class MessageRouterPublisherImpl implements MessageRouterPublisher { HttpResponse httpResponse, List<JsonElement> batch) { final ImmutableMessageRouterPublishResponse.Builder builder = ImmutableMessageRouterPublishResponse.builder(); + return httpResponse.successful() ? builder.items(batch).build() : builder.failReason(extractFailReason(httpResponse)).build(); diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/MessageRouterPublishRequest.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/MessageRouterPublishRequest.java index 77f92e77..29904138 100644 --- a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/MessageRouterPublishRequest.java +++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/MessageRouterPublishRequest.java @@ -23,6 +23,7 @@ package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model; import org.immutables.value.Value; import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink; import org.onap.dcaegen2.services.sdk.rest.services.annotations.ExperimentalApi; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.ContentType; /** * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a> @@ -34,7 +35,7 @@ public interface MessageRouterPublishRequest extends DmaapRequest { MessageRouterSink sinkDefinition(); @Value.Default - default String contentType() { - return "application/json"; + default ContentType contentType() { + return ContentType.APPLICATION_JSON; } } diff --git a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterTestsUtils.java b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/MessageRouterTestsUtils.java index 8695b727..52946f56 100644 --- a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterTestsUtils.java +++ b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/MessageRouterTestsUtils.java @@ -18,7 +18,7 @@ * ============LICENSE_END===================================== */ -package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api; +package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client; import com.google.gson.Gson; import com.google.gson.JsonElement; @@ -29,6 +29,8 @@ import io.vavr.collection.List; import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableMessageRouterSink; import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableMessageRouterSource; import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterPublisher; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterSubscriber; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishRequest; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishResponse; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeRequest; @@ -39,11 +41,16 @@ import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRo import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeResponse; import reactor.core.publisher.Flux; -final class MessageRouterTestsUtils { + +public final class MessageRouterTestsUtils { private static final JsonParser parser = new JsonParser(); private MessageRouterTestsUtils() {} - static MessageRouterPublishRequest createPublishRequest(String topicUrl){ + public static MessageRouterPublishRequest createPublishRequest(String topicUrl){ + return createPublishRequest(topicUrl, ContentType.APPLICATION_JSON); + } + + public static MessageRouterPublishRequest createPublishRequest(String topicUrl, ContentType contentType){ MessageRouterSink sinkDefinition = ImmutableMessageRouterSink.builder() .name("the topic") .topicUrl(topicUrl) @@ -51,10 +58,11 @@ final class MessageRouterTestsUtils { return ImmutableMessageRouterPublishRequest.builder() .sinkDefinition(sinkDefinition) + .contentType(contentType) .build(); } - static MessageRouterSubscribeRequest createMRSubscribeRequest(String topicUrl, + public static MessageRouterSubscribeRequest createMRSubscribeRequest(String topicUrl, String consumerGroup, String consumerId) { ImmutableMessageRouterSource sourceDefinition = ImmutableMessageRouterSource.builder() .name("the topic") @@ -69,51 +77,59 @@ final class MessageRouterTestsUtils { .build(); } - static List<JsonElement> getAsJsonElements(List<String> messages){ + public static List<JsonElement> getAsJsonElements(List<String> messages){ return messages.map(parser::parse); } - static JsonObject getAsJsonObject(String item){ + public static List<JsonObject> getAsJsonObjects(List<String> messages){ + return getAsJsonElements(messages).map(JsonElement::getAsJsonObject); + } + + public static List<JsonPrimitive> getAsJsonPrimitives(List<String> messages){ + return getAsJsonElements(messages).map(JsonElement::getAsJsonPrimitive); + } + + public static JsonObject getAsJsonObject(String item){ return new Gson().fromJson(item, JsonObject.class); } - static Flux<JsonObject> jsonBatch(List<String> messages){ - return Flux.fromIterable(messages).map(parser::parse).map(JsonElement::getAsJsonObject); + public static Flux<JsonElement> plainBatch(List<String> messages){ + return Flux.fromIterable(getAsJsonElements(messages)); } - static Flux<JsonPrimitive> plainBatch(List<String> messages){ - return Flux.fromIterable(messages).map(JsonPrimitive::new); + public static Flux<JsonObject> jsonBatch(List<String> messages){ + return Flux.fromIterable(getAsJsonObjects(messages)); } - static MessageRouterSubscribeResponse errorSubscribeResponse(String failReasonFormat, Object... formatArgs){ + public static MessageRouterSubscribeResponse errorSubscribeResponse(String failReasonFormat, Object... formatArgs){ return ImmutableMessageRouterSubscribeResponse .builder() .failReason(String.format(failReasonFormat, formatArgs)) .build(); } - static MessageRouterSubscribeResponse successSubscribeResponse(List<JsonElement> items){ + public static MessageRouterSubscribeResponse successSubscribeResponse(List<JsonElement> items){ return ImmutableMessageRouterSubscribeResponse .builder() .items(items) .build(); } - static MessageRouterPublishResponse errorPublishResponse(String failReasonFormat, Object... formatArgs){ + public static MessageRouterPublishResponse errorPublishResponse(String failReasonFormat, Object... formatArgs){ return ImmutableMessageRouterPublishResponse .builder() .failReason(String.format(failReasonFormat, formatArgs)) .build(); } - static MessageRouterPublishResponse successPublishResponse(List<JsonElement> items){ + public static MessageRouterPublishResponse successPublishResponse(List<JsonElement> items){ return ImmutableMessageRouterPublishResponse .builder() .items(items) .build(); } - static void registerTopic(MessageRouterPublisher publisher, MessageRouterPublishRequest publishRequest, + public static void registerTopic(MessageRouterPublisher publisher, MessageRouterPublishRequest publishRequest, MessageRouterSubscriber subscriber, MessageRouterSubscribeRequest subscribeRequest) { final List<String> sampleJsonMessages = List.of("{\"message\":\"message1\"}", "{\"differentMessage\":\"message2\"}"); diff --git a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherIT.java b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherIT.java index c746bfec..0afea748 100644 --- a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherIT.java +++ b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherIT.java @@ -20,15 +20,15 @@ package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api; -import static org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterTestsUtils.*; +import static org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.MessageRouterTestsUtils.*; import com.google.gson.JsonElement; import com.google.gson.JsonObject; -import com.google.gson.JsonPrimitive; import io.vavr.collection.List; import java.time.Duration; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.ContentType; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeRequest; @@ -95,7 +95,7 @@ class MessageRouterPublisherIT { //given final String topic = "TOPIC2"; final List<String> threePlainTextMessages = List.of("I", "like", "pizza"); - final Flux<JsonPrimitive> messageBatch = plainBatch(threePlainTextMessages); + final Flux<JsonElement> messageBatch = plainBatch(threePlainTextMessages); final MessageRouterPublishRequest mrRequest = createPublishRequest(String.format("%s/%s", EVENTS_PATH, topic)); final MessageRouterPublishResponse expectedResponse = errorPublishResponse( DMAAP_400_ERROR_RESPONSE_FORMAT, topic); @@ -112,6 +112,7 @@ class MessageRouterPublisherIT { @Test void publisher_shouldSuccessfullyPublishSingleMessage(){ + //given final String topic = "TOPIC3"; final String topicUrl = String.format("%s/%s", EVENTS_PATH, topic); final List<String> singleJsonMessage = List.of("{\"message\":\"message1\"}"); @@ -136,7 +137,7 @@ class MessageRouterPublisherIT { @Test void publisher_shouldSuccessfullyPublishMultipleMessages(){ - final String topic = "TOPIC4"; + final String topic = "TOPIC5"; final String topicUrl = String.format("%s/%s", EVENTS_PATH, topic); final List<String> singleJsonMessage = List.of("{\"message\":\"message1\"}", "{\"differentMessage\":\"message2\"}"); @@ -158,4 +159,112 @@ class MessageRouterPublisherIT { .expectComplete() .verify(); } + + @Test + void publisher_shouldSuccessfullyPublishSingleJsonMessageWithPlainContentType(){ + //given + final String topic = "TOPIC6"; + final String topicUrl = String.format("%s/%s", EVENTS_PATH, topic); + + final List<String> singleJsonMessage = List.of("{\"message\":\"message1\"}"); + final List<JsonElement> expectedItems = getAsJsonElements(singleJsonMessage); + final Flux<JsonElement> plainBatch = plainBatch(singleJsonMessage); + + final MessageRouterPublishRequest publishRequest = createPublishRequest(topicUrl, ContentType.TEXT_PLAIN); + final MessageRouterSubscribeRequest subscribeRequest = createMRSubscribeRequest(topicUrl, "sampleGroup", "sampleId"); + final MessageRouterSubscribeResponse expectedResponse = successSubscribeResponse(expectedItems); + + //when + registerTopic(publisher, publishRequest, subscriber, subscribeRequest); + Mono<MessageRouterSubscribeResponse> response = publisher + .put(publishRequest, plainBatch) + .then(subscriber.get(subscribeRequest)); + + //then + StepVerifier.create(response) + .expectNext(expectedResponse) + .expectComplete() + .verify(); + } + + @Test + void publisher_shouldSuccessfullyPublishMultipleJsonMessagesWithPlainContentType(){ + //given + final String topic = "TOPIC7"; + final String topicUrl = String.format("%s/%s", EVENTS_PATH, topic); + + final List<String> twoJsonMessage = List.of("{\"message\":\"message1\"}", "{\"message2\":\"message2\"}"); + final List<JsonElement> expectedItems = getAsJsonElements(twoJsonMessage); + final Flux<JsonElement> plainBatch = plainBatch(twoJsonMessage); + + final MessageRouterPublishRequest publishRequest = createPublishRequest(topicUrl, ContentType.TEXT_PLAIN); + final MessageRouterSubscribeRequest subscribeRequest = createMRSubscribeRequest(topicUrl, "sampleGroup", "sampleId"); + final MessageRouterSubscribeResponse expectedResponse = successSubscribeResponse(expectedItems); + + //when + registerTopic(publisher, publishRequest, subscriber, subscribeRequest); + Mono<MessageRouterSubscribeResponse> response = publisher + .put(publishRequest, plainBatch) + .then(subscriber.get(subscribeRequest)); + + //then + StepVerifier.create(response) + .expectNext(expectedResponse) + .expectComplete() + .verify(); + } + + @Test + void publisher_shouldSuccessfullyPublishSinglePlainMessageWithPlainContentType(){ + //given + final String topic = "TOPIC8"; + final String topicUrl = String.format("%s/%s", EVENTS_PATH, topic); + + final List<String> singlePlainMessage = List.of("kebab"); + final List<JsonElement> expectedItems = getAsJsonElements(singlePlainMessage); + final Flux<JsonElement> plainBatch = plainBatch(singlePlainMessage); + + final MessageRouterPublishRequest publishRequest = createPublishRequest(topicUrl, ContentType.TEXT_PLAIN); + final MessageRouterSubscribeRequest subscribeRequest = createMRSubscribeRequest(topicUrl, "sampleGroup", "sampleId"); + final MessageRouterSubscribeResponse expectedResponse = successSubscribeResponse(expectedItems); + + //when + registerTopic(publisher, publishRequest, subscriber, subscribeRequest); + Mono<MessageRouterSubscribeResponse> response = publisher + .put(publishRequest, plainBatch) + .then(subscriber.get(subscribeRequest)); + + //then + StepVerifier.create(response) + .expectNext(expectedResponse) + .expectComplete() + .verify(); + } + + @Test + void publisher_shouldSuccessfullyPublishMultiplePlainMessagesWithPlainContentType(){ + //given + final String topic = "TOPIC9"; + final String topicUrl = String.format("%s/%s", EVENTS_PATH, topic); + + final List<String> singlePlainMessage = List.of("I", "like", "pizza"); + final List<JsonElement> expectedItems = getAsJsonElements(singlePlainMessage); + final Flux<JsonElement> plainBatch = plainBatch(singlePlainMessage); + + final MessageRouterPublishRequest publishRequest = createPublishRequest(topicUrl, ContentType.TEXT_PLAIN); + final MessageRouterSubscribeRequest subscribeRequest = createMRSubscribeRequest(topicUrl, "sampleGroup", "sampleId"); + final MessageRouterSubscribeResponse expectedResponse = successSubscribeResponse(expectedItems); + + //when + registerTopic(publisher, publishRequest, subscriber, subscribeRequest); + Mono<MessageRouterSubscribeResponse> response = publisher + .put(publishRequest, plainBatch) + .then(subscriber.get(subscribeRequest)); + + //then + StepVerifier.create(response) + .expectNext(expectedResponse) + .expectComplete() + .verify(); + } } diff --git a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberIT.java b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberIT.java index c2e96b58..a758b2de 100644 --- a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberIT.java +++ b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberIT.java @@ -20,7 +20,7 @@ package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api; -import static org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterTestsUtils.*; +import static org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.MessageRouterTestsUtils.*; import com.google.gson.JsonElement; import com.google.gson.JsonObject; diff --git a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImplTest.java b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImplTest.java index f44e86a5..38659acd 100644 --- a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImplTest.java +++ b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImplTest.java @@ -26,32 +26,28 @@ import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.MessageRouterTestsUtils.*; import com.google.gson.JsonArray; import com.google.gson.JsonElement; import com.google.gson.JsonObject; -import com.google.gson.JsonParser; import com.google.gson.Gson; +import com.google.gson.JsonPrimitive; import io.netty.buffer.ByteBufAllocator; import io.netty.handler.codec.http.HttpHeaderValues; +import io.vavr.collection.List; import java.nio.charset.StandardCharsets; import java.time.Duration; -import java.util.Arrays; -import java.util.List; -import java.util.stream.Collectors; -import java.util.stream.Stream; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; -import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableMessageRouterSink; -import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink; import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpHeaders; import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpMethod; import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpRequest; import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse; import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.ImmutableHttpResponse; import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.ContentType; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterPublisher; -import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishRequest; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse; import reactor.core.publisher.Flux; @@ -64,71 +60,123 @@ import reactor.test.StepVerifier; */ class MessageRouterPublisherImplTest { private static final Duration TIMEOUT = Duration.ofSeconds(5); - private static final JsonParser parser = new JsonParser(); - private static final MessageRouterSink sinkDefinition = ImmutableMessageRouterSink.builder() - .name("the topic") - .topicUrl("https://dmaap-mr/TOPIC") - .build(); + private static final String TOPIC_URL = "https://dmaap-mr/TOPIC"; + private static final int MAX_BATCH_SIZE = 3; private final RxHttpClient httpClient = mock(RxHttpClient.class); - private final MessageRouterPublisher cut = new MessageRouterPublisherImpl(httpClient, 3, Duration.ofMinutes(1)); + private final MessageRouterPublisher cut = new MessageRouterPublisherImpl(httpClient, MAX_BATCH_SIZE, Duration.ofMinutes(1)); private final ArgumentCaptor<HttpRequest> httpRequestArgumentCaptor = ArgumentCaptor.forClass(HttpRequest.class); - private final MessageRouterPublishRequest mrRequestTextPlain = createMRRPublishRequest(); - private final MessageRouterPublishRequest mrRequestJson = createMRRPublishRequest(); + private final MessageRouterPublishRequest plainPublishRequest = createPublishRequest(TOPIC_URL, ContentType.TEXT_PLAIN); + private final MessageRouterPublishRequest jsonPublishRequest = createPublishRequest(TOPIC_URL); private final HttpResponse successHttpResponse = createHttpResponse("OK", 200); @Test void puttingElementsShouldYieldNonChunkedHttpRequest() { // given - final List<String> threeJsonMessages = getAsMRJsonMessages(Arrays.asList("I", "like", "cookies")); + final List<String> threeJsonMessages = getAsMRJsonMessages(List.of("I", "like", "cookies")); final Flux<JsonObject> singleJsonMessageBatch = jsonBatch(threeJsonMessages); given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse)); // when final Flux<MessageRouterPublishResponse> responses = cut - .put(mrRequestTextPlain, singleJsonMessageBatch); + .put(jsonPublishRequest, singleJsonMessageBatch); responses.then().block(); // then verify(httpClient).call(httpRequestArgumentCaptor.capture()); final HttpRequest httpRequest = httpRequestArgumentCaptor.getValue(); assertThat(httpRequest.method()).isEqualTo(HttpMethod.POST); - assertThat(httpRequest.url()).isEqualTo(sinkDefinition.topicUrl()); + assertThat(httpRequest.url()).isEqualTo(TOPIC_URL); assertThat(httpRequest.body()).isNotNull(); assertThat(httpRequest.body().length()).isGreaterThan(0); } @Test - void puttingLowNumberOfElementsShouldYieldSingleHttpRequest() { + void onPut_givenJsonMessages_whenTheirAmountIsNotAboveMaxBatchSize_shouldSendSingleHttpRequest(){ // given - final List<String> threeJsonMessages = getAsMRJsonMessages(Arrays.asList("I", "like", "cookies")); - final Flux<JsonObject> singleJsonMessageBatch = jsonBatch(threeJsonMessages); + final List<String> threeJsonMessages = getAsMRJsonMessages(List.of("I", "like", "cookies")); + final List<JsonObject> parsedThreeMessages = getAsJsonObjects(threeJsonMessages); + + final Flux<JsonObject> jsonMessagesMaxBatch = jsonBatch(threeJsonMessages); given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse)); // when final Flux<MessageRouterPublishResponse> responses = cut - .put(mrRequestJson, singleJsonMessageBatch); + .put(jsonPublishRequest, jsonMessagesMaxBatch); responses.then().block(); // then verify(httpClient).call(httpRequestArgumentCaptor.capture()); final HttpRequest httpRequest = httpRequestArgumentCaptor.getValue(); final JsonArray elementsInRequest = extractNonEmptyJsonRequestBody(httpRequest); - assertThat(elementsInRequest.size()).isEqualTo(3); - assertThat(elementsInRequest.get(0).toString()).isEqualTo(threeJsonMessages.get(0)); - assertThat(elementsInRequest.get(1).toString()).isEqualTo(threeJsonMessages.get(1)); - assertThat(elementsInRequest.get(2).toString()).isEqualTo(threeJsonMessages.get(2)); + + assertThat(elementsInRequest.size()).describedAs("Http request batch size") + .isEqualTo(MAX_BATCH_SIZE); + assertListsContainSameElements(elementsInRequest, parsedThreeMessages); + } + + + + @Test + void onPut_givenJsonMessagesWithPlainContentType_whenTheirAmountIsNotAboveMaxBatchSize_shouldSendSingleHttpRequest(){ + // given + final List<String> threeJsonMessages = getAsMRJsonMessages(List.of("I", "like", "cookies")); + final List<JsonObject> parsedThreeMessages = getAsJsonObjects(threeJsonMessages); + + final Flux<JsonElement> plainMessagesMaxBatch = plainBatch(threeJsonMessages); + given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse)); + + // when + final Flux<MessageRouterPublishResponse> responses = cut + .put(plainPublishRequest, plainMessagesMaxBatch); + responses.then().block(); + + // then + verify(httpClient).call(httpRequestArgumentCaptor.capture()); + final HttpRequest httpRequest = httpRequestArgumentCaptor.getValue(); + final List<JsonObject> elementsInRequest = extractNonEmptyPlainRequestBody(httpRequest) + .map(JsonElement::getAsJsonObject); + + + assertThat(elementsInRequest.size()).describedAs("Http request batch size") + .isEqualTo(MAX_BATCH_SIZE); + assertListsContainSameElements(elementsInRequest, parsedThreeMessages); + } + + @Test + void onPut_givenPlainMessages_whenTheirAmountIsNotAboveMaxBatchSize_shouldSendSingleHttpRequest(){ + // given + final List<String> threePlainMessages = List.of("I", "like", "cookies"); + final List<JsonPrimitive> parsedThreeMessages = getAsJsonPrimitives(threePlainMessages); + + final Flux<JsonElement> plainMessagesMaxBatch = plainBatch(threePlainMessages); + given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse)); + + // when + final Flux<MessageRouterPublishResponse> responses = cut + .put(plainPublishRequest, plainMessagesMaxBatch); + responses.then().block(); + + // then + verify(httpClient).call(httpRequestArgumentCaptor.capture()); + final HttpRequest httpRequest = httpRequestArgumentCaptor.getValue(); + final List<JsonPrimitive> elementsInRequest = extractNonEmptyPlainRequestBody(httpRequest) + .map(JsonElement::getAsJsonPrimitive); + + assertThat(elementsInRequest.size()).describedAs("Http request batch size") + .isEqualTo(MAX_BATCH_SIZE); + assertListsContainSameElements(elementsInRequest, parsedThreeMessages); } @Test void puttingElementsWithoutContentTypeSetShouldUseApplicationJson(){ // given - final List<String> threeJsonMessages = getAsMRJsonMessages(Arrays.asList("I", "like", "cookies")); + final List<String> threeJsonMessages = getAsMRJsonMessages(List.of("I", "like", "cookies")); final Flux<JsonObject> singleJsonMessageBatch = jsonBatch(threeJsonMessages); given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse)); // when final Flux<MessageRouterPublishResponse> responses = cut - .put(mrRequestJson, singleJsonMessageBatch); + .put(jsonPublishRequest, singleJsonMessageBatch); responses.then().block(); // then @@ -139,124 +187,237 @@ class MessageRouterPublisherImplTest { } @Test - void puttingLowNumberOfElementsShouldReturnSingleResponse() { + void onPut_givenJsonMessages_whenTheirAmountIsNotAboveMaxBatchSize_shouldReturnSingleResponse() { // given - final List<String> threeJsonMessages = getAsMRJsonMessages(Arrays.asList("I", "like", "cookies")); - final Flux<JsonObject> singleJsonMessageBatch = jsonBatch(threeJsonMessages); + final List<String> threeJsonMessages = getAsMRJsonMessages(List.of("I", "like", "cookies")); + final List<JsonObject> parsedThreeMessages = getAsJsonObjects(threeJsonMessages); + + final Flux<JsonObject> jsonMessagesMaxBatch = jsonBatch(threeJsonMessages); given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse)); // when final Flux<MessageRouterPublishResponse> responses = cut - .put(mrRequestJson, singleJsonMessageBatch); + .put(jsonPublishRequest, jsonMessagesMaxBatch); // then - StepVerifier.create(responses) - .consumeNextWith(response -> { - assertThat(response.successful()).describedAs("successful").isTrue(); - assertThat(response.items()).containsExactly( - getAsJsonObject(threeJsonMessages.get(0)), - getAsJsonObject(threeJsonMessages.get(1)), - getAsJsonObject(threeJsonMessages.get(2))); - }) - .expectComplete() - .verify(TIMEOUT); + verifySingleResponse(parsedThreeMessages, responses); } @Test - void puttingHighNumberOfElementsShouldYieldMultipleHttpRequests() { + void onPut_givenJsonMessagesWithPlainContentType_whenTheirAmountIsNotAboveMaxBatchSize_shouldReturnSingleResponse() { // given - final List<String> threeJsonMessages = getAsMRJsonMessages(Arrays.asList("I", "like", "cookies")); - final List<String> twoJsonMessages = getAsMRJsonMessages(Arrays.asList("and", "pierogi")); - final Flux<JsonObject> doubleJsonMessageBatch = jsonBatch(concat( - threeJsonMessages, twoJsonMessages)); + final List<String> threeJsonMessages = getAsMRJsonMessages(List.of("I", "like", "cookies")); + final List<JsonObject> parsedThreeMessages = getAsJsonObjects(threeJsonMessages); + final Flux<JsonElement> plainMessagesMaxBatch = plainBatch(threeJsonMessages); given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse)); // when final Flux<MessageRouterPublishResponse> responses = cut - .put(mrRequestJson, doubleJsonMessageBatch); + .put(plainPublishRequest, plainMessagesMaxBatch); + + // then + verifySingleResponse(parsedThreeMessages, responses); + } + + @Test + void onPut_givenPlainMessages_whenTheirAmountIsNotAboveMaxBatchSize_shouldReturnSingleResponse() { + // given + final List<String> threePlainMessages = List.of("I", "like", "cookies"); + final List<JsonPrimitive> parsedThreeMessages = getAsJsonPrimitives(threePlainMessages); + + final Flux<JsonElement> plainMessagesMaxBatch = plainBatch(threePlainMessages); + given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse)); + + // when + final Flux<MessageRouterPublishResponse> responses = cut + .put(plainPublishRequest, plainMessagesMaxBatch); + + // then + verifySingleResponse(parsedThreeMessages, responses); + } + + @Test + void onPut_givenJsonMessages_whenTheirAmountIsAboveMaxBatchSize_shouldYieldMultipleHttpRequests() { + // given + final List<String> threeJsonMessages = getAsMRJsonMessages(List.of("I", "like", "cookies")); + final List<String> twoJsonMessages = getAsMRJsonMessages(List.of("and", "pierogi")); + + final List<JsonObject> parsedThreeMessages = getAsJsonObjects(threeJsonMessages); + final List<JsonObject> parsedTwoMessages = getAsJsonObjects(twoJsonMessages); + + final Flux<JsonObject> doubleJsonMessageBatch = jsonBatch( + threeJsonMessages.appendAll(twoJsonMessages)); + given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse)); + + // when + final Flux<MessageRouterPublishResponse> responses = cut + .put(jsonPublishRequest, doubleJsonMessageBatch); // then responses.then().block(); verify(httpClient, times(2)).call(httpRequestArgumentCaptor.capture()); - final List<HttpRequest> httpRequests = httpRequestArgumentCaptor.getAllValues(); + final List<HttpRequest> httpRequests = List.ofAll(httpRequestArgumentCaptor.getAllValues()); assertThat(httpRequests.size()).describedAs("number of requests").isEqualTo(2); final JsonArray firstRequest = extractNonEmptyJsonRequestBody(httpRequests.get(0)); - assertThat(firstRequest.size()).isEqualTo(3); - assertThat(firstRequest.get(0).toString()).isEqualTo(threeJsonMessages.get(0)); - assertThat(firstRequest.get(1).toString()).isEqualTo(threeJsonMessages.get(1)); - assertThat(firstRequest.get(2).toString()).isEqualTo(threeJsonMessages.get(2)); + assertThat(firstRequest.size()).describedAs("Http request first batch size") + .isEqualTo(MAX_BATCH_SIZE); + assertListsContainSameElements(firstRequest, parsedThreeMessages); final JsonArray secondRequest = extractNonEmptyJsonRequestBody(httpRequests.get(1)); - assertThat(secondRequest.size()).isEqualTo(2); - assertThat(secondRequest.get(0).toString()).isEqualTo(twoJsonMessages.get(0)); - assertThat(secondRequest.get(1).toString()).isEqualTo(twoJsonMessages.get(1)); + assertThat(secondRequest.size()).describedAs("Http request second batch size") + .isEqualTo(MAX_BATCH_SIZE-1); + assertListsContainSameElements(secondRequest, parsedTwoMessages); } @Test - void puttingHighNumberOfElementsShouldReturnMoreResponses() { + void onPut_givenJsonMessagesWithPlainContentType_whenTheirAmountIsAboveMaxBatchSize_shouldYieldMultipleHttpRequests() { // given - final List<String> threeJsonMessages = getAsMRJsonMessages(Arrays.asList("I", "like", "cookies")); - final List<String> twoJsonMessages = getAsMRJsonMessages(Arrays.asList("and", "pierogi")); - final Flux<JsonObject> doubleJsonMessageBatch = jsonBatch(concat( - threeJsonMessages, twoJsonMessages)); + final List<String> threeJsonMessages = getAsMRJsonMessages(List.of("I", "like", "cookies")); + final List<String> twoJsonMessages = getAsMRJsonMessages(List.of("and", "pierogi")); + + final List<JsonObject> parsedThreeMessages = getAsJsonObjects(threeJsonMessages); + final List<JsonObject> parsedTwoMessages = getAsJsonObjects(twoJsonMessages); + + final Flux<JsonElement> doublePlainMessageBatch = plainBatch( + threeJsonMessages.appendAll(twoJsonMessages)); given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse)); // when final Flux<MessageRouterPublishResponse> responses = cut - .put(mrRequestJson, doubleJsonMessageBatch); + .put(plainPublishRequest, doublePlainMessageBatch); + // then + responses.then().block(); + + verify(httpClient, times(2)).call(httpRequestArgumentCaptor.capture()); + final List<HttpRequest> httpRequests = List.ofAll(httpRequestArgumentCaptor.getAllValues()); + assertThat(httpRequests.size()).describedAs("number of requests").isEqualTo(2); + + final List<JsonObject> firstRequest = extractNonEmptyPlainRequestBody(httpRequests.get(0)) + .map(JsonElement::getAsJsonObject); + assertThat(firstRequest.size()).describedAs("Http request first batch size") + .isEqualTo(MAX_BATCH_SIZE); + assertListsContainSameElements(firstRequest, parsedThreeMessages); + + final List<JsonObject> secondRequest = extractNonEmptyPlainRequestBody(httpRequests.get(1)) + .map(JsonElement::getAsJsonObject); + assertThat(secondRequest.size()).describedAs("Http request second batch size") + .isEqualTo(MAX_BATCH_SIZE-1); + assertListsContainSameElements(secondRequest, parsedTwoMessages); + } + + @Test + void onPut_givenPlainMessages_whenTheirAmountIsAboveMaxBatchSize_shouldYieldMultipleHttpRequests() { + // given + final List<String> threePlainMessages = List.of("I", "like", "cookies"); + final List<String> twoPlainMessages = List.of("and", "pierogi"); + + final List<JsonPrimitive> parsedThreePlainMessages = getAsJsonPrimitives(threePlainMessages); + final List<JsonPrimitive> parsedTwoPlainMessages = getAsJsonPrimitives(twoPlainMessages); + + final Flux<JsonElement> doublePlainMessageBatch = plainBatch( + threePlainMessages.appendAll(twoPlainMessages)); + given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse)); + // when + final Flux<MessageRouterPublishResponse> responses = cut + .put(plainPublishRequest, doublePlainMessageBatch); // then - StepVerifier.create(responses) - .consumeNextWith(response -> { - assertThat(response.successful()).describedAs("successful").isTrue(); - assertThat(response.items()).containsExactly( - getAsJsonObject(threeJsonMessages.get(0)), - getAsJsonObject(threeJsonMessages.get(1)), - getAsJsonObject(threeJsonMessages.get(2))); - }) - .consumeNextWith(response -> { - assertThat(response.successful()).describedAs("successful").isTrue(); - assertThat(response.items()).containsExactly( - getAsJsonObject(twoJsonMessages.get(0)), - getAsJsonObject(twoJsonMessages.get(1))); - }) - .expectComplete() - .verify(TIMEOUT); + responses.then().block(); + + verify(httpClient, times(2)).call(httpRequestArgumentCaptor.capture()); + final List<HttpRequest> httpRequests = List.ofAll(httpRequestArgumentCaptor.getAllValues()); + assertThat(httpRequests.size()).describedAs("number of requests").isEqualTo(2); + + final List<JsonPrimitive> firstRequest = extractNonEmptyPlainRequestBody(httpRequests.get(0)) + .map(JsonElement::getAsJsonPrimitive); + assertThat(firstRequest.size()).describedAs("Http request first batch size") + .isEqualTo(MAX_BATCH_SIZE); + assertListsContainSameElements(firstRequest, parsedThreePlainMessages); + + final List<JsonPrimitive> secondRequest = extractNonEmptyPlainRequestBody(httpRequests.get(1)) + .map(JsonElement::getAsJsonPrimitive); + assertThat(secondRequest.size()).describedAs("Http request second batch size") + .isEqualTo(MAX_BATCH_SIZE-1); + assertListsContainSameElements(secondRequest, parsedTwoPlainMessages); } - private static List<String> getAsMRJsonMessages(List<String> plainTextMessages){ - return plainTextMessages.stream() - .map(message -> String.format("{\"message\":\"%s\"}", message)) - .collect(Collectors.toList()); + @Test + void onPut_givenJsonMessages_whenTheirAmountIsAboveMaxBatchSize_shouldReturnMoreResponses() { + // given + final List<String> threeJsonMessages = getAsMRJsonMessages(List.of("I", "like", "cookies")); + final List<String> twoJsonMessages = getAsMRJsonMessages(List.of("and", "pierogi")); + + final List<JsonObject> parsedThreeMessages = getAsJsonObjects(threeJsonMessages); + final List<JsonObject> parsedTwoMessages = getAsJsonObjects(twoJsonMessages); + + final Flux<JsonObject> doubleJsonMessageBatch = jsonBatch(threeJsonMessages.appendAll(twoJsonMessages)); + given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse)); + + // when + final Flux<MessageRouterPublishResponse> responses = cut + .put(jsonPublishRequest, doubleJsonMessageBatch); + + // then + verifyDoubleResponse(parsedThreeMessages, parsedTwoMessages, responses); + } + + @Test + void onPut_givenJsonMessagesWithPlainContentType_whenTheirAmountIsAboveMaxBatchSize_shouldReturnMoreResponses() { + // given + final List<String> threeJsonMessages = getAsMRJsonMessages(List.of("I", "like", "cookies")); + final List<String> twoJsonMessages = getAsMRJsonMessages(List.of("and", "pierogi")); + + final List<JsonObject> parsedThreeMessages = getAsJsonObjects(threeJsonMessages); + final List<JsonObject> parsedTwoMessages = getAsJsonObjects(twoJsonMessages); + + final Flux<JsonElement> doubleJsonMessageBatch = plainBatch(threeJsonMessages.appendAll(twoJsonMessages)); + given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse)); + + // when + final Flux<MessageRouterPublishResponse> responses = cut + .put(plainPublishRequest, doubleJsonMessageBatch); + + // then + verifyDoubleResponse(parsedThreeMessages, parsedTwoMessages, responses); } + @Test + void onPut_givenPlainMessages_whenTheirAmountIsAboveMaxBatchSize_shouldReturnMoreResponses() { + // given + final List<String> threePlainMessages = List.of("I", "like", "cookies"); + final List<String> twoPlainMessages = List.of("and", "pierogi"); + + final List<JsonPrimitive> parsedThreeMessages = getAsJsonPrimitives(threePlainMessages); + final List<JsonPrimitive> parsedTwoMessages = getAsJsonPrimitives(twoPlainMessages); - private static Flux<JsonObject> jsonBatch(List<String> messages){ - return Flux.fromIterable(messages).map(parser::parse).map(JsonElement::getAsJsonObject); + final Flux<JsonElement> doublePlainMessageBatch = plainBatch( + threePlainMessages.appendAll(twoPlainMessages)); + given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse)); + + // when + final Flux<MessageRouterPublishResponse> responses = cut + .put(plainPublishRequest, doublePlainMessageBatch); + + // then + verifyDoubleResponse(parsedThreeMessages, parsedTwoMessages, responses); } - private static List<String> concat(List<String> firstList, List<String> secondList){ - return Stream.concat(firstList.stream(), secondList.stream()).collect(Collectors.toList()); + private static List<String> getAsMRJsonMessages(List<String> plainTextMessages){ + return plainTextMessages + .map(message -> String.format("{\"message\":\"%s\"}", message)); } private static HttpResponse createHttpResponse(String statusReason, int statusCode){ return ImmutableHttpResponse.builder() .statusCode(statusCode) - .url(sinkDefinition.topicUrl()) + .url(TOPIC_URL) .statusReason(statusReason) .rawBody("[]".getBytes()) .build(); } - private static MessageRouterPublishRequest createMRRPublishRequest(){ - return ImmutableMessageRouterPublishRequest - .builder() - .sinkDefinition(sinkDefinition) - .build(); - } - private String collectNonEmptyRequestBody(HttpRequest httpRequest){ final String body = Flux.from(httpRequest.body().contents()) .collect(ByteBufAllocator.DEFAULT::compositeBuffer, @@ -272,7 +433,68 @@ class MessageRouterPublisherImplTest { return new Gson().fromJson(collectNonEmptyRequestBody(httpRequest), JsonArray.class); } - private JsonObject getAsJsonObject(String item){ - return new Gson().fromJson(item, JsonObject.class); + private List<JsonElement> extractNonEmptyPlainRequestBody(HttpRequest httpRequest){ + return getAsJsonElements( + List.of( + collectNonEmptyRequestBody(httpRequest) + .split("\n") + ) + ); + } + + private void assertListsContainSameElements(List<? extends JsonElement> actualMessages, + List<? extends JsonElement> expectedMessages){ + for (int i = 0; i < actualMessages.size(); i++) { + assertThat(actualMessages.get(i)) + .describedAs(String.format("Http request element at position %d", i)) + .isEqualTo(expectedMessages.get(i)); + } + } + + private void assertListsContainSameElements(JsonArray actualMessages, + List<? extends JsonElement> expectedMessages){ + assertThat(actualMessages.size()).describedAs("Http request batch size") + .isEqualTo(expectedMessages.size()); + + for (int i = 0; i < actualMessages.size(); i++) { + assertThat(actualMessages.get(i)) + .describedAs(String.format("Http request element at position %d", i)) + .isEqualTo(expectedMessages.get(i)); + } + } + + private void verifySingleResponse(List<? extends JsonElement> threeMessages, + Flux<MessageRouterPublishResponse> responses) { + StepVerifier.create(responses) + .consumeNextWith(response -> { + assertThat(response.successful()).describedAs("successful").isTrue(); + assertThat(response.items()).containsExactly( + threeMessages.get(0), + threeMessages.get(1), + threeMessages.get(2)); + }) + .expectComplete() + .verify(TIMEOUT); + } + + private void verifyDoubleResponse(List<? extends JsonElement> threeMessages, + List<? extends JsonElement> twoMessages, Flux<MessageRouterPublishResponse> responses) { + + StepVerifier.create(responses) + .consumeNextWith(response -> { + assertThat(response.successful()).describedAs("successful").isTrue(); + assertThat(response.items()).containsExactly( + threeMessages.get(0), + threeMessages.get(1), + threeMessages.get(2)); + }) + .consumeNextWith(response -> { + assertThat(response.successful()).describedAs("successful").isTrue(); + assertThat(response.items()).containsExactly( + twoMessages.get(0), + twoMessages.get(1)); + }) + .expectComplete() + .verify(TIMEOUT); } }
\ No newline at end of file |