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 + 6 files changed, 251 insertions(+), 131 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 (limited to 'rest-services/dmaap-client/src/main/java') 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(); + } } -- cgit 1.2.3-korg