blob: 1f5865f27d5c32bfc6a8f5dc115388c9b53fc1c0 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
|
package org.openecomp.sdc.be.components.distribution.engine;
import com.att.nsa.mr.client.MRConsumer;
import org.openecomp.sdc.be.config.ConfigurationManager;
import org.openecomp.sdc.be.config.DmaapConsumerConfiguration;
import org.openecomp.sdc.common.log.wrappers.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.lang.Thread.UncaughtExceptionHandler;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
/**
* Allows consuming DMAAP topic according to received consumer parameters
* Allows processing received messages.
*/
@Service
public class DmaapConsumer {
private final ExecutorFactory executorFactory;
private final DmaapClientFactory dmaapClientFactory;
private static final Logger logger = Logger.getLogger(DmaapClientFactory.class.getName());
@Autowired
private DmaapHealth dmaapHealth;
/**
* Allows to create an object of type DmaapConsumer
* @param executorFactory
* @param dmaapClientFactory
*/
@Autowired
public DmaapConsumer(ExecutorFactory executorFactory, DmaapClientFactory dmaapClientFactory) {
this.executorFactory = executorFactory;
this.dmaapClientFactory = dmaapClientFactory;
}
/**
* Allows consuming DMAAP topic according to received consumer parameters
* @param notificationReceived
* @param exceptionHandler
* @throws Exception
*/
public void consumeDmaapTopic(Consumer<String> notificationReceived, UncaughtExceptionHandler exceptionHandler) throws Exception {
DmaapConsumerConfiguration dmaapConsumerParams = ConfigurationManager.getConfigurationManager().getConfiguration().getDmaapConsumerConfiguration();
String topic = dmaapConsumerParams.getTopic();
logger.info("Starting to consume topic {} for DMAAP consumer with the next parameters {}. ", topic, dmaapConsumerParams);
MRConsumer consumer = dmaapClientFactory.create(dmaapConsumerParams);
ScheduledExecutorService pollExecutor = executorFactory.createScheduled(topic + "Client");
ExecutorService notificationExecutor = executorFactory.create(topic + "Consumer", exceptionHandler);
pollExecutor.scheduleWithFixedDelay(() -> {
logger.info("Trying to fetch messages from topic: {}", topic);
boolean isTopicAvailable = false;
try {
Iterable<String> messages = consumer.fetch();
isTopicAvailable = true ;
if (messages != null) {
for (String msg : messages) {
logger.info("The DMAAP message {} received. The topic is {}.", msg, topic);
notificationExecutor.execute(() -> notificationReceived.accept(msg));
}
}
//successfully fetched
}
catch (Exception e) {
logger.error("The exception occured upon fetching DMAAP message", e);
}
dmaapHealth.report( isTopicAvailable );
}, 0L, dmaapConsumerParams.getPollingInterval(), TimeUnit.SECONDS);
}
}
|