commit ee83f26c35d2874a58f52f183d90b2f3a27367b6
author Aaron Dunlap <aaron.m.dunlap@windstream.com> Mon Jan 14 12:32:38 2019 -0600
committer Aaron Dunlap <aaron.m.dunlap@windstream.com> Mon Jan 14 14:23:11 2019 -0600
tree 1ab459a00f8c0dfcd38eb7e8f2ab24e7fab78533
parent 5d65d11a755244653d7245bf5c6b66536ab27e49

Added code to allow HOST events to be sent via Kafka

Change-Id: I66e9df2fcb6cce75216aa1b3b2969cd12aec092d

apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/kafka/EventListener.java

diff --git a/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/kafka/EventListener.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/kafka/EventListener.java
index 18baccb..91e2028 100644
--- a/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/kafka/EventListener.java
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/kafka/EventListener.java
@@ -34,6 +34,9 @@
 import org.onosproject.net.link.LinkEvent;
 import org.onosproject.net.link.LinkListener;
 import org.onosproject.net.link.LinkService;
+import org.onosproject.net.host.HostEvent;
+import org.onosproject.net.host.HostListener;
+import org.onosproject.net.host.HostService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -44,6 +47,7 @@
 import static org.onlab.util.Tools.groupedThreads;
 import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
 import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.HOST;
 
 
 /**
@@ -66,6 +70,9 @@
     protected LinkService linkService;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected HostService hostService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
     protected KafkaEventStorageService kafkaStoreService;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY)
@@ -76,6 +83,7 @@
 
     private final DeviceListener deviceListener = new InternalDeviceListener();
     private final LinkListener linkListener = new InternalLinkListener();
+    private final HostListener hostListener = new InternalHostListener();
 
     protected ExecutorService eventExecutor;
 
@@ -89,6 +97,7 @@
         eventExecutor = newSingleThreadScheduledExecutor(groupedThreads("onos/onosEvents", "events-%d", log));
         deviceService.addListener(deviceListener);
         linkService.addListener(linkListener);
+        hostService.addListener(hostListener);
 
         localNodeId = clusterService.getLocalNode().id();
 
@@ -101,6 +110,7 @@
     protected void deactivate() {
         deviceService.removeListener(deviceListener);
         linkService.removeListener(linkListener);
+        hostService.removeListener(hostListener);
 
         eventExecutor.shutdownNow();
         eventExecutor = null;
@@ -153,4 +163,27 @@
 
         }
     }
+
+    private class InternalHostListener implements HostListener {
+
+        @Override
+        public void event(HostEvent event) {
+
+            // do not allow to proceed without leadership
+            NodeId leaderNodeId = leadershipService.getLeader(PUBLISHER_TOPIC);
+            if (!Objects.equals(localNodeId, leaderNodeId)) {
+                log.debug("Not a Leader, cannot publish!");
+                return;
+            }
+
+            if (!eventSubscriptionService.getEventSubscribers(HOST).isEmpty()) {
+                OnosEvent onosEvent = eventConversionService.convertEvent(event);
+                eventExecutor.execute(() -> {
+                    kafkaStoreService.publishEvent(onosEvent);
+                });
+                log.debug("Pushed event {} to kafka storage", onosEvent);
+            }
+
+        }
+    }
 }
\ No newline at end of file