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             logger.debug("InfiniteFlux.onRequest {}", no);
  99             for (long i = 0; i < no; ++i) {
 100                 sink.next(counter++);
 101             }
 102         }
 103
 104         void next(FluxSink<Integer> sink) {
 105             logger.debug("InfiniteFlux.next");
 106             this.sink = sink;
 107             sink.next(counter++);
 108         }
 109
 110     }
 111
 112     @Autowired
 113     public DmaapMessageConsumer(ApplicationConfig applicationConfig, SecurityContext securityContext) {
 114         this.applicationConfig = applicationConfig;
 115         GsonBuilder gsonBuilder = new GsonBuilder();
 116         this.gson = gsonBuilder.create();
 117         this.restClientFactory = new AsyncRestClientFactory(applicationConfig.getWebClientConfig(), securityContext);
 118     }
 119
 120     /**
 121      * Starts the DMAAP consumer. If there is a DMaaP configuration, it will start
 122      * polling for messages. Otherwise it will check regularly for the
 123      * configuration.
 124      *
 125      */
 126     public void start() {
 127         infiniteSubmitter.stop();
 128
 129         createTask().subscribe(//
 130                 value -> logger.debug("DmaapMessageConsumer next: {}", value), //
 131                 throwable -> logger.error("DmaapMessageConsumer error: {}", throwable.getMessage()), //
 132                 () -> logger.warn("DmaapMessageConsumer stopped") //
 133         );
 134     }
 135
 136     protected Flux<String> createTask() {
 137         return infiniteFlux() //
 138                 .flatMap(notUsed -> fetchFromDmaap(), 1) //
 139                 .doOnNext(message -> logger.debug("Message Reveived from DMAAP : {}", message)) //
 140                 .flatMap(this::parseReceivedMessage, 1)//
 141                 .flatMap(this::handleDmaapMsg, 1) //
 142                 .onErrorResume(throwable -> Mono.empty());
 143     }
 144
 145     protected Flux<Integer> infiniteFlux() {
 146         return infiniteSubmitter.start();
 147     }
 148
 149     protected Mono<Object> delay() {
 150         return Mono.delay(TIME_BETWEEN_DMAAP_RETRIES).flatMap(o -> Mono.empty());
 151     }
 152
 153     private <T> List<T> parseList(String jsonString, Class<T> clazz) {
 154         List<T> result = new ArrayList<>();
 155         JsonArray jsonArr = JsonParser.parseString(jsonString).getAsJsonArray();
 156         for (JsonElement jsonElement : jsonArr) {
 157             // The element can either be a JsonObject or a JsonString
 158             if (jsonElement.isJsonPrimitive()) {
 159                 T json = gson.fromJson(jsonElement.getAsString(), clazz);
 160                 result.add(json);
 161             } else {
 162                 T json = gson.fromJson(jsonElement.toString(), clazz);
 163                 result.add(json);
 164             }
 165         }
 166         return result;
 167     }
 168
 169     protected boolean isDmaapConfigured() {
 170         String producerTopicUrl = applicationConfig.getDmaapProducerTopicUrl();
 171         String consumerTopicUrl = applicationConfig.getDmaapConsumerTopicUrl();
 172         return (producerTopicUrl != null && consumerTopicUrl != null && !producerTopicUrl.isEmpty()
 173                 && !consumerTopicUrl.isEmpty());
 174     }
 175
 176     protected Mono<String> handleDmaapMsg(DmaapRequestMessage dmaapRequestMessage) {
 177         return getDmaapMessageHandler().handleDmaapMsg(dmaapRequestMessage);
 178     }
 179
 180     protected Mono<String> getFromMessageRouter(String topicUrl) {
 181         logger.trace("getFromMessageRouter {}", topicUrl);
 182         AsyncRestClient c = restClientFactory.createRestClientNoHttpProxy("");
 183         return c.get(topicUrl);
 184     }
 185
 186     protected Flux<DmaapRequestMessage> parseReceivedMessage(String jsonString) {
 187         try {
 188             logger.trace("parseMessages {}", jsonString);
 189             return Flux.fromIterable(parseList(jsonString, DmaapRequestMessage.class));
 190         } catch (Exception e) {
 191             logger.error("parseMessages error {}", jsonString);
 192             return sendErrorResponse("Could not parse: " + jsonString) //
 193                     .flatMapMany(s -> Flux.empty());
 194         }
 195     }
 196
 197     protected Mono<String> sendErrorResponse(String response) {
 198         logger.debug("sendErrorResponse {}", response);
 199         DmaapRequestMessage fakeRequest = DmaapRequestMessage.builder() //
 200                 .apiVersion("") //
 201                 .correlationId("") //
 202                 .operation(DmaapRequestMessage.Operation.PUT) //
 203                 .originatorId("") //
 204                 .payload(null) //
 205                 .requestId("") //
 206                 .target("") //
 207                 .timestamp("") //
 208                 .url("URL") //
 209                 .build();
 210         return getDmaapMessageHandler().sendDmaapResponse(response, fakeRequest, HttpStatus.BAD_REQUEST) //
 211                 .onErrorResume(e -> Mono.empty());
 212     }
 213
 214     private Mono<String> fetchFromDmaap() {
 215         if (!this.isDmaapConfigured()) {
 216             logger.debug("fetchFromDmaap, no action DMAAP not configured");
 217             return delay().flatMap(o -> Mono.empty());
 218         }
 219         logger.debug("fetchFromDmaap");
 220         String topicUrl = this.applicationConfig.getDmaapConsumerTopicUrl();
 221
 222         return getFromMessageRouter(topicUrl) //
 223                 .onErrorResume(throwable -> delay().flatMap(o -> Mono.empty()));
 224     }
 225
 226     private DmaapMessageHandler getDmaapMessageHandler() {
 227         if (this.dmaapMessageHandler == null) {
 228             String pmsBaseUrl = "http://localhost:" + this.localServerHttpPort;
 229             AsyncRestClient pmsClient = restClientFactory.createRestClientNoHttpProxy(pmsBaseUrl);
 230             AsyncRestClient producer =
 231                     restClientFactory.createRestClientNoHttpProxy(this.applicationConfig.getDmaapProducerTopicUrl());
 232             this.dmaapMessageHandler = new DmaapMessageHandler(producer, pmsClient);
 233         }
 234         return this.dmaapMessageHandler;
 235     }
 236
 237 }