gerrit.onap Code Review - ccsdk/oran.git/blob

   1 /*-
   2  * ========================LICENSE_START=================================
   3  * ONAP : ccsdk oran
   4  * ======================================================================
   5  * Copyright (C) 2020 Nordix Foundation. 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
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  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===================================
  19  */
  20
  21 package org.onap.ccsdk.oran.a1policymanagementservice.dmaap;
  22
  23 import com.google.gson.Gson;
  24 import com.google.gson.GsonBuilder;
  25 import com.google.gson.JsonArray;
  26 import com.google.gson.JsonElement;
  27 import com.google.gson.JsonParser;
  28
  29 import java.time.Duration;
  30 import java.util.ArrayList;
  31 import java.util.List;
  32
  33 import org.onap.ccsdk.oran.a1policymanagementservice.clients.AsyncRestClient;
  34 import org.onap.ccsdk.oran.a1policymanagementservice.clients.AsyncRestClientFactory;
  35 import org.onap.ccsdk.oran.a1policymanagementservice.clients.SecurityContext;
  36 import org.onap.ccsdk.oran.a1policymanagementservice.configuration.ApplicationConfig;
  37 import org.slf4j.Logger;
  38 import org.slf4j.LoggerFactory;
  39 import org.springframework.beans.factory.annotation.Autowired;
  40 import org.springframework.beans.factory.annotation.Value;
  41 import org.springframework.http.HttpStatus;
  42 import org.springframework.stereotype.Component;
  43
  44 import reactor.core.publisher.Flux;
  45 import reactor.core.publisher.FluxSink;
  46 import reactor.core.publisher.Mono;
  47
  48 /**
  49  * The class fetches incoming requests from DMAAP. It uses the timeout parameter
  50  * that lets the MessageRouter keep the connection with the Kafka open until
  51  * requests are sent in.
  52  *
  53  * <p>
  54  * this service will regularly check the configuration and start polling DMaaP
  55  * if the configuration is added. If the DMaaP configuration is removed, then
  56  * the service will stop polling and resume checking for configuration.
  57  *
  58  * <p>
  59  * Each received request is processed by {@link DmaapMessageHandler}.
  60  */
  61 @Component
  62 public class DmaapMessageConsumer {
  63
  64     protected static final Duration TIME_BETWEEN_DMAAP_RETRIES = Duration.ofSeconds(10);
  65
  66     private static final Logger logger = LoggerFactory.getLogger(DmaapMessageConsumer.class);
  67
  68     private final ApplicationConfig applicationConfig;
  69
  70     private DmaapMessageHandler dmaapMessageHandler = null;
  71
  72     private final Gson gson;
  73
  74     private final AsyncRestClientFactory restClientFactory;
  75
  76     private final InfiniteFlux infiniteSubmitter = new InfiniteFlux();
  77
  78     @Value("${server.http-port}")
  79     private int localServerHttpPort;
  80
  81     private static class InfiniteFlux {
  82         private FluxSink<Integer> sink;
  83         private int counter = 0;
  84
  85         public synchronized Flux<Integer> start() {
  86             stop();
  87             return Flux.create(this::next).doOnRequest(this::onRequest);
  88         }
  89
  90         public synchronized void stop() {
  91             if (this.sink != null) {
  92                 this.sink.complete();
  93                 this.sink = null;
  94             }
  95         }
  96
  97         void onRequest(long no) {
  98             for (long i = 0; i < no; ++i) {
  99                 sink.next(counter++);
 100             }
 101         }
 102
 103         void next(FluxSink<Integer> sink) {
 104             this.sink = sink;
 105             sink.next(counter++);
 106         }
 107
 108     }
 109
 110     @Autowired
 111     public DmaapMessageConsumer(ApplicationConfig applicationConfig, SecurityContext securityContext) {
 112         this.applicationConfig = applicationConfig;
 113         GsonBuilder gsonBuilder = new GsonBuilder();
 114         this.gson = gsonBuilder.create();
 115         this.restClientFactory = new AsyncRestClientFactory(applicationConfig.getWebClientConfig(), securityContext);
 116     }
 117
 118     /**
 119      * Starts the DMAAP consumer. If there is a DMaaP configuration, it will start
 120      * polling for messages. Otherwise it will check regularly for the
 121      * configuration.
 122      *
 123      */
 124     public void start() {
 125         infiniteSubmitter.stop();
 126
 127         createTask().subscribe(//
 128                 value -> logger.debug("DmaapMessageConsumer next: {}", value), //
 129                 throwable -> logger.error("DmaapMessageConsumer error: {}", throwable.getMessage()), //
 130                 () -> logger.warn("DmaapMessageConsumer stopped") //
 131         );
 132     }
 133
 134     protected Flux<String> createTask() {
 135         return infiniteFlux() //
 136                 .flatMap(notUsed -> fetchFromDmaap(), 1) //
 137                 .doOnNext(message -> logger.debug("Message Reveived from DMAAP : {}", message)) //
 138                 .flatMap(this::parseReceivedMessage, 1)//
 139                 .flatMap(this::handleDmaapMsg, 1) //
 140                 .onErrorResume(throwable -> Mono.empty());
 141     }
 142
 143     protected Flux<Integer> infiniteFlux() {
 144         return infiniteSubmitter.start();
 145     }
 146
 147     protected Mono<Object> delay() {
 148         return Mono.delay(TIME_BETWEEN_DMAAP_RETRIES).flatMap(o -> Mono.empty());
 149     }
 150
 151     private <T> List<T> parseList(String jsonString, Class<T> clazz) {
 152         List<T> result = new ArrayList<>();
 153         JsonArray jsonArr = JsonParser.parseString(jsonString).getAsJsonArray();
 154         for (JsonElement jsonElement : jsonArr) {
 155             // The element can either be a JsonObject or a JsonString
 156             if (jsonElement.isJsonPrimitive()) {
 157                 T json = gson.fromJson(jsonElement.getAsString(), clazz);
 158                 result.add(json);
 159             } else {
 160                 T json = gson.fromJson(jsonElement.toString(), clazz);
 161                 result.add(json);
 162             }
 163         }
 164         return result;
 165     }
 166
 167     protected boolean isDmaapConfigured() {
 168         String producerTopicUrl = applicationConfig.getDmaapProducerTopicUrl();
 169         String consumerTopicUrl = applicationConfig.getDmaapConsumerTopicUrl();
 170         return (producerTopicUrl != null && consumerTopicUrl != null && !producerTopicUrl.isEmpty()
 171                 && !consumerTopicUrl.isEmpty());
 172     }
 173
 174     protected Mono<String> handleDmaapMsg(DmaapRequestMessage dmaapRequestMessage) {
 175         return getDmaapMessageHandler().handleDmaapMsg(dmaapRequestMessage);
 176     }
 177
 178     protected Mono<String> getFromMessageRouter(String topicUrl) {
 179         logger.trace("getFromMessageRouter {}", topicUrl);
 180         AsyncRestClient c = restClientFactory.createRestClientNoHttpProxy("");
 181         return c.get(topicUrl);
 182     }
 183
 184     protected Flux<DmaapRequestMessage> parseReceivedMessage(String jsonString) {
 185         try {
 186             logger.trace("parseMessages {}", jsonString);
 187             return Flux.fromIterable(parseList(jsonString, DmaapRequestMessage.class));
 188         } catch (Exception e) {
 189             logger.error("parseMessages error {}", jsonString);
 190             return sendErrorResponse("Could not parse: " + jsonString) //
 191                     .flatMapMany(s -> Flux.empty());
 192         }
 193     }
 194
 195     protected Mono<String> sendErrorResponse(String response) {
 196         logger.debug("sendErrorResponse {}", response);
 197         DmaapRequestMessage fakeRequest = DmaapRequestMessage.builder() //
 198                 .apiVersion("") //
 199                 .correlationId("") //
 200                 .operation(DmaapRequestMessage.Operation.PUT) //
 201                 .originatorId("") //
 202                 .payload(null) //
 203                 .requestId("") //
 204                 .target("") //
 205                 .timestamp("") //
 206                 .url("URL") //
 207                 .build();
 208         return getDmaapMessageHandler().sendDmaapResponse(response, fakeRequest, HttpStatus.BAD_REQUEST) //
 209                 .onErrorResume(e -> Mono.empty());
 210     }
 211
 212     private Mono<String> fetchFromDmaap() {
 213         if (!this.isDmaapConfigured()) {
 214             return delay().flatMap(o -> Mono.empty());
 215         }
 216         logger.debug("fetchFromDmaap");
 217         String topicUrl = this.applicationConfig.getDmaapConsumerTopicUrl();
 218
 219         return getFromMessageRouter(topicUrl) //
 220                 .onErrorResume(throwable -> delay().flatMap(o -> Mono.empty()));
 221     }
 222
 223     private DmaapMessageHandler getDmaapMessageHandler() {
 224         if (this.dmaapMessageHandler == null) {
 225             String pmsBaseUrl = "http://localhost:" + this.localServerHttpPort;
 226             AsyncRestClient pmsClient = restClientFactory.createRestClientNoHttpProxy(pmsBaseUrl);
 227             AsyncRestClient producer =
 228                     restClientFactory.createRestClientNoHttpProxy(this.applicationConfig.getDmaapProducerTopicUrl());
 229             this.dmaapMessageHandler = new DmaapMessageHandler(producer, pmsClient);
 230         }
 231         return this.dmaapMessageHandler;
 232     }
 233
 234 }