commit d1ce10ad9e0ccac9a089dd65ddaa08f1718b506e
author Jian Li <pyguni@gmail.com> Tue Jun 12 13:47:23 2018 +0900
committer Jian Li <pyguni@gmail.com> Thu Jun 21 01:20:33 2018 +0000
tree d7e1739b02a45bdca3afbfca337bd03d600c54f1
parent 4a3fffad43a3a8c8af70d8991f7aba8fdf75dd52

[ONOS-7683] Initial implementation of openstack telemetry service

Change-Id: I621bed9cff108af194998b7a1e8269a9a6ffd32c

apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/KafkaTelemetryManager.java

diff --git a/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/KafkaTelemetryManager.java b/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/KafkaTelemetryManager.java
index e935111..8d6f62f 100644
--- a/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/KafkaTelemetryManager.java
+++ b/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/KafkaTelemetryManager.java
@@ -18,12 +18,15 @@
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
 import org.apache.felix.scr.annotations.Service;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.onosproject.openstacktelemetry.api.KafkaTelemetryAdminService;
+import org.onosproject.openstacktelemetry.api.OpenstackTelemetryService;
 import org.onosproject.openstacktelemetry.api.config.KafkaTelemetryConfig;
 import org.onosproject.openstacktelemetry.api.config.TelemetryConfig;
 import org.slf4j.Logger;
@@ -50,16 +53,25 @@
     private static final String KEY_SERIALIZER = "key.serializer";
     private static final String VALUE_SERIALIZER = "value.serializer";
 
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected OpenstackTelemetryService openstackTelemetryService;
+
     private Producer<String, byte[]> producer = null;
 
     @Activate
     protected void activate() {
+
+        openstackTelemetryService.addTelemetryService(this);
+
         log.info("Started");
     }
 
     @Deactivate
     protected void deactivate() {
         stop();
+
+        openstackTelemetryService.removeTelemetryService(this);
+
         log.info("Stopped");
     }
 
@@ -110,6 +122,17 @@
 
     @Override
     public Future<RecordMetadata> publish(ProducerRecord<String, byte[]> record) {
+
+        if (producer == null) {
+            log.warn("Kafka telemetry service has not been enabled!");
+            return null;
+        }
+
         return producer.send(record);
     }
+
+    @Override
+    public boolean isRunning() {
+        return producer != null;
+    }
 }