Kafka Integration Application (Fix Javadoc errors + review comments)
1. Refactored the application into two java projects api and app as per convention
2. Deleted the onos-app-gpb project. The proto files are consolidated in the
incubator-protobuf project as per suggestions.
3. Some code to translate ONOS Event pojo messages to GPB format.
4. Implementation of Subscribe and Unsubscribe APIs.
5. Minor changes due to review comments from 9212 and 9053
6. Refactored the proto fileso that its a 1:1 mapping between the core type to proto message.
Change-Id: I2bcc0de96150f838ccfe9e49293fe61d94062628
diff --git a/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/LinkEventConverter.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/LinkEventConverter.java
new file mode 100644
index 0000000..2d65537
--- /dev/null
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/LinkEventConverter.java
@@ -0,0 +1,104 @@
+/**
+ * Copyright 2016 Open Networking Laboratory
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.kafkaintegration.converter;
+
+import org.onosproject.event.Event;
+import org.onosproject.grpc.net.Link.ConnectPoint;
+import org.onosproject.grpc.net.Link.LinkCore;
+import org.onosproject.grpc.net.Link.LinkState;
+import org.onosproject.grpc.net.Link.LinkType;
+import org.onosproject.grpc.net.LinkEvent.LinkEventType;
+import org.onosproject.grpc.net.LinkEvent.LinkNotification;
+import org.onosproject.net.link.LinkEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.protobuf.GeneratedMessage;
+
+/**
+ * Converts for ONOS Link event message to GPB format.
+ *
+ */
+class LinkEventConverter implements EventConverter {
+
+ private final Logger log = LoggerFactory.getLogger(getClass());
+
+ @Override
+ public GeneratedMessage convertToProtoMessage(Event<?, ?> event) {
+
+ LinkEvent linkEvent = (LinkEvent) event;
+
+ if (!linkEventSubtypeSupported(linkEvent)) {
+ log.error("Unsupported Onos Event {}. There is no matching"
+ + "proto Event type", linkEvent.type().toString());
+ return null;
+ }
+
+ return buildDeviceProtoMessage(linkEvent);
+ }
+
+ private boolean linkEventSubtypeSupported(LinkEvent event) {
+ LinkType[] kafkaLinkEvents = LinkType.values();
+ for (LinkType linkEventType : kafkaLinkEvents) {
+ if (linkEventType.name().equals(event.type().name())) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ private LinkNotification buildDeviceProtoMessage(LinkEvent linkEvent) {
+ LinkNotification notification = LinkNotification.newBuilder()
+ .setLinkEventType(getProtoType(linkEvent))
+ .setLink(LinkCore.newBuilder()
+ .setState(LinkState
+ .valueOf(linkEvent.subject().state().name()))
+ .setType(LinkType
+ .valueOf(linkEvent.subject().type().name()))
+ .setDst(ConnectPoint.newBuilder()
+ .setDeviceId(linkEvent.subject().dst()
+ .deviceId().toString())
+ .setPortNumber(linkEvent.subject().dst().port()
+ .toString()))
+ .setSrc(ConnectPoint.newBuilder()
+ .setDeviceId(linkEvent.subject().src()
+ .deviceId().toString())
+ .setPortNumber(linkEvent.subject().src().port()
+ .toString())))
+ .build();
+
+ return notification;
+ }
+
+ /**
+ * Returns the specific Kafka Device Event Type for the corresponding ONOS
+ * Device Event Type.
+ *
+ * @param event ONOS Device Event
+ * @return Kafka Device Event Type
+ */
+ private LinkEventType getProtoType(LinkEvent event) {
+ LinkEventType generatedEventType = null;
+ LinkEventType[] kafkaEvents = LinkEventType.values();
+ for (LinkEventType linkEventType : kafkaEvents) {
+ if (linkEventType.name().equals(event.type().name())) {
+ generatedEventType = linkEventType;
+ }
+ }
+
+ return generatedEventType;
+ }
+}