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