2 * ========================LICENSE_START=================================
5 * Copyright (C) 2020 Nordix Foundation
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.oransc.policyagent.dmaap;
23 import com.google.common.collect.Iterables;
25 import java.io.IOException;
26 import java.time.Duration;
27 import java.util.Properties;
29 import org.onap.dmaap.mr.client.MRBatchingPublisher;
30 import org.onap.dmaap.mr.client.MRClientFactory;
31 import org.onap.dmaap.mr.client.MRConsumer;
32 import org.onap.dmaap.mr.client.response.MRConsumerResponse;
33 import org.oransc.policyagent.clients.AsyncRestClient;
34 import org.oransc.policyagent.configuration.ApplicationConfig;
35 import org.oransc.policyagent.exceptions.ServiceException;
36 import org.oransc.policyagent.tasks.RefreshConfigTask;
37 import org.slf4j.Logger;
38 import org.slf4j.LoggerFactory;
39 import org.springframework.beans.factory.annotation.Autowired;
40 import org.springframework.beans.factory.annotation.Value;
41 import org.springframework.stereotype.Component;
44 * The class fetches incoming requests from DMAAP. It uses the timeout parameter
45 * that lets the MessageRouter keep the connection with the Kafka open until
46 * requests are sent in.
49 * this service will regularly check the configuration and start polling DMaaP
50 * if the configuration is added. If the DMaaP configuration is removed, then
51 * the service will stop polling and resume checking for configuration.
54 * Each received request is processed by {@link DmaapMessageHandler}.
57 public class DmaapMessageConsumer {
59 protected static final Duration TIME_BETWEEN_DMAAP_RETRIES = Duration.ofSeconds(10);
61 private static final Logger logger = LoggerFactory.getLogger(DmaapMessageConsumer.class);
63 private final ApplicationConfig applicationConfig;
65 private DmaapMessageHandler dmaapMessageHandler = null;
66 private MRConsumer messageRouterConsumer = null;
68 @Value("${server.port}")
69 private int localServerPort;
72 public DmaapMessageConsumer(ApplicationConfig applicationConfig) {
73 this.applicationConfig = applicationConfig;
77 * Starts the consumer. If there is a DMaaP configuration, it will start polling for messages. Otherwise it will
78 * check regularly for the configuration.
80 * @return the running thread, for test purposes.
82 public Thread start() {
83 Thread thread = new Thread(this::checkConfigLoop);
88 private void checkConfigLoop() {
89 while (!isStopped()) {
90 if (isDmaapConfigured()) {
91 messageHandlingLoop();
93 sleep(RefreshConfigTask.CONFIG_REFRESH_INTERVAL);
98 private void messageHandlingLoop() {
99 while (!isStopped() && isDmaapConfigured()) {
101 Iterable<String> dmaapMsgs = fetchAllMessages();
102 if (dmaapMsgs != null && Iterables.size(dmaapMsgs) > 0) {
103 logger.debug("Fetched all the messages from DMAAP and will start to process the messages");
104 for (String msg : dmaapMsgs) {
108 } catch (Exception e) {
109 logger.warn("Cannot fetch because of {}", e.getMessage());
110 sleep(TIME_BETWEEN_DMAAP_RETRIES);
115 protected boolean isStopped() {
119 protected boolean isDmaapConfigured() {
120 Properties consumerCfg = applicationConfig.getDmaapConsumerConfig();
121 Properties producerCfg = applicationConfig.getDmaapPublisherConfig();
122 return (consumerCfg != null && consumerCfg.size() > 0 && producerCfg != null && producerCfg.size() > 0);
125 protected Iterable<String> fetchAllMessages() throws ServiceException, IOException {
126 Properties dmaapConsumerProperties = this.applicationConfig.getDmaapConsumerConfig();
127 MRConsumer consumer = getMessageRouterConsumer(dmaapConsumerProperties);
128 MRConsumerResponse response = consumer.fetchWithReturnConsumerResponse();
129 if (response == null || !"200".equals(response.getResponseCode())) {
130 String errorMessage = "DMaaP NULL response received";
131 if (response != null) {
132 errorMessage = "Error respons " + response.getResponseCode() + " " + response.getResponseMessage()
135 throw new ServiceException(errorMessage);
137 logger.debug("DMaaP consumer received {} : {}", response.getResponseCode(), response.getResponseMessage());
138 return response.getActualMessages();
142 private void processMsg(String msg) throws IOException {
143 logger.debug("Message Reveived from DMAAP : {}", msg);
144 getDmaapMessageHandler().handleDmaapMsg(msg);
147 protected DmaapMessageHandler getDmaapMessageHandler() throws IOException {
148 if (this.dmaapMessageHandler == null) {
149 String agentBaseUrl = "https://localhost:" + this.localServerPort;
150 AsyncRestClient agentClient = new AsyncRestClient(agentBaseUrl);
151 Properties dmaapPublisherProperties = applicationConfig.getDmaapPublisherConfig();
152 MRBatchingPublisher producer = MRClientFactory.createBatchingPublisher(dmaapPublisherProperties);
153 this.dmaapMessageHandler = new DmaapMessageHandler(producer, agentClient);
155 return this.dmaapMessageHandler;
158 protected void sleep(Duration duration) {
160 Thread.sleep(duration.toMillis());
161 } catch (Exception e) {
162 logger.error("Failed to put the thread to sleep", e);
166 protected MRConsumer getMessageRouterConsumer(Properties dmaapConsumerProperties) throws IOException {
167 if (this.messageRouterConsumer == null) {
168 this.messageRouterConsumer = MRClientFactory.createConsumer(dmaapConsumerProperties);
170 return this.messageRouterConsumer;