2 * ============LICENSE_START=======================================================
3 * DCAEGEN2-SERVICES-SDK
4 * ================================================================================
5 * Copyright (C) 2018 NOKIA Intellectual Property. 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.service.consumer;
23 import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.REQUEST_ID;
24 import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.X_INVOCATION_ID;
25 import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.X_ONAP_REQUEST_ID;
28 import java.util.UUID;
29 import java.util.function.Consumer;
30 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
32 import org.springframework.http.HttpHeaders;
33 import org.springframework.http.HttpStatus;
34 import org.springframework.web.reactive.function.client.WebClient;
35 import org.springframework.web.util.DefaultUriBuilderFactory;
36 import reactor.core.publisher.Mono;
39 * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 6/26/18
41 public class DMaaPConsumerReactiveHttpClient {
43 private final String dmaapHostName;
44 private final String dmaapProtocol;
45 private final Integer dmaapPortNumber;
46 private final String dmaapTopicName;
47 private final String consumerGroup;
48 private final String consumerId;
49 private final String contentType;
50 private final WebClient webClient;
53 * Constructor of DMaaPConsumerReactiveHttpClient.
55 * @param consumerConfiguration - DMaaP consumer configuration object
57 DMaaPConsumerReactiveHttpClient(DmaapConsumerConfiguration consumerConfiguration, WebClient webClient) {
58 this.dmaapHostName = consumerConfiguration.dmaapHostName();
59 this.dmaapProtocol = consumerConfiguration.dmaapProtocol();
60 this.dmaapPortNumber = consumerConfiguration.dmaapPortNumber();
61 this.dmaapTopicName = consumerConfiguration.dmaapTopicName();
62 this.consumerGroup = consumerConfiguration.consumerGroup();
63 this.consumerId = consumerConfiguration.consumerId();
64 this.contentType = consumerConfiguration.dmaapContentType();
65 this.webClient = webClient;
69 * Function for calling DMaaP HTTP consumer - consuming messages from Kafka/DMaaP from topic.
71 * @return reactive response from DMaaP in string format
73 public Mono<String> getDMaaPConsumerResponse() {
77 .headers(getHeaders())
79 .onStatus(HttpStatus::is4xxClientError, clientResponse ->
80 Mono.error(new RuntimeException("DmaaPConsumer HTTP " + clientResponse.statusCode()))
82 .onStatus(HttpStatus::is5xxServerError, clientResponse ->
83 Mono.error(new RuntimeException("DmaaPConsumer HTTP " + clientResponse.statusCode())))
84 .bodyToMono(String.class);
87 private Consumer<HttpHeaders> getHeaders() {
88 return httpHeaders -> {
89 httpHeaders.set(X_ONAP_REQUEST_ID, MDC.get(REQUEST_ID));
90 httpHeaders.set(X_INVOCATION_ID, UUID.randomUUID().toString());
91 httpHeaders.set(HttpHeaders.CONTENT_TYPE, contentType);
95 private String createRequestPath() {
96 return dmaapTopicName + "/" + consumerGroup + "/" + consumerId;
101 return new DefaultUriBuilderFactory().builder().scheme(dmaapProtocol).host(dmaapHostName).port(dmaapPortNumber)
102 .path(createRequestPath()).build();