2 * ============LICENSE_START====================================
3 * DCAEGEN2-SERVICES-SDK
4 * =========================================================
5 * Copyright (C) 2019-2021 Nokia. All rights reserved.
6 * =========================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=====================================
21 package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl;
23 import com.google.gson.Gson;
24 import com.google.gson.JsonArray;
25 import com.google.gson.JsonElement;
26 import com.google.gson.JsonParser;
27 import io.netty.handler.timeout.ReadTimeoutException;
28 import io.vavr.collection.List;
29 import io.vavr.control.Option;
30 import org.jetbrains.annotations.NotNull;
31 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpMethod;
32 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpRequest;
33 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
34 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.ImmutableHttpRequest;
35 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
36 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterSubscriber;
37 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.error.ClientErrorReason;
38 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.error.ClientErrorReasonPresenter;
39 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.error.ClientErrorReasons;
40 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeResponse;
41 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeRequest;
42 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeResponse;
43 import org.slf4j.Logger;
44 import org.slf4j.LoggerFactory;
45 import reactor.core.publisher.Mono;
47 import java.nio.charset.StandardCharsets;
49 import static org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.Commons.extractFailReason;
52 * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
55 public class MessageRouterSubscriberImpl implements MessageRouterSubscriber {
56 private final RxHttpClient httpClient;
57 private final Gson gson;
58 private final ClientErrorReasonPresenter clientErrorReasonPresenter;
59 private static final Logger LOGGER = LoggerFactory.getLogger(MessageRouterSubscriberImpl.class);
61 public MessageRouterSubscriberImpl(RxHttpClient httpClient, Gson gson,
62 ClientErrorReasonPresenter clientErrorReasonPresenter) {
63 this.httpClient = httpClient;
65 this.clientErrorReasonPresenter = clientErrorReasonPresenter;
69 public Mono<MessageRouterSubscribeResponse> get(MessageRouterSubscribeRequest request) {
70 LOGGER.debug("Requesting new items from DMaaP MR: {}", request);
71 return httpClient.call(buildGetHttpRequest(request))
72 .map(this::buildGetResponse)
73 .doOnError(ReadTimeoutException.class,
74 e -> LOGGER.error("Timeout exception occurred when subscribe items from DMaaP MR", e))
75 .onErrorResume(ReadTimeoutException.class, e -> createErrorResponse(ClientErrorReasons.TIMEOUT));
79 private @NotNull HttpRequest buildGetHttpRequest(MessageRouterSubscribeRequest request) {
80 ImmutableHttpRequest.Builder requestBuilder = ImmutableHttpRequest.builder()
81 .method(HttpMethod.GET)
82 .url(buildSubscribeUrl(request))
83 .diagnosticContext(request.diagnosticContext().withNewInvocationId());
85 return Option.of(request.timeoutConfig())
86 .map(timeoutConfig -> requestBuilder.timeout(timeoutConfig.getTimeout()).build())
87 .getOrElse(requestBuilder::build);
90 private @NotNull MessageRouterSubscribeResponse buildGetResponse(HttpResponse httpResponse) {
91 final ImmutableMessageRouterSubscribeResponse.Builder builder =
92 ImmutableMessageRouterSubscribeResponse.builder();
93 return httpResponse.successful()
94 ? builder.items(getAsJsonElements(httpResponse)).build()
95 : builder.failReason(extractFailReason(httpResponse)).build();
98 private List<JsonElement> getAsJsonElements(HttpResponse httpResponse) {
99 JsonArray bodyAsJsonArray = httpResponse
100 .bodyAsJson(StandardCharsets.UTF_8, gson, JsonArray.class);
102 return List.ofAll(bodyAsJsonArray).map(arrayElement -> JsonParser.parseString(arrayElement.getAsString()));
105 private String buildSubscribeUrl(MessageRouterSubscribeRequest request) {
106 return String.format("%s/%s/%s", request.sourceDefinition().topicUrl(), request.consumerGroup(),
107 request.consumerId());
110 private Mono<MessageRouterSubscribeResponse> createErrorResponse(ClientErrorReason clientErrorReason) {
111 String failReason = clientErrorReasonPresenter.present(clientErrorReason);
112 return Mono.just(ImmutableMessageRouterSubscribeResponse.builder()
113 .failReason(failReason)