8c4d707226610fc331107b07b50062539d4a88e6
[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             throws DatafileTaskException {
54         logger.info("Publishing on DMaaP DataRouter {}", consumerDmaapModels);
55         return dmaapProducerReactiveHttpClient.getDmaapProducerResponse(consumerDmaapModels);
56     }
57
58     @Override
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");
63         }
64         dmaapProducerReactiveHttpClient = resolveClient();
65         logger.trace("Method called with arg {}", consumerDmaapModels);
66         return publish(consumerDmaapModels);
67     }
68
69     @Override
70     protected DmaapPublisherConfiguration resolveConfiguration() {
71         return datafileAppConfig.getDmaapPublisherConfiguration();
72     }
73
74     @Override
75     DmaapProducerReactiveHttpClient resolveClient() {
76         return dmaapProducerReactiveHttpClient == null
77                 ? new DmaapProducerReactiveHttpClient(resolveConfiguration()).createDmaapWebClient(buildWebClient())
78                 : dmaapProducerReactiveHttpClient;
79     }
80 }