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.FileNotFoundException;
26 import java.io.IOException;
27 import java.time.Duration;
28 import java.util.Properties;
30 import org.onap.dmaap.mr.client.MRBatchingPublisher;
31 import org.onap.dmaap.mr.client.MRClientFactory;
32 import org.onap.dmaap.mr.client.MRConsumer;
33 import org.onap.dmaap.mr.client.response.MRConsumerResponse;
34 import org.oransc.policyagent.clients.AsyncRestClient;
35 import org.oransc.policyagent.configuration.ApplicationConfig;
36 import org.oransc.policyagent.exceptions.ServiceException;
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 public class DmaapMessageConsumer implements Runnable {
46 private static final Logger logger = LoggerFactory.getLogger(DmaapMessageConsumer.class);
48 private final static Duration TIME_BETWEEN_DMAAP_POLLS = Duration.ofSeconds(10);
50 private final ApplicationConfig applicationConfig;
52 @Value("${server.port}")
53 private int localServerPort;
56 public DmaapMessageConsumer(ApplicationConfig applicationConfig) {
57 this.applicationConfig = applicationConfig;
59 Thread thread = new Thread(this);
63 DmaapMessageConsumer(ApplicationConfig applicationConfig, boolean start) {
64 this.applicationConfig = applicationConfig;
67 private boolean isDmaapConfigured() {
68 Properties consumerCfg = applicationConfig.getDmaapConsumerConfig();
69 Properties producerCfg = applicationConfig.getDmaapPublisherConfig();
70 return (consumerCfg != null && consumerCfg.size() > 0 && producerCfg != null && producerCfg.size() > 0);
75 while (sleep(TIME_BETWEEN_DMAAP_POLLS) && isDmaapConfigured()) {
77 Iterable<String> dmaapMsgs = fetchAllMessages();
78 if (dmaapMsgs != null && Iterables.size(dmaapMsgs) > 0) {
79 logger.debug("Fetched all the messages from DMAAP and will start to process the messages");
80 for (String msg : dmaapMsgs) {
84 } catch (Exception e) {
85 logger.warn("{}: cannot fetch because of {}", this, e.getMessage());
86 sleep(TIME_BETWEEN_DMAAP_POLLS);
91 private Iterable<String> fetchAllMessages() throws ServiceException, IOException {
92 Properties dmaapConsumerProperties = this.applicationConfig.getDmaapConsumerConfig();
93 MRConsumer consumer = getMessageRouterConsumer(dmaapConsumerProperties);
94 MRConsumerResponse response = consumer.fetchWithReturnConsumerResponse();
95 if (response == null || !"200".equals(response.getResponseCode())) {
96 String errorMessage = "DMaaP NULL response received";
97 if (response != null) {
98 errorMessage = "Error respons " + response.getResponseCode() + " " + response.getResponseMessage()
101 throw new ServiceException(errorMessage);
103 logger.debug("DMaaP consumer received {} : {}", response.getResponseCode(), response.getResponseMessage());
104 return response.getActualMessages();
108 private void processMsg(String msg) throws IOException {
109 logger.debug("Message Reveived from DMAAP : {}", msg);
110 getDmaapMessageHandler().handleDmaapMsg(msg);
113 private DmaapMessageHandler getDmaapMessageHandler() throws IOException {
114 String agentBaseUrl = "http://localhost:" + this.localServerPort;
115 AsyncRestClient agentClient = createRestClient(agentBaseUrl);
116 Properties dmaapPublisherProperties = applicationConfig.getDmaapPublisherConfig();
117 MRBatchingPublisher producer = getMessageRouterPublisher(dmaapPublisherProperties);
119 return createDmaapMessageHandler(agentClient, producer);
122 boolean sleep(Duration duration) {
124 Thread.sleep(duration.toMillis());
126 } catch (Exception e) {
127 logger.error("Failed to put the thread to sleep", e);
132 MRConsumer getMessageRouterConsumer(Properties dmaapConsumerProperties) throws FileNotFoundException, IOException {
133 return MRClientFactory.createConsumer(dmaapConsumerProperties);
136 DmaapMessageHandler createDmaapMessageHandler(AsyncRestClient agentClient, MRBatchingPublisher producer) {
137 return new DmaapMessageHandler(producer, agentClient);
140 AsyncRestClient createRestClient(String agentBaseUrl) {
141 return new AsyncRestClient(agentBaseUrl);
144 MRBatchingPublisher getMessageRouterPublisher(Properties dmaapPublisherProperties)
145 throws FileNotFoundException, IOException {
146 return MRClientFactory.createBatchingPublisher(dmaapPublisherProperties);