Kafka Application - Updated the Application to a Distributed Application
1. Distributed Work Queue to publish events.
2. Only one instance can publish to queue and only one instance can consume from the queue.
3. KafkaPublisherAdminService which does start,stop and restart a kafka producer.
3. Other cosmectic changes (some files have been renamed)
Change-Id: I15b2015ed303eae1c9fb9f4820fb14476726161e
diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/DeviceEventConverter.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/DeviceEventConverter.java
index 3d2eea1..b5bd01b 100644
--- a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/DeviceEventConverter.java
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/DeviceEventConverter.java
@@ -35,7 +35,7 @@
private final Logger log = LoggerFactory.getLogger(getClass());
@Override
- public GeneratedMessageV3 convertToProtoMessage(Event<?, ?> event) {
+ public byte[] convertToProtoMessage(Event<?, ?> event) {
DeviceEvent deviceEvent = (DeviceEvent) event;
@@ -45,7 +45,7 @@
return null;
}
- return buildDeviceProtoMessage(deviceEvent);
+ return ((GeneratedMessageV3) buildDeviceProtoMessage(deviceEvent)).toByteArray();
}
/**
diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/EventConverter.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/EventConverter.java
index 42736b4..83c62d6 100644
--- a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/EventConverter.java
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/EventConverter.java
@@ -16,7 +16,6 @@
import org.onosproject.event.Event;
-import com.google.protobuf.GeneratedMessageV3;
/**
*
@@ -30,8 +29,7 @@
* to Kafka.
*
* @param event ONOS Event object
- * @return converted data in protobuf format.
+ * @return converted data in protobuf format as a byte array.
*/
- // FIXME reconsider return type, something similar to "OnosEvent"?
- GeneratedMessageV3 convertToProtoMessage(Event<?, ?> event);
+ byte[] convertToProtoMessage(Event<?, ?> event);
}
diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/LinkEventConverter.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/LinkEventConverter.java
index c878a26..c2cf60d 100644
--- a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/LinkEventConverter.java
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/LinkEventConverter.java
@@ -35,7 +35,7 @@
private final Logger log = LoggerFactory.getLogger(getClass());
@Override
- public GeneratedMessageV3 convertToProtoMessage(Event<?, ?> event) {
+ public byte[] convertToProtoMessage(Event<?, ?> event) {
LinkEvent linkEvent = (LinkEvent) event;
@@ -45,7 +45,7 @@
return null;
}
- return buildDeviceProtoMessage(linkEvent);
+ return ((GeneratedMessageV3) buildDeviceProtoMessage(linkEvent)).toByteArray();
}
private boolean linkEventTypeSupported(LinkEvent event) {