2 * ============LICENSE_START======================================================================
3 * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
4 * ===============================================================================================
5 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
6 * in compliance with the License. You may obtain a copy of the License at
8 * http://www.apache.org/licenses/LICENSE-2.0
10 * Unless required by applicable law or agreed to in writing, software distributed under the License
11 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
12 * or implied. See the License for the specific language governing permissions and limitations under
14 * ============LICENSE_END========================================================================
17 package org.onap.dcaegen2.collectors.datafile.tasks;
19 import java.util.List;
21 import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration;
22 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
23 import org.onap.dcaegen2.collectors.datafile.configuration.Config;
24 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
25 import org.onap.dcaegen2.collectors.datafile.exceptions.DmaapNotFoundException;
26 import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
27 import org.onap.dcaegen2.collectors.datafile.service.producer.DmaapProducerReactiveHttpClient;
28 import org.slf4j.Logger;
29 import org.slf4j.LoggerFactory;
30 import org.springframework.beans.factory.annotation.Autowired;
31 import org.springframework.stereotype.Component;
33 import reactor.core.publisher.Mono;
36 * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/13/18
37 * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
40 public class DmaapPublisherTaskImpl extends DmaapPublisherTask {
42 private static final Logger logger = LoggerFactory.getLogger(DmaapPublisherTaskImpl.class);
43 private final Config datafileAppConfig;
44 private DmaapProducerReactiveHttpClient dmaapProducerReactiveHttpClient;
47 public DmaapPublisherTaskImpl(AppConfig datafileAppConfig) {
48 this.datafileAppConfig = datafileAppConfig;
52 public Mono<String> publish(Mono<List<ConsumerDmaapModel>> consumerDmaapModels)
53 throws DatafileTaskException {
54 logger.info("Publishing on DMaaP DataRouter {}", consumerDmaapModels);
55 return dmaapProducerReactiveHttpClient.getDmaapProducerResponse(consumerDmaapModels);
59 public Mono<String> execute(Mono<List<ConsumerDmaapModel>> consumerDmaapModels)
60 throws DatafileTaskException {
61 if (consumerDmaapModels == null) {
62 throw new DmaapNotFoundException("Invoked null object to DMaaP task");
64 dmaapProducerReactiveHttpClient = resolveClient();
65 logger.trace("Method called with arg {}", consumerDmaapModels);
66 return publish(consumerDmaapModels);
70 protected DmaapPublisherConfiguration resolveConfiguration() {
71 return datafileAppConfig.getDmaapPublisherConfiguration();
75 DmaapProducerReactiveHttpClient resolveClient() {
76 return dmaapProducerReactiveHttpClient == null
77 ? new DmaapProducerReactiveHttpClient(resolveConfiguration()).createDmaapWebClient(buildWebClient())
78 : dmaapProducerReactiveHttpClient;