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 static org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.Commons.extractFailReason;
25 import com.google.gson.Gson;
26 import com.google.gson.JsonArray;
27 import com.google.gson.JsonElement;
28 import com.google.gson.JsonParser;
29 import io.netty.handler.timeout.ReadTimeoutException;
30 import io.vavr.collection.List;
31 import java.nio.charset.StandardCharsets;
33 import io.vavr.control.Option;
34 import org.jetbrains.annotations.NotNull;
35 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpMethod;
36 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpRequest;
37 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
38 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.ImmutableHttpRequest;
39 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
40 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterSubscriber;
41 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.error.ClientErrorReason;
42 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.error.ClientErrorReasonPresenter;
43 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.error.ClientErrorReasons;
44 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeResponse;
45 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeRequest;
46 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeResponse;
47 import org.slf4j.Logger;
48 import org.slf4j.LoggerFactory;
49 import reactor.core.publisher.Mono;
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 static final Logger LOGGER = LoggerFactory.getLogger(MessageRouterSubscriberImpl.class);
60 public MessageRouterSubscriberImpl(RxHttpClient httpClient, Gson gson) {
61 this.httpClient = httpClient;
66 public Mono<MessageRouterSubscribeResponse> get(MessageRouterSubscribeRequest request) {
67 LOGGER.debug("Requesting new items from DMaaP MR: {}", request);
68 return httpClient.call(buildGetHttpRequest(request))
69 .map(this::buildGetResponse)
70 .doOnError(ReadTimeoutException.class, e -> LOGGER.error("Timeout exception occurred when subscribe items from DMaaP MR", e))
71 .onErrorResume(ReadTimeoutException.class, e -> createErrorResponse(ClientErrorReasons.TIMEOUT));
75 private @NotNull HttpRequest buildGetHttpRequest(MessageRouterSubscribeRequest request) {
76 ImmutableHttpRequest.Builder requestBuilder = ImmutableHttpRequest.builder()
77 .method(HttpMethod.GET)
78 .url(buildSubscribeUrl(request))
79 .diagnosticContext(request.diagnosticContext().withNewInvocationId());
81 return Option.of(request.timeoutConfig())
82 .map(timeoutConfig -> requestBuilder.timeout(timeoutConfig.getTimeout()).build())
83 .getOrElse(requestBuilder::build);
86 private @NotNull MessageRouterSubscribeResponse buildGetResponse(HttpResponse httpResponse) {
87 final ImmutableMessageRouterSubscribeResponse.Builder builder =
88 ImmutableMessageRouterSubscribeResponse.builder();
89 return httpResponse.successful()
90 ? builder.items(getAsJsonElements(httpResponse)).build()
91 : builder.failReason(extractFailReason(httpResponse)).build();
94 private List<JsonElement> getAsJsonElements(HttpResponse httpResponse){
95 JsonArray bodyAsJsonArray = httpResponse
96 .bodyAsJson(StandardCharsets.UTF_8, gson, JsonArray.class);
98 return List.ofAll(bodyAsJsonArray).map(arrayElement -> JsonParser.parseString(arrayElement.getAsString()));
101 private String buildSubscribeUrl(MessageRouterSubscribeRequest request) {
102 return String.format("%s/%s/%s", request.sourceDefinition().topicUrl(), request.consumerGroup(),
103 request.consumerId());
106 private Mono<MessageRouterSubscribeResponse> createErrorResponse(ClientErrorReason clientErrorReason) {
107 String failReason = ClientErrorReasonPresenter.present(clientErrorReason);
108 return Mono.just(ImmutableMessageRouterSubscribeResponse.builder()
109 .failReason(failReason)