gerrit.onap Code Review - dcaegen2/collectors/datafile.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2019 Nordix Foundation.
   4  * ================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  * SPDX-License-Identifier: Apache-2.0
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.dcaegen2.collectors.datafile.tasks;
  22
  23 import com.google.gson.JsonElement;
  24
  25 import java.util.Optional;
  26
  27 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
  28 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
  29 import org.onap.dcaegen2.collectors.datafile.model.FileReadyMessage;
  30 import org.onap.dcaegen2.collectors.datafile.service.JsonMessageParser;
  31 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer.ConsumerReactiveHttpClientFactory;
  32 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer.DMaaPConsumerReactiveHttpClient;
  33 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer.DMaaPReactiveWebClientFactory;
  34 import org.slf4j.Logger;
  35 import org.slf4j.LoggerFactory;
  36
  37 import reactor.core.publisher.Flux;
  38 import reactor.core.publisher.Mono;
  39
  40 /**
  41  * Component used to get messages from the MessageRouter.
  42  *
  43  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  44  */
  45 public class DMaaPMessageConsumer {
  46     private static final Logger logger = LoggerFactory.getLogger(DMaaPMessageConsumer.class);
  47     private final AppConfig datafileAppConfig;
  48     private final JsonMessageParser jsonMessageParser;
  49     private final ConsumerReactiveHttpClientFactory httpClientFactory;
  50
  51     public DMaaPMessageConsumer(AppConfig datafileAppConfig) {
  52         this(datafileAppConfig, new JsonMessageParser(),
  53             new ConsumerReactiveHttpClientFactory(new DMaaPReactiveWebClientFactory()));
  54     }
  55
  56     protected DMaaPMessageConsumer(AppConfig datafileAppConfig, JsonMessageParser jsonMessageParser,
  57         ConsumerReactiveHttpClientFactory httpClientFactory) {
  58         this.datafileAppConfig = datafileAppConfig;
  59         this.jsonMessageParser = jsonMessageParser;
  60         this.httpClientFactory = httpClientFactory;
  61     }
  62
  63     /**
  64      * Gets the response from the MessageRouter and turns it into a stream of fileReady messages.
  65      *
  66      * @return a stream of fileReady messages.
  67      */
  68     public Flux<FileReadyMessage> getMessageRouterResponse() {
  69         logger.trace("getMessageRouterResponse called");
  70         try {
  71             DMaaPConsumerReactiveHttpClient dMaaPConsumerReactiveHttpClient = createHttpClient();
  72             return consume((dMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse(Optional.empty())));
  73         } catch (DatafileTaskException e) {
  74             logger.warn("Unable to get response from message router", e);
  75             return Flux.empty();
  76         }
  77     }
  78
  79     private Flux<FileReadyMessage> consume(Mono<JsonElement> message) {
  80         logger.trace("consume called with arg {}", message);
  81         return jsonMessageParser.getMessagesFromJson(message);
  82     }
  83
  84     public DMaaPConsumerReactiveHttpClient createHttpClient() throws DatafileTaskException {
  85
  86         return httpClientFactory.create(datafileAppConfig.getDmaapConsumerConfiguration().toDmaap());
  87     }
  88
  89 }