From e2fc40091989cba8a0fe278712158c8f813d571b Mon Sep 17 00:00:00 2001 From: Piotr Jaszczyk Date: Thu, 11 Apr 2019 09:57:37 +0200 Subject: Enhance dmaap-client v2 * include sent messages in dmaap mr response * split publisher and subscriber * write some unit tests Issue-ID: DCAEGEN2-1421 Change-Id: Ie71e9344efd4a520406b87923413a45d51c28225 Signed-off-by: Piotr Jaszczyk --- .../dmaap/client/api/DmaapClientFactory.java | 30 ++- .../rest/services/dmaap/client/impl/Commons.java | 38 ++++ .../dmaap/client/impl/MessageRouterClientImpl.java | 126 ------------- .../client/impl/MessageRouterPublisherImpl.java | 101 ++++++++++ .../client/impl/MessageRouterSubscriberImpl.java | 81 ++++++++ .../client/model/MessageRouterPublishResponse.java | 6 + .../client/api/MessageRouterPublisherTest.java | 57 ------ .../client/api/MessageRouterSubscriberTest.java | 65 ------- .../impl/MessageRouterPublisherImplTest.java | 210 +++++++++++++++++++++ .../impl/MessageRouterSubscriberImplTest.java | 67 +++++++ 10 files changed, 528 insertions(+), 253 deletions(-) create mode 100644 rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/Commons.java delete mode 100644 rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterClientImpl.java create mode 100644 rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImpl.java create mode 100644 rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterSubscriberImpl.java delete mode 100644 rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherTest.java delete mode 100644 rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberTest.java create mode 100644 rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImplTest.java create mode 100644 rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterSubscriberImplTest.java (limited to 'rest-services/dmaap-client/src') 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; /** * WARNING: 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 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 Piotr Jaszczyk + * @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/MessageRouterClientImpl.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterClientImpl.java deleted file mode 100644 index 0ef06201..00000000 --- a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterClientImpl.java +++ /dev/null @@ -1,126 +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.impl; - -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; - -/** - * @author Piotr Jaszczyk - * @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; - private final RxHttpClient httpClient; - private final Gson gson; - - public MessageRouterClientImpl(RxHttpClient httpClient, Gson gson) { - this.httpClient = httpClient; - this.gson = gson; - } - - @Override - public Flux put( - MessageRouterPublishRequest request, - Flux 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 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 = - ImmutableMessageRouterSubscribeResponse.builder(); - return httpResponse.successful() - ? builder.items(httpResponse.bodyAsJson(StandardCharsets.UTF_8, gson, JsonArray.class)).build() - : 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() - .method(HttpMethod.GET) - .url(buildSubscribeUrl(request)) - .diagnosticContext(request.diagnosticContext()) - .build(); - } - - private String buildSubscribeUrl(MessageRouterSubscribeRequest request) { - 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/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 Piotr Jaszczyk + * @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 put( + MessageRouterPublishRequest request, + Flux items) { + return items.bufferTimeout(maxBatchSize, maxBatchDuration) + .flatMap(subItems -> subItems.isEmpty() ? Mono.empty() : pushBatchToMr(request, List.ofAll(subItems))); + } + + private Publisher pushBatchToMr( + MessageRouterPublishRequest request, + List batch) { + return httpClient.call(buildHttpRequest(request, asJsonBody(batch))) + .map(httpResponse -> buildResponse(httpResponse, batch)); + } + + private @NotNull RequestBody asJsonBody(List 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 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/MessageRouterSubscriberImpl.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterSubscriberImpl.java new file mode 100644 index 00000000..e91a77fa --- /dev/null +++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterSubscriberImpl.java @@ -0,0 +1,81 @@ +/* + * ============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.Gson; +import com.google.gson.JsonArray; +import java.nio.charset.StandardCharsets; +import org.jetbrains.annotations.NotNull; +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.RxHttpClient; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterSubscriber; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeResponse; +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 reactor.core.publisher.Mono; + +/** + * @author Piotr Jaszczyk + * @since March 2019 + */ +// TODO: This is a PoC. It's untested. +public class MessageRouterSubscriberImpl implements MessageRouterSubscriber { + private final RxHttpClient httpClient; + private final Gson gson; + + public MessageRouterSubscriberImpl(RxHttpClient httpClient, Gson gson) { + this.httpClient = httpClient; + this.gson = gson; + } + + @Override + public Mono get(MessageRouterSubscribeRequest request) { + return httpClient.call(buildGetHttpRequest(request)).map(this::buildGetResponse); + } + + + private @NotNull MessageRouterSubscribeResponse buildGetResponse(HttpResponse httpResponse) { + final ImmutableMessageRouterSubscribeResponse.Builder builder = + ImmutableMessageRouterSubscribeResponse.builder(); + return httpResponse.successful() + ? builder.items(httpResponse.bodyAsJson(StandardCharsets.UTF_8, gson, JsonArray.class)).build() + : builder.failReason(extractFailReason(httpResponse)).build(); + } + + + private @NotNull HttpRequest buildGetHttpRequest(MessageRouterSubscribeRequest request) { + return ImmutableHttpRequest.builder() + .method(HttpMethod.GET) + .url(buildSubscribeUrl(request)) + .diagnosticContext(request.diagnosticContext()) + .build(); + } + + private String buildSubscribeUrl(MessageRouterSubscribeRequest request) { + 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 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 Piotr Jaszczyk - * @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 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/api/MessageRouterSubscriberTest.java b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberTest.java deleted file mode 100644 index b8bcde92..00000000 --- a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberTest.java +++ /dev/null @@ -1,65 +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.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.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; -import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest; -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 reactor.core.publisher.Flux; - -/** - * @author Piotr Jaszczyk - * @since March 2019 - */ -@Disabled -class MessageRouterSubscriberTest { - - private final MessageRouterSubscriber cut = mock(MessageRouterSubscriber.class); - private final MessageRouterSource sinkDefinition = mock(MessageRouterSource.class); - private final MessageRouterSubscribeRequest request = ImmutableMessageRouterSubscribeRequest.builder() - .sourceDefinition(sinkDefinition) - .build(); - - @Test - void getShouldBeUsable() { - cut.get(request) - .filter(DmaapResponse::successful) - .map(MessageRouterSubscribeResponse::items) - .subscribe(System.out::println); - } - - @Test - void getElementsShouldBeUsable() { - cut.getElements(request) - .collectList() - .subscribe(System.out::println); - } -} \ 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 Piotr Jaszczyk + * @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 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 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 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 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 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 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 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/impl/MessageRouterSubscriberImplTest.java b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterSubscriberImplTest.java new file mode 100644 index 00000000..c9587206 --- /dev/null +++ b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterSubscriberImplTest.java @@ -0,0 +1,67 @@ +/* + * ============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.mockito.Mockito.mock; + +import com.google.gson.JsonPrimitive; +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; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest; +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 reactor.core.publisher.Flux; + +/** + * @author Piotr Jaszczyk + * @since March 2019 + */ +// TODO: Write proper unit tests +@Disabled +class MessageRouterSubscriberImplTest { + + private final MessageRouterSubscriber cut = mock(MessageRouterSubscriber.class); + private final MessageRouterSource sinkDefinition = mock(MessageRouterSource.class); + private final MessageRouterSubscribeRequest request = ImmutableMessageRouterSubscribeRequest.builder() + .sourceDefinition(sinkDefinition) + .build(); + + @Test + void getShouldBeUsable() { + cut.get(request) + .filter(DmaapResponse::successful) + .map(MessageRouterSubscribeResponse::items) + .subscribe(System.out::println); + } + + @Test + void getElementsShouldBeUsable() { + cut.getElements(request) + .collectList() + .subscribe(System.out::println); + } +} \ No newline at end of file -- cgit 1.2.3-korg