X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=policy-agent%2Fsrc%2Fmain%2Fjava%2Forg%2Foransc%2Fpolicyagent%2Fdmaap%2FDmaapMessageConsumer.java;h=6312e375aaccc7bd55a4c1f9c6f0cccb1fd57a4f;hb=0b10c7fa768f05ae5146ce2f3a69998bb8f97a9f;hp=da209a8609888f4c1d07d1aed58e686c1dd24acb;hpb=d917ed06c852dbdd282d7aeec6e3f77e83e8c791;p=nonrtric.git diff --git a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumer.java b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumer.java index da209a86..6312e375 100644 --- a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumer.java +++ b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumer.java @@ -33,6 +33,7 @@ import org.onap.dmaap.mr.client.response.MRConsumerResponse; import org.oransc.policyagent.clients.AsyncRestClient; import org.oransc.policyagent.configuration.ApplicationConfig; import org.oransc.policyagent.exceptions.ServiceException; +import org.oransc.policyagent.tasks.RefreshConfigTask; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -40,15 +41,23 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; /** - * The class fetches incoming requests from DMAAP on regular intervals. Each - * received request is proceesed by DmaapMessageHandler. + * The class fetches incoming requests from DMAAP. It uses the timeout parameter that lets the MessageRouter keep the + * connection with the Kafka open until requests are sent in. + * + *
+ * If there is no DMaaP configuration in the application configuration, then this service will regularly check the + * configuration and start polling DMaaP if the configuration is added. If the DMaaP configuration is removed, then the + * service will stop polling and resume checking for configuration. + * + *
+ * Each received request is processed by {@link DmaapMessageHandler}.
*/
@Component
-public class DmaapMessageConsumer implements Runnable {
+public class DmaapMessageConsumer {
- private static final Logger logger = LoggerFactory.getLogger(DmaapMessageConsumer.class);
+ protected static final Duration TIME_BETWEEN_DMAAP_RETRIES = Duration.ofSeconds(10);
- private static final Duration TIME_BETWEEN_DMAAP_POLLS = Duration.ofSeconds(10);
+ private static final Logger logger = LoggerFactory.getLogger(DmaapMessageConsumer.class);
private final ApplicationConfig applicationConfig;
@@ -58,20 +67,32 @@ public class DmaapMessageConsumer implements Runnable {
@Autowired
public DmaapMessageConsumer(ApplicationConfig applicationConfig) {
this.applicationConfig = applicationConfig;
+ }
- Thread thread = new Thread(this);
+ /**
+ * Starts the consumer. If there is a DMaaP configuration, it will start polling for messages. Otherwise it will
+ * check regularly for the configuration.
+ *
+ * @return the running thread, for test purposes.
+ */
+ public Thread start() {
+ Thread thread = new Thread(this::checkConfigLoop);
thread.start();
+ return thread;
}
- private boolean isDmaapConfigured() {
- Properties consumerCfg = applicationConfig.getDmaapConsumerConfig();
- Properties producerCfg = applicationConfig.getDmaapPublisherConfig();
- return (consumerCfg != null && consumerCfg.size() > 0 && producerCfg != null && producerCfg.size() > 0);
+ private void checkConfigLoop() {
+ while (!isStopped()) {
+ if (isDmaapConfigured()) {
+ messageHandlingLoop();
+ } else {
+ sleep(RefreshConfigTask.CONFIG_REFRESH_INTERVAL);
+ }
+ }
}
- @Override
- public void run() {
- while (sleep(TIME_BETWEEN_DMAAP_POLLS) && isDmaapConfigured()) {
+ private void messageHandlingLoop() {
+ while (!isStopped() && isDmaapConfigured()) {
try {
Iterable