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
11 * http://www.apache.org/licenses/LICENSE-2.0
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=========================================================
21 package org.onap.bbs.event.processor.tasks;
23 import javax.net.ssl.SSLException;
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;
37 import reactor.core.publisher.Flux;
38 import reactor.core.publisher.Mono;
41 public class DmaapReRegistrationConsumerTaskImpl implements DmaapReRegistrationConsumerTask {
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;
49 public DmaapReRegistrationConsumerTaskImpl(ApplicationConfiguration configuration) {
50 this(configuration, new ReRegistrationDmaapConsumerJsonParser(),
51 new ConsumerReactiveHttpClientFactory(new DMaaPReactiveWebClientFactory()));
54 DmaapReRegistrationConsumerTaskImpl(ApplicationConfiguration configuration,
55 ReRegistrationDmaapConsumerJsonParser reRegDmaapConsumerJsonParser,
56 ConsumerReactiveHttpClientFactory httpClientFactory) {
57 this.configuration = configuration;
58 this.reRegistrationDmaapConsumerJsonParser = reRegDmaapConsumerJsonParser;
59 this.httpClientFactory = httpClientFactory;
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")))
71 if (!(e instanceof EmptyDmaapResponseException)) {
72 LOGGER.error("DMaaP Consumption Exception: {}", e.getMessage());
78 public DMaaPConsumerReactiveHttpClient resolveClient() throws SSLException {
79 return httpClientFactory.create(configuration.getDmaapReRegistrationConsumerConfiguration());