diff options
Diffstat (limited to 'rest-services/dmaap-client')
9 files changed, 394 insertions, 117 deletions
diff --git a/rest-services/dmaap-client/pom.xml b/rest-services/dmaap-client/pom.xml index 703d337a..c465cef4 100644 --- a/rest-services/dmaap-client/pom.xml +++ b/rest-services/dmaap-client/pom.xml @@ -46,7 +46,6 @@ <dependency> <groupId>org.junit.jupiter</groupId> <artifactId>junit-jupiter-engine</artifactId> - <scope>test</scope> </dependency> <dependency> <groupId>io.projectreactor</groupId> @@ -56,7 +55,10 @@ <dependency> <groupId>org.mockito</groupId> <artifactId>mockito-core</artifactId> - <scope>test</scope> + </dependency> + <dependency> + <groupId>org.assertj</groupId> + <artifactId>assertj-core</artifactId> </dependency> <dependency> <groupId>org.onap.dcaegen2.services.sdk.rest.services</groupId> diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/DmaapClientFactory.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/DmaapClientFactory.java index 7eb72f7c..0ac2d0bd 100644 --- a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/DmaapClientFactory.java +++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/DmaapClientFactory.java @@ -21,11 +21,15 @@ package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api; import com.google.gson.Gson; +import io.netty.handler.ssl.SslContext; import io.vavr.Lazy; +import java.time.Duration; import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient; import org.onap.dcaegen2.services.sdk.rest.services.annotations.ExperimentalApi; -import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.MessageRouterClientImpl; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.MessageRouterPublisherImpl; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.MessageRouterSubscriberImpl; /** * <b>WARNING</b>: This is a proof-of-concept. It is untested. API may change or be removed. Use at your own risk. @@ -37,17 +41,33 @@ import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.MessageRou @ExperimentalApi public final class DmaapClientFactory { - private static final Lazy<MessageRouterClientImpl> THE_CLIENT = Lazy.of(() -> - new MessageRouterClientImpl(RxHttpClient.create(), new Gson())); + private static final Duration DEFAULT_MAX_BATCH_DURATION = Duration.ofSeconds(1); + private static final int DEFAULT_MAX_BATCH_SIZE = 512; private DmaapClientFactory() { } public static @NotNull MessageRouterPublisher createMessageRouterPublisher() { - return THE_CLIENT.get(); + return new MessageRouterPublisherImpl( + RxHttpClient.create(), + DEFAULT_MAX_BATCH_SIZE, + DEFAULT_MAX_BATCH_DURATION); + } + + public static @NotNull MessageRouterPublisher createMessageRouterPublisher(@NotNull SslContext sslContext) { + return new MessageRouterPublisherImpl( + RxHttpClient.create(sslContext), + DEFAULT_MAX_BATCH_SIZE, + DEFAULT_MAX_BATCH_DURATION); } public static @NotNull MessageRouterSubscriber createMessageRouterSubscriber() { - return THE_CLIENT.get(); + return new MessageRouterSubscriberImpl(RxHttpClient.create(), new Gson()); + } + + public static @NotNull MessageRouterSubscriber createMessageRouterSubscriber(@NotNull SslContext sslContext) { + return new MessageRouterSubscriberImpl( + RxHttpClient.create(sslContext), + new Gson()); } } diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/Commons.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/Commons.java new file mode 100644 index 00000000..5211807a --- /dev/null +++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/Commons.java @@ -0,0 +1,38 @@ +/* + * ============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.impl; + +import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse; + +/** + * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a> + * @since April 2019 + */ +final class Commons { + + private Commons() { + } + + static String extractFailReason(HttpResponse httpResponse) { + return String.format("%d %s%n%s", httpResponse.statusCode(), httpResponse.statusReason(), + httpResponse.bodyAsString()); + } +} 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 new file mode 100644 index 00000000..f09c5397 --- /dev/null +++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImpl.java @@ -0,0 +1,101 @@ +/* + * ============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.impl; + +import static org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.Commons.extractFailReason; + +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import io.vavr.collection.HashMap; +import io.vavr.collection.List; +import java.time.Duration; +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; +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.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.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; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse; +import org.reactivestreams.Publisher; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a> + * @since March 2019 + */ +// TODO: This is a PoC. It's untested. +public class MessageRouterPublisherImpl implements MessageRouterPublisher { + private final RxHttpClient httpClient; + private final int maxBatchSize; + private final Duration maxBatchDuration; + + public MessageRouterPublisherImpl(RxHttpClient httpClient, int maxBatchSize, Duration maxBatchDuration) { + this.httpClient = httpClient; + this.maxBatchSize = maxBatchSize; + this.maxBatchDuration = maxBatchDuration; + } + + @Override + public Flux<MessageRouterPublishResponse> put( + MessageRouterPublishRequest request, + Flux<? extends JsonElement> items) { + return items.bufferTimeout(maxBatchSize, maxBatchDuration) + .flatMap(subItems -> subItems.isEmpty() ? Mono.empty() : pushBatchToMr(request, List.ofAll(subItems))); + } + + private Publisher<? extends MessageRouterPublishResponse> pushBatchToMr( + MessageRouterPublishRequest request, + List<JsonElement> batch) { + return httpClient.call(buildHttpRequest(request, asJsonBody(batch))) + .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 HttpRequest buildHttpRequest(MessageRouterPublishRequest request, RequestBody body) { + return ImmutableHttpRequest.builder() + .method(HttpMethod.POST) + .url(request.sinkDefinition().topicUrl()) + .diagnosticContext(request.diagnosticContext()) + .customHeaders(HashMap.of(HttpHeaders.CONTENT_TYPE, request.contentType())) + .body(body) + .build(); + } + + private MessageRouterPublishResponse buildResponse( + 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/impl/MessageRouterClientImpl.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterSubscriberImpl.java index 0ef06201..e91a77fa 100644 --- a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterClientImpl.java +++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterSubscriberImpl.java @@ -20,31 +20,21 @@ package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl; +import static org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.Commons.extractFailReason; + import com.google.gson.Gson; import com.google.gson.JsonArray; -import com.google.gson.JsonElement; -import io.netty.buffer.ByteBuf; -import io.vavr.collection.HashMap; import java.nio.charset.StandardCharsets; -import java.time.Duration; 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; 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.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.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.ImmutableMessageRouterPublishResponse; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeResponse; -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; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeResponse; -import org.reactivestreams.Publisher; -import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; /** @@ -52,42 +42,20 @@ import reactor.core.publisher.Mono; * @since March 2019 */ // TODO: This is a PoC. It's untested. -public class MessageRouterClientImpl implements MessageRouterPublisher, MessageRouterSubscriber { - - private static final Duration WINDOW_MAX_TIME = Duration.ofSeconds(1); - private static final int WINDOW_MAX_SIZE = 512; +public class MessageRouterSubscriberImpl implements MessageRouterSubscriber { private final RxHttpClient httpClient; private final Gson gson; - public MessageRouterClientImpl(RxHttpClient httpClient, Gson gson) { + public MessageRouterSubscriberImpl(RxHttpClient httpClient, Gson gson) { this.httpClient = httpClient; this.gson = gson; } @Override - public Flux<MessageRouterPublishResponse> put( - MessageRouterPublishRequest request, - Flux<? extends JsonElement> items) { - return items.windowTimeout(WINDOW_MAX_SIZE, WINDOW_MAX_TIME).flatMap(subItems -> - subItems.collect(JsonArray::new, JsonArray::add) - .filter(arr -> arr.size() > 0) - .map(RequestBody::fromJson) - .flatMap(body -> httpClient.call(buildPostHttpRequest(request, body))) - .map(this::buildPutResponse)); - } - - @Override public Mono<MessageRouterSubscribeResponse> get(MessageRouterSubscribeRequest request) { return httpClient.call(buildGetHttpRequest(request)).map(this::buildGetResponse); } - private @NotNull MessageRouterPublishResponse buildPutResponse(HttpResponse httpResponse) { - final ImmutableMessageRouterPublishResponse.Builder builder = - ImmutableMessageRouterPublishResponse.builder(); - return httpResponse.successful() - ? builder.build() - : builder.failReason(extractFailReason(httpResponse)).build(); - } private @NotNull MessageRouterSubscribeResponse buildGetResponse(HttpResponse httpResponse) { final ImmutableMessageRouterSubscribeResponse.Builder builder = @@ -97,20 +65,6 @@ public class MessageRouterClientImpl implements MessageRouterPublisher, MessageR : builder.failReason(extractFailReason(httpResponse)).build(); } - private String extractFailReason(HttpResponse httpResponse) { - return String.format("%d %s%n%s", httpResponse.statusCode(), httpResponse.statusReason(), - httpResponse.bodyAsString()); - } - - private @NotNull HttpRequest buildPostHttpRequest(MessageRouterPublishRequest request, RequestBody body) { - return ImmutableHttpRequest.builder() - .method(HttpMethod.POST) - .url(request.sinkDefinition().topicUrl()) - .diagnosticContext(request.diagnosticContext()) - .customHeaders(HashMap.of(HttpHeaders.CONTENT_TYPE, request.contentType())) - .body(body) - .build(); - } private @NotNull HttpRequest buildGetHttpRequest(MessageRouterSubscribeRequest request) { return ImmutableHttpRequest.builder() @@ -121,6 +75,7 @@ public class MessageRouterClientImpl implements MessageRouterPublisher, MessageR } private String buildSubscribeUrl(MessageRouterSubscribeRequest request) { - return String.format("%s/%s/%s", request.sourceDefinition().topicUrl(), request.consumerGroup(), request.consumerId()); + return String.format("%s/%s/%s", request.sourceDefinition().topicUrl(), request.consumerGroup(), + request.consumerId()); } } diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/MessageRouterPublishResponse.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/MessageRouterPublishResponse.java index 62175e09..cc038a6e 100644 --- a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/MessageRouterPublishResponse.java +++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/MessageRouterPublishResponse.java @@ -21,6 +21,8 @@ package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model; +import com.google.gson.JsonElement; +import io.vavr.collection.List; import org.immutables.value.Value; import org.onap.dcaegen2.services.sdk.rest.services.annotations.ExperimentalApi; @@ -32,4 +34,8 @@ import org.onap.dcaegen2.services.sdk.rest.services.annotations.ExperimentalApi; @Value.Immutable public interface MessageRouterPublishResponse extends DmaapResponse { + @Value.Default + default List<JsonElement> items() { + return List.empty(); + } } diff --git a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherTest.java b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherTest.java deleted file mode 100644 index 9656ae86..00000000 --- a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherTest.java +++ /dev/null @@ -1,57 +0,0 @@ -/* - * ============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.api; - -import static org.mockito.Mockito.mock; - -import com.google.gson.JsonPrimitive; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.Disabled; -import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink; -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 reactor.core.publisher.Flux; - -/** - * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a> - * @since March 2019 - */ -@Disabled -class MessageRouterPublisherTest { - - private final MessageRouterPublisher cut = mock(MessageRouterPublisher.class); - private final MessageRouterSink sinkDefinition = mock(MessageRouterSink.class); - private final MessageRouterPublishRequest request = ImmutableMessageRouterPublishRequest.builder() - .sinkDefinition(sinkDefinition).build(); - - @Test - void apiShouldBeUsableWithTransform() { - Flux.just(1, 2, 3) - .map(JsonPrimitive::new) - .transform(input -> cut.put(request, input)); - } - - @Test - void apiShouldBeUsableWithSingleCall() { - final Flux<JsonPrimitive> input = Flux.just(1, 2, 3).map(JsonPrimitive::new); - cut.put(request, input); - } -}
\ No newline at end of file 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 new file mode 100644 index 00000000..103b4807 --- /dev/null +++ b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImplTest.java @@ -0,0 +1,210 @@ +/* + * ============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.impl; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +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 com.google.gson.Gson; +import com.google.gson.JsonArray; +import com.google.gson.JsonObject; +import com.google.gson.JsonPrimitive; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufAllocator; +import io.netty.buffer.CompositeByteBuf; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.List; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.verification.VerificationMode; +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.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.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; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +/** + * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a> + * @since April 2019 + */ +class MessageRouterPublisherImplTest { + + private static final Duration TIMEOUT = Duration.ofSeconds(5); + private final RxHttpClient httpClient = mock(RxHttpClient.class); + private final MessageRouterPublisher cut = new MessageRouterPublisherImpl(httpClient, 3, Duration.ofMinutes(1)); + + private final ArgumentCaptor<HttpRequest> httpRequestArgumentCaptor = ArgumentCaptor.forClass(HttpRequest.class); + private final MessageRouterSink sinkDefinition = ImmutableMessageRouterSink.builder() + .name("the topic") + .topicUrl("https://dmaap-mr/TOPIC") + .build(); + private final MessageRouterPublishRequest mrRequest = ImmutableMessageRouterPublishRequest.builder() + .sinkDefinition(sinkDefinition) + .build(); + private final HttpResponse httpResponse = ImmutableHttpResponse.builder() + .statusCode(200) + .statusReason("OK") + .url(sinkDefinition.topicUrl()) + .rawBody("{}".getBytes()) + .build(); + + @Test + void puttingElementsShouldYieldNonChunkedHttpRequest() { + // given + given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(httpResponse)); + + // when + final Flux<MessageRouterPublishResponse> responses = cut + .put(mrRequest, Flux.just("I", "like", "cookies").map(JsonPrimitive::new)); + 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.body()).isNotNull(); + assertThat(httpRequest.body().length()).isGreaterThan(0); + } + + @Test + void puttingLowNumberOfElementsShouldYieldSingleHttpRequest() { + // given + given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(httpResponse)); + + // when + final Flux<MessageRouterPublishResponse> responses = cut + .put(mrRequest, Flux.just("I", "like", "cookies").map(JsonPrimitive::new)); + responses.then().block(); + + // then + verify(httpClient).call(httpRequestArgumentCaptor.capture()); + final HttpRequest httpRequest = httpRequestArgumentCaptor.getValue(); + final JsonArray elementsInRequest = extractNonEmptyRequestBody(httpRequest); + assertThat(elementsInRequest.size()).isEqualTo(3); + assertThat(elementsInRequest.get(0).getAsString()).isEqualTo("I"); + assertThat(elementsInRequest.get(1).getAsString()).isEqualTo("like"); + assertThat(elementsInRequest.get(2).getAsString()).isEqualTo("cookies"); + } + + @Test + void puttingLowNumberOfElementsShouldReturnSingleResponse() { + // given + given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(httpResponse)); + + // when + final Flux<MessageRouterPublishResponse> responses = cut + .put(mrRequest, Flux.just("I", "like", "cookies").map(JsonPrimitive::new)); + + // then + StepVerifier.create(responses) + .consumeNextWith(response -> { + assertThat(response.successful()).describedAs("successful").isTrue(); + assertThat(response.items()).containsExactly( + new JsonPrimitive("I"), + new JsonPrimitive("like"), + new JsonPrimitive("cookies")); + }) + .expectComplete() + .verify(TIMEOUT); + } + + + @Test + void puttingHighNumberOfElementsShouldYieldMultipleHttpRequests() { + // given + given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(httpResponse)); + + // when + final Flux<MessageRouterPublishResponse> responses = cut + .put(mrRequest, Flux.just("I", "like", "cookies", "and", "pierogi").map(JsonPrimitive::new)); + + // then + responses.then().block(); + + verify(httpClient, times(2)).call(httpRequestArgumentCaptor.capture()); + final List<HttpRequest> httpRequests = httpRequestArgumentCaptor.getAllValues(); + assertThat(httpRequests.size()).describedAs("number of requests").isEqualTo(2); + + final JsonArray firstRequest = extractNonEmptyRequestBody(httpRequests.get(0)); + assertThat(firstRequest.size()).isEqualTo(3); + assertThat(firstRequest.get(0).getAsString()).isEqualTo("I"); + assertThat(firstRequest.get(1).getAsString()).isEqualTo("like"); + assertThat(firstRequest.get(2).getAsString()).isEqualTo("cookies"); + + final JsonArray secondRequest = extractNonEmptyRequestBody(httpRequests.get(1)); + assertThat(secondRequest.size()).isEqualTo(2); + assertThat(secondRequest.get(0).getAsString()).isEqualTo("and"); + assertThat(secondRequest.get(1).getAsString()).isEqualTo("pierogi"); + } + + @Test + void puttingHighNumberOfElementsShouldReturnMoreResponses() { + // given + given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(httpResponse)); + + // when + final Flux<MessageRouterPublishResponse> responses = cut + .put(mrRequest, Flux.just("I", "like", "cookies", "and", "pierogi").map(JsonPrimitive::new)); + + // then + StepVerifier.create(responses) + .consumeNextWith(response -> { + assertThat(response.successful()).describedAs("successful").isTrue(); + assertThat(response.items()).containsExactly( + new JsonPrimitive("I"), + new JsonPrimitive("like"), + new JsonPrimitive("cookies")); + }) + .consumeNextWith(response -> { + assertThat(response.successful()).describedAs("successful").isTrue(); + assertThat(response.items()).containsExactly( + new JsonPrimitive("and"), + new JsonPrimitive("pierogi")); + }) + .expectComplete() + .verify(TIMEOUT); + } + + private JsonArray extractNonEmptyRequestBody(HttpRequest httpRequest) { + final String body = Flux.from(httpRequest.body().contents()) + .collect(ByteBufAllocator.DEFAULT::compositeBuffer, + (byteBufs, buffer) -> byteBufs.addComponent(true, buffer)) + .map(byteBufs -> byteBufs.toString(StandardCharsets.UTF_8)) + .block(); + assertThat(body).describedAs("request body").isNotBlank(); + return new Gson().fromJson(body, JsonArray.class); + } +}
\ No newline at end of file diff --git a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberTest.java b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterSubscriberImplTest.java index b8bcde92..c9587206 100644 --- a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberTest.java +++ b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterSubscriberImplTest.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.impl; import static org.mockito.Mockito.mock; @@ -27,6 +27,7 @@ import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink; import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSource; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterSubscriber; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.DmaapResponse; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishRequest; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeRequest; @@ -39,8 +40,9 @@ import reactor.core.publisher.Flux; * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a> * @since March 2019 */ +// TODO: Write proper unit tests @Disabled -class MessageRouterSubscriberTest { +class MessageRouterSubscriberImplTest { private final MessageRouterSubscriber cut = mock(MessageRouterSubscriber.class); private final MessageRouterSource sinkDefinition = mock(MessageRouterSource.class); |