d861021540793e1a09bf05545b5aa5c15d997785
[dcaegen2/services.git] /
1 /*
2  * ============LICENSE_START=======================================================
3  * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
4  * ================================================================================
5  * Copyright (C) 2019 NOKIA Intellectual Property. 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.bbs.event.processor.tasks;
22
23 import javax.net.ssl.SSLException;
24
25 import org.onap.bbs.event.processor.config.ApplicationConfiguration;
26 import org.onap.bbs.event.processor.exceptions.EmptyDmaapResponseException;
27 import org.onap.bbs.event.processor.model.ReRegistrationConsumerDmaapModel;
28 import org.onap.bbs.event.processor.utilities.ReRegistrationDmaapConsumerJsonParser;
29 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer.ConsumerReactiveHttpClientFactory;
30 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer.DMaaPConsumerReactiveHttpClient;
31 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer.DMaaPReactiveWebClientFactory;
32 import org.slf4j.Logger;
33 import org.slf4j.LoggerFactory;
34 import org.springframework.beans.factory.annotation.Autowired;
35 import org.springframework.stereotype.Component;
36
37 import reactor.core.publisher.Flux;
38 import reactor.core.publisher.Mono;
39
40 @Component
41 public class DmaapReRegistrationConsumerTaskImpl implements DmaapReRegistrationConsumerTask {
42
43     private static final Logger LOGGER = LoggerFactory.getLogger(DmaapReRegistrationConsumerTaskImpl.class);
44     private final ApplicationConfiguration configuration;
45     private final ReRegistrationDmaapConsumerJsonParser reRegistrationDmaapConsumerJsonParser;
46     private final ConsumerReactiveHttpClientFactory httpClientFactory;
47
48     @Autowired
49     public DmaapReRegistrationConsumerTaskImpl(ApplicationConfiguration configuration) {
50         this(configuration, new ReRegistrationDmaapConsumerJsonParser(),
51                 new ConsumerReactiveHttpClientFactory(new DMaaPReactiveWebClientFactory()));
52     }
53
54     DmaapReRegistrationConsumerTaskImpl(ApplicationConfiguration configuration,
55                                                 ReRegistrationDmaapConsumerJsonParser reRegDmaapConsumerJsonParser,
56                                                 ConsumerReactiveHttpClientFactory httpClientFactory) {
57         this.configuration = configuration;
58         this.reRegistrationDmaapConsumerJsonParser = reRegDmaapConsumerJsonParser;
59         this.httpClientFactory = httpClientFactory;
60     }
61
62     @Override
63     public Flux<ReRegistrationConsumerDmaapModel> execute(String taskName) throws SSLException {
64         LOGGER.debug("Executing task for Re-Registration with name \"{}\"", taskName);
65         DMaaPConsumerReactiveHttpClient dmaaPConsumerReactiveHttpClient = resolveClient();
66         Mono<String> response = dmaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse();
67         return reRegistrationDmaapConsumerJsonParser.extractModelFromDmaap(response)
68                 .switchIfEmpty(Flux.error(
69                         new EmptyDmaapResponseException("Re-Registration: Got an empty response from DMaaP")))
70                 .doOnError(e -> {
71                     if (!(e instanceof EmptyDmaapResponseException)) {
72                         LOGGER.error("DMaaP Consumption Exception: {}", e.getMessage());
73                     }
74                 });
75     }
76
77     @Override
78     public DMaaPConsumerReactiveHttpClient resolveClient() throws SSLException {
79         return httpClientFactory.create(configuration.getDmaapReRegistrationConsumerConfiguration());
80     }
81 }