5779051c5133082e3c36ded2eeacb9d994018f9c
[dcaegen2/collectors/datafile.git] /
1 /*
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
7  *
8  * http://www.apache.org/licenses/LICENSE-2.0
9  *
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
13  * the License.
14  * ============LICENSE_END========================================================================
15  */
16
17 package org.onap.dcaegen2.collectors.datafile.tasks;
18
19 import java.util.List;
20
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;
32
33 import reactor.core.publisher.Mono;
34
35 /**
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>
38  */
39 @Component
40 public class DmaapPublisherTaskImpl extends DmaapPublisherTask {
41
42     private static final Logger logger = LoggerFactory.getLogger(DmaapPublisherTaskImpl.class);
43     private final Config datafileAppConfig;
44     private DmaapProducerReactiveHttpClient dmaapProducerReactiveHttpClient;
45
46     @Autowired
47     public DmaapPublisherTaskImpl(AppConfig datafileAppConfig) {
48         this.datafileAppConfig = datafileAppConfig;
49     }
50
51     @Override
52     public Mono<String> publish(Mono<List<ConsumerDmaapModel>> consumerDmaapModels) {
53         logger.info("Publishing on DMaaP DataRouter {}", consumerDmaapModels);
54         return dmaapProducerReactiveHttpClient.getDmaapProducerResponse(consumerDmaapModels);
55     }
56
57     @Override
58     public Mono<String> execute(Mono<List<ConsumerDmaapModel>> consumerDmaapModels)
59         throws DatafileTaskException {
60         if (consumerDmaapModels == null) {
61             throw new DmaapNotFoundException("Invoked null object to DMaaP task");
62         }
63         dmaapProducerReactiveHttpClient = resolveClient();
64         logger.trace("Method called with arg {}", consumerDmaapModels);
65         return publish(consumerDmaapModels);
66     }
67
68     @Override
69     protected DmaapPublisherConfiguration resolveConfiguration() {
70         return datafileAppConfig.getDmaapPublisherConfiguration();
71     }
72
73     @Override
74     DmaapProducerReactiveHttpClient resolveClient() {
75         return new DmaapProducerReactiveHttpClient(resolveConfiguration()).createDmaapWebClient(buildWebClient());
76     }
77
78 }