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

+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2016-present 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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+              xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+              xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <artifactId>onos-kafka</artifactId>
+        <groupId>org.onosproject</groupId>
+        <version>1.7.0-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>onos-app-kafka-api</artifactId>
+    <packaging>bundle</packaging>
+
+    <url>http://onosproject.org</url>
+
+    <description>Kafka Integration Application</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+            <version>3.0.0-beta-2</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+       <plugins>
+          <plugin>
+              <groupId>org.apache.felix</groupId>
+              <artifactId>maven-bundle-plugin</artifactId>
+              <extensions>true</extensions>
+              <configuration>
+                  <instructions>
+                  </instructions>
+              </configuration>
+          </plugin>
+       </plugins>
+    </build>
+</project>
+

@@ -18,11 +18,13 @@ import org.onosproject.kafkaintegration.api.dto.EventSubscriber;
 import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId;
 import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId;
 import org.onosproject.kafkaintegration.errors.InvalidApplicationException;
 import org.onosproject.kafkaintegration.errors.InvalidApplicationException;
 import org.onosproject.kafkaintegration.errors.InvalidGroupIdException;
 import org.onosproject.kafkaintegration.errors.InvalidGroupIdException;
-import org.onosproject.kafkaintegration.errors.UnsupportedEventException;
+
+import com.google.common.annotations.Beta;
 
 
 /**
 /**
  * APIs for subscribing to Onos Event Messages.
  * APIs for subscribing to Onos Event Messages.
  */
  */
+@Beta
 public interface EventExporterService {
 public interface EventExporterService {
 
 
     /**
     /**
@@ -44,13 +46,11 @@ public interface EventExporterService {
      * Allows registered listener to subscribe for a specific event type.
      * Allows registered listener to subscribe for a specific event type.
      *
      *
      * @param subscriber Subscription data containing the event type
      * @param subscriber Subscription data containing the event type
-     * @throws UnsupportedEventException
      * @throws InvalidGroupIdException
      * @throws InvalidGroupIdException
      * @throws InvalidApplicationException
      * @throws InvalidApplicationException
      */
      */
     void subscribe(EventSubscriber subscriber)
     void subscribe(EventSubscriber subscriber)
-            throws UnsupportedEventException, InvalidGroupIdException,
-            InvalidApplicationException;
+            throws InvalidGroupIdException, InvalidApplicationException;
 
 
     /**
     /**
      * Allows the registered listener to unsubscribe for a specific event.
      * Allows the registered listener to unsubscribe for a specific event.

+/**
+ * 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.api;
+
+import org.onosproject.event.EventListener;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent;
+
+import com.google.common.annotations.Beta;
+
+/**
+ * API for listeners to listen for Events generated by the ONOS event listener.
+ * At present the only listener will be a Kafka Manager or Module whose sole
+ * purpose is to publish the received data to a Kafka message bus.
+ *
+ */
+@Beta
+public interface ExportableEventListener extends EventListener<OnosEvent> {
+
+}

@@ -14,5 +14,6 @@
 
 
 /**
 /**
  * Application specific Exception classes.
  * Application specific Exception classes.
+ *
  */
  */
 package org.onosproject.kafkaintegration.errors;
 package org.onosproject.kafkaintegration.errors;

+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2016-present 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.
+  -->
+<app name="org.onosproject.kafkaintegration" origin="Calix" version="${project.version}"
+     category="Utility" url="http://onosproject.org" title="Kafka Integration Application"
+     featuresRepo="mvn:${project.groupId}/${project.artifactId}/${project.version}/xml/features"
+     features="${project.artifactId}" apps="org.onosproject.incubator.protobuf">
+    <description>${project.description}</description>
+    <artifact>mvn:${project.groupId}/${project.artifactId}/${project.version}</artifact>
+    <artifact>mvn:${project.groupId}/onos-app-kafka-api/${project.version}</artifact>
+</app>

+<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+<!--
+  ~ Copyright 2016-present 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.
+  -->
+<features xmlns="http://karaf.apache.org/xmlns/features/v1.2.0" name="${project.artifactId}-${project.version}">
+    <feature name="${project.artifactId}" version="${project.version}"
+             description="${project.description}">
+        <feature>onos-api</feature>
+        <bundle>mvn:${project.groupId}/onos-app-kafka-api/${project.version}</bundle>
+        <bundle>mvn:${project.groupId}/onos-app-kafka/${project.version}</bundle>
+    </feature>
+</features>

+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2016-present 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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>org.onosproject</groupId>
+        <artifactId>onos-kafka</artifactId>
+        <version>1.7.0-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>onos-app-kafka</artifactId>
+
+    <packaging>bundle</packaging>
+    <description>
+          Kafka Integration Application.
+          This will export ONOS Events to Northbound Kafka Server.
+    </description>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <onos.version>${project.version}</onos.version>
+        <onos.app.name>org.onosproject.kafkaintegration</onos.app.name>
+        <onos.app.title>Kafka Integration Application</onos.app.title>
+        <onos.app.origin>Calix, Inc.</onos.app.origin>
+        <web.context>/onos/kafka</web.context>
+        <api.version>1.0.0</api.version>
+        <api.package>org.onosproject.kafkaintegration.rest</api.package>
+        <api.title>Kafka Integration Application REST API</api.title>
+        <api.description>
+               APIs for subscribing to Events generated by ONOS
+        </api.description>
+        <onos.app.category>Utility</onos.app.category>
+        <onos.app.url>https://wiki.onosproject.org/display/ONOS/Kafka+Integration</onos.app.url>
+        <onos.app.readme>Export ONOS events to a Northbound Kafka server</onos.app.readme>
+        <onos.app.requires>org.onosproject.incubator.protobuf</onos.app.requires>
+    </properties>
+
+       <dependencies>
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-api</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-app-kafka-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-incubator-protobuf</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onlab-osgi</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-rest</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-api</artifactId>
+            <scope>test</scope>
+            <classifier>tests</classifier>
+        </dependency>
+
+        <dependency>
+            <groupId>javax.ws.rs</groupId>
+            <artifactId>javax.ws.rs-api</artifactId>
+            <version>2.0.1</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+            <version>3.0.0-beta-2</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.codehaus.jackson</groupId>
+            <artifactId>jackson-core-asl</artifactId>
+            <version>1.9.13</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.codehaus.jackson</groupId>
+            <artifactId>jackson-mapper-asl</artifactId>
+            <version>1.9.13</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.glassfish.jersey.containers</groupId>
+            <artifactId>jersey-container-servlet</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-core-serializers</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.felix</groupId>
+            <artifactId>org.apache.felix.scr.annotations</artifactId>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.felix</groupId>
+                <artifactId>maven-bundle-plugin</artifactId>
+                <extensions>true</extensions>
+                <configuration>
+                    <instructions>
+                    <Bundle-SymbolicName>
+                        ${project.groupId}.${project.artifactId}
+                    </Bundle-SymbolicName>
+                    <_wab>src/main/webapp/</_wab>
+                    <Include-Resource>
+                        WEB-INF/classes/apidoc/swagger.json=target/swagger.json,
+                                    {maven-resources}
+                    </Include-Resource>
+                    <Import-Package>
+                        org.slf4j,
+                        org.osgi.framework,
+                        javax.ws.rs,
+                        javax.ws.rs.core,
+                        org.glassfish.jersey.servlet,
+                        com.fasterxml.jackson.databind,
+                        com.fasterxml.jackson.databind.node,
+                        com.fasterxml.jackson.core,
+                        org.onlab.packet.*,
+                        org.onosproject.*,
+                        com.google.common.*
+                    </Import-Package>
+                    <Web-ContextPath>${web.context}</Web-ContextPath>
+                </instructions>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>1.8</source>
+                    <target>1.8</target>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.felix</groupId>
+                <artifactId>maven-scr-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>generate-scr-srcdescriptor</id>
+                        <goals>
+                            <goal>scr</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <supportedProjectTypes>
+                        <supportedProjectType>bundle</supportedProjectType>
+                        <supportedProjectType>war</supportedProjectType>
+                    </supportedProjectTypes>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.onosproject</groupId>
+                <artifactId>onos-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>cfg</id>
+                        <phase>generate-resources</phase>
+                        <goals>
+                            <goal>cfg</goal>
+                        </goals>
+                    </execution>
+                    <execution>
+                        <id>swagger</id>
+                        <goals>
+                            <goal>swagger</goal>
+                        </goals>
+                    </execution>
+                    <execution>
+                        <id>app</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>app</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

+/**
+ * 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 static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;;
+
+/**
+ * Returns the appropriate converter object based on the ONOS event type.
+ *
+ */
+public final class ConversionFactory {
+
+    // Store converters for all supported events
+    private Map<Type, EventConverter> converters =
+            new HashMap<Type, EventConverter>() {
+                {
+                    put(DEVICE, new DeviceEventConverter());
+                    put(LINK, new LinkEventConverter());
+                }
+            };
+
+    // Exists to defeat instantiation
+    private ConversionFactory() {
+    }
+
+    private static class SingletonHolder {
+        private static final ConversionFactory INSTANCE =
+                new ConversionFactory();
+    }
+
+    /**
+     * Returns a static reference to the Conversion Factory.
+     *
+     * @return singleton object
+     */
+    public static ConversionFactory getInstance() {
+        return SingletonHolder.INSTANCE;
+    }
+
+    /**
+     * Returns an Event converter object for the specified ONOS event type.
+     *
+     * @param event ONOS event type
+     * @return Event Converter object
+     */
+    public EventConverter getConverter(Type event) {
+        return converters.get(event);
+    }
+
+}

+/**
+ * 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.Device.DeviceCore;
+import org.onosproject.grpc.net.Device.DeviceType;
+import org.onosproject.grpc.net.DeviceEvent.DeviceEventType;
+import org.onosproject.grpc.net.DeviceEvent.DeviceNotification;
+import org.onosproject.grpc.net.Port.PortCore;
+import org.onosproject.grpc.net.Port.PortType;
+import org.onosproject.net.device.DeviceEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.protobuf.GeneratedMessage;
+
+/**
+ * Converts ONOS Device event message to GPB format.
+ *
+ */
+class DeviceEventConverter implements EventConverter {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public GeneratedMessage convertToProtoMessage(Event<?, ?> event) {
+
+        DeviceEvent deviceEvent = (DeviceEvent) event;
+
+        if (!deviceEventSubtypeSupported(deviceEvent)) {
+            log.error("Unsupported Onos Device Event {}. There is no matching"
+                    + "proto Device Event type", deviceEvent.type().toString());
+            return null;
+        }
+
+        return buildDeviceProtoMessage(deviceEvent);
+    }
+
+    /**
+     * Checks if the ONOS Device Event type is supported.
+     *
+     * @param event ONOS Device event
+     * @return true if there is a match and false otherwise
+     */
+    private boolean deviceEventSubtypeSupported(DeviceEvent event) {
+        DeviceEventType[] deviceEvents = DeviceEventType.values();
+        for (DeviceEventType deviceEventType : deviceEvents) {
+            if (deviceEventType.name().equals(event.type().name())) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    private DeviceNotification buildDeviceProtoMessage(DeviceEvent deviceEvent) {
+        DeviceNotification notification = DeviceNotification.newBuilder()
+                .setDeviceEventType(getProtoType(deviceEvent))
+                .setDevice(DeviceCore.newBuilder()
+                        .setChassisId(deviceEvent.subject().chassisId().id()
+                                .toString())
+                        .setDeviceId(deviceEvent.subject().id().toString())
+                        .setHwVersion(deviceEvent.subject().hwVersion())
+                        .setManufacturer(deviceEvent.subject().manufacturer())
+                        .setSerialNumber(deviceEvent.subject().serialNumber())
+                        .setSwVersion(deviceEvent.subject().swVersion())
+                        .setType(DeviceType.valueOf(deviceEvent.type().name()))
+                        .build())
+                .setPort(PortCore.newBuilder()
+                        .setIsEnabled(deviceEvent.port().isEnabled())
+                        .setPortNumber(deviceEvent.port().number().toString())
+                        .setPortSpeed(deviceEvent.port().portSpeed())
+                        .setType(PortType
+                                .valueOf(deviceEvent.port().type().name()))
+                        .build())
+                .build();
+
+        return notification;
+    }
+
+    /**
+     * Retrieves the protobuf generated device event type.
+     *
+     * @param event ONOS Device Event
+     * @return generated Device Event Type
+     */
+    private DeviceEventType getProtoType(DeviceEvent event) {
+        DeviceEventType protobufEventType = null;
+        DeviceEventType[] deviceEvents = DeviceEventType.values();
+        for (DeviceEventType deviceEventType : deviceEvents) {
+            if (deviceEventType.name().equals(event.type().name())) {
+                protobufEventType = deviceEventType;
+            }
+        }
+
+        return protobufEventType;
+    }
+}

+/**
+ * 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 com.google.protobuf.GeneratedMessage;
+
+/**
+ *
+ * APIs for converting between ONOS event objects and GPB data objects.
+ *
+ */
+public interface EventConverter {
+
+    /**
+     * Converts ONOS specific event data to a format that is suitable for export
+     * to Kafka.
+     *
+     * @param event ONOS Event object
+     * @return converted data in GPB format.
+     */
+    public GeneratedMessage convertToProtoMessage(Event<?, ?> event);
+}

+/**
+ * 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;
+    }
+}

+/**
+ * Copyright 2016-present 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.
+ */
+
+/**
+ * Converters for converting various ONOS events to their corresponding Protocol
+ * Buffer format.
+ *
+ */
+package org.onosproject.kafkaintegration.converter;

+/**
+ * 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.impl;
+
+import org.onosproject.event.AbstractListenerManager;
+import org.onosproject.kafkaintegration.api.ExportableEventListener;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.protobuf.GeneratedMessage;
+
+/**
+ * Dispatch ONOS Events to all interested Listeners.
+ *
+ */
+public final class Dispatcher
+        extends AbstractListenerManager<OnosEvent, ExportableEventListener> {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    // Exists to defeat instantiation
+    private Dispatcher() {
+    }
+
+    private static class SingletonHolder {
+        private static final Dispatcher INSTANCE = new Dispatcher();
+    }
+
+    /**
+     * Returns a static reference to the Listener Factory.
+     *
+     * @return singleton object
+     */
+    public static Dispatcher getInstance() {
+        return SingletonHolder.INSTANCE;
+    }
+
+    /**
+     * Publish the ONOS Event to all listeners.
+     *
+     * @param eventType the ONOS eventtype
+     * @param message generated Protocol buffer message from ONOS event data
+     */
+    public void publish(Type eventType, GeneratedMessage message) {
+        log.debug("Dispatching ONOS Event {}", eventType);
+        post(new OnosEvent(eventType, message));
+    }
+}

@@ -13,10 +13,14 @@
  * See the License for the specific language governing permissions and
  * See the License for the specific language governing permissions and
  * limitations under the License.
  * limitations under the License.
  */
  */
-package org.onosproject.kafkaintegration;
+package org.onosproject.kafkaintegration.impl;
 
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkNotNull;
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
 
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.Map;
 import java.util.UUID;
 import java.util.UUID;
 
 
@@ -31,9 +35,11 @@ import org.onosproject.core.CoreService;
 import org.onosproject.kafkaintegration.api.EventExporterService;
 import org.onosproject.kafkaintegration.api.EventExporterService;
 import org.onosproject.kafkaintegration.api.dto.EventSubscriber;
 import org.onosproject.kafkaintegration.api.dto.EventSubscriber;
 import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId;
 import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
 import org.onosproject.kafkaintegration.errors.InvalidApplicationException;
 import org.onosproject.kafkaintegration.errors.InvalidApplicationException;
 import org.onosproject.kafkaintegration.errors.InvalidGroupIdException;
 import org.onosproject.kafkaintegration.errors.InvalidGroupIdException;
-import org.onosproject.kafkaintegration.errors.UnsupportedEventException;
+import org.onosproject.kafkaintegration.listener.ListenerFactory;
+import org.onosproject.kafkaintegration.listener.OnosEventListener;
 import org.onosproject.net.device.DeviceService;
 import org.onosproject.net.device.DeviceService;
 import org.onosproject.net.link.LinkService;
 import org.onosproject.net.link.LinkService;
 import org.onosproject.store.serializers.KryoNamespaces;
 import org.onosproject.store.serializers.KryoNamespaces;
@@ -53,11 +59,14 @@ public class EventExporterManager implements EventExporterService {
     private final Logger log = LoggerFactory.getLogger(getClass());
     private final Logger log = LoggerFactory.getLogger(getClass());
 
 
     // Stores the currently registered applications for event export service.
     // Stores the currently registered applications for event export service.
-    // Map of Appname to groupId
     private Map<ApplicationId, EventSubscriberGroupId> registeredApps;
     private Map<ApplicationId, EventSubscriberGroupId> registeredApps;
 
 
+    private Map<Type, List<EventSubscriber>> subscriptions;
+
     private static final String REGISTERED_APPS = "registered-applications";
     private static final String REGISTERED_APPS = "registered-applications";
 
 
+    private static final String SUBSCRIBED_APPS = "event-subscriptions";
+
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected DeviceService deviceService;
     protected DeviceService deviceService;
 
 
@@ -70,8 +79,6 @@ public class EventExporterManager implements EventExporterService {
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected StorageService storageService;
     protected StorageService storageService;
 
 
-    private static final String NOT_YET_SUPPORTED = "Not yet supported.";
-
     private ApplicationId appId;
     private ApplicationId appId;
 
 
     @Activate
     @Activate
@@ -87,6 +94,13 @@ public class EventExporterManager implements EventExporterService {
                                                  UUID.class))
                                                  UUID.class))
                 .build().asJavaMap();
                 .build().asJavaMap();
 
 
+        subscriptions = storageService
+                .<Type, List<EventSubscriber>>consistentMapBuilder()
+                .withName(SUBSCRIBED_APPS)
+                .withSerializer(Serializer.using(KryoNamespaces.API,
+                                                 EventSubscriber.class))
+                .build().asJavaMap();
+
         log.info("Started");
         log.info("Started");
     }
     }
 
 
@@ -117,16 +131,193 @@ public class EventExporterManager implements EventExporterService {
 
 
     @Override
     @Override
     public void subscribe(EventSubscriber subscriber)
     public void subscribe(EventSubscriber subscriber)
-            throws UnsupportedEventException, InvalidGroupIdException,
-            InvalidApplicationException {
+            throws InvalidGroupIdException, InvalidApplicationException {
 
 
-        throw new UnsupportedOperationException(NOT_YET_SUPPORTED);
+        checkNotNull(subscriber);
+
+        if (!registeredApplication(subscriber.appName())) {
+            throw new InvalidApplicationException("Application is not "
+                    + "registered to make this request.");
+        }
+
+        if (!validGroupId(subscriber.subscriberGroupId(),
+                          subscriber.appName())) {
+            throw new InvalidGroupIdException("Incorrect group id in the request");
+        }
+
+        OnosEventListener onosListener = getListener(subscriber.eventType());
+        checkNotNull(onosListener,
+                     "No listener for the supported event type - {}",
+                     subscriber.eventType());
+
+        applyListenerAction(subscriber.eventType(), onosListener,
+                            ListenerAction.START);
+
+        // update internal state
+        List<EventSubscriber> subscriptionList =
+                subscriptions.get(subscriber.eventType());
+        if (subscriptionList == null) {
+            subscriptionList = new ArrayList<EventSubscriber>();
+        }
+        subscriptionList.add(subscriber);
+        subscriptions.put(subscriber.eventType(), subscriptionList);
+
+        log.info("Subscription for {} event by {} successfull",
+                 subscriber.eventType(), subscriber.appName());
+    }
+
+    /**
+     * Checks if the application has registered.
+     *
+     * @param appName application name
+     * @return true if application has registered
+     */
+    private boolean registeredApplication(String appName) {
+
+        checkNotNull(appName);
+        ApplicationId appId = checkNotNull(coreService.getAppId(appName));
+        if (registeredApps.containsKey(appId)) {
+            return true;
+        }
+
+        log.debug("{} is not registered", appName);
+        return false;
+    }
+
+    /**
+     * Actions that can be performed on the ONOS Event Listeners.
+     *
+     */
+    private enum ListenerAction {
+        START, STOP;
+    }
+
+    /**
+     * Applies the specified action on the Listener.
+     *
+     * @param eventType the ONOS Event type registered by the application
+     * @param onosListener ONOS event listener
+     * @param action to be performed on the listener
+     */
+    private void applyListenerAction(Type eventType,
+                                     OnosEventListener onosListener,
+                                     ListenerAction action) {
+        switch (eventType) {
+        case DEVICE:
+            if (action == ListenerAction.START) {
+                onosListener.startListener(DEVICE, deviceService);
+            } else {
+                onosListener.stopListener(DEVICE, deviceService);
+            }
+            break;
+        case LINK:
+            if (action == ListenerAction.START) {
+                onosListener.startListener(LINK, linkService);
+            } else {
+                onosListener.stopListener(LINK, linkService);
+            }
+            break;
+        default:
+            log.error("Cannot {} listener. Unsupported event type {} ",
+                      action.toString(), eventType.toString());
+        }
+    }
+
+    /**
+     * Returns the ONOS event listener corresponding to the ONOS Event type.
+     *
+     * @param eventType ONOS event type
+     * @return ONOS event listener
+     */
+    private OnosEventListener getListener(Type eventType) {
+        checkNotNull(eventType);
+        ListenerFactory factory = ListenerFactory.getInstance();
+        OnosEventListener onosListener = factory.getListener(eventType);
+        return onosListener;
+    }
+
+    /**
+     * Checks if the group id is valid for this registered application.
+     *
+     * @param groupId GroupId assigned to the subscriber
+     * @param appName Registered Application name
+     * @return true if valid groupId and false otherwise
+     */
+    private boolean validGroupId(EventSubscriberGroupId groupId,
+                                 String appName) {
+
+        checkNotNull(groupId);
+
+        ApplicationId appId = coreService.getAppId(appName);
+        EventSubscriberGroupId registeredGroupId = registeredApps.get(appId);
+        if (registeredGroupId.equals(groupId)) {
+            return true;
+        }
+
+        return false;
     }
     }
 
 
     @Override
     @Override
     public void unsubscribe(EventSubscriber subscriber)
     public void unsubscribe(EventSubscriber subscriber)
             throws InvalidGroupIdException, InvalidApplicationException {
             throws InvalidGroupIdException, InvalidApplicationException {
 
 
-        throw new UnsupportedOperationException(NOT_YET_SUPPORTED);
+        checkNotNull(subscriber);
+
+        if (!registeredApplication(subscriber.appName())) {
+            throw new InvalidApplicationException("Application is not "
+                    + "registered to make this request.");
+        }
+
+        if (!validGroupId(subscriber.subscriberGroupId(),
+                          subscriber.appName())) {
+            throw new InvalidGroupIdException("Incorrect group id in the request");
         }
         }
+
+        if (!eventSubscribed(subscriber)) {
+            log.error("No subscription to {} was found",
+                      subscriber.eventType());
+            return;
+        }
+
+        // If this is the only subscriber listening for this event,
+        // stop the listener.
+        List<EventSubscriber> subscribers =
+                subscriptions.get(subscriber.eventType());
+        if (subscribers.size() == 1) {
+            OnosEventListener onosListener =
+                    getListener(subscriber.eventType());
+            checkNotNull(onosListener,
+                         "No listener for the supported event type - {}",
+                         subscriber.eventType());
+            applyListenerAction(subscriber.eventType(), onosListener,
+                                ListenerAction.STOP);
+        }
+
+        // update internal state.
+        subscribers.remove(subscriber);
+        subscriptions.put(subscriber.eventType(), subscribers);
+
+        log.info("Unsubscribed {} for {} events", subscriber.appName(),
+                 subscriber.eventType());
+    }
+
+    /**
+     * Checks if the subscriber has already subscribed to the requested event
+     * type.
+     *
+     * @param subscriber the subscriber to a specific ONOS event
+     * @return true if subscriber has subscribed to the ONOS event
+     */
+    private boolean eventSubscribed(EventSubscriber subscriber) {
+
+        List<EventSubscriber> subscriberList =
+                subscriptions.get(subscriber.eventType());
+
+        if (subscriberList == null) {
+            return false;
+        }
+
+        return subscriberList.contains(subscriber);
+    }
+
 }
 }

@@ -15,4 +15,4 @@
 /**
 /**
  * API implementation classes.
  * API implementation classes.
  */
  */
-package org.onosproject.kafkaintegration;
+package org.onosproject.kafkaintegration.impl;

+/**
+ * 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.listener;
+
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
+
+import org.onosproject.event.ListenerService;
+import org.onosproject.kafkaintegration.impl.Dispatcher;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+import org.onosproject.kafkaintegration.converter.ConversionFactory;
+import org.onosproject.kafkaintegration.converter.EventConverter;
+import org.onosproject.net.device.DeviceEvent;
+import org.onosproject.net.device.DeviceListener;
+import org.onosproject.net.device.DeviceService;
+
+import com.google.protobuf.GeneratedMessage;
+
+/**
+ * Listens for ONOS Device events.
+ *
+ */
+final class DeviceEventsListener implements OnosEventListener {
+
+    private boolean listenerRunning = false;
+
+    private InnerListener listener = null;
+
+    // Exists to defeat instantiation
+    private DeviceEventsListener() {
+    }
+
+    private static class SingletonHolder {
+        private static final DeviceEventsListener INSTANCE =
+                new DeviceEventsListener();
+    }
+
+    /**
+     * Returns a static reference to the Listener Factory.
+     *
+     * @return singleton object
+     */
+    public static DeviceEventsListener getInstance() {
+        return SingletonHolder.INSTANCE;
+    }
+
+    @Override
+    public void startListener(Type e, ListenerService<?, ?> service) {
+        if (!listenerRunning) {
+            listener = new InnerListener();
+            DeviceService deviceService = (DeviceService) service;
+            deviceService.addListener(listener);
+            listenerRunning = true;
+        }
+    }
+
+    private class InnerListener implements DeviceListener {
+
+        @Override
+        public void event(DeviceEvent arg0) {
+
+            // Convert the event to GPB format
+            ConversionFactory conversionFactory =
+                    ConversionFactory.getInstance();
+            EventConverter converter = conversionFactory.getConverter(DEVICE);
+            GeneratedMessage message = converter.convertToProtoMessage(arg0);
+
+            // Call Dispatcher and publish event
+            Dispatcher.getInstance().publish(DEVICE, message);
+        }
+    }
+
+    @Override
+    public void stopListener(Type e, ListenerService<?, ?> service) {
+        if (listenerRunning) {
+            DeviceService deviceService = (DeviceService) service;
+            deviceService.removeListener(listener);
+            listener = null;
+            listenerRunning = false;
+        }
+    }
+
+}

+/**
+ * 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.listener;
+
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
+
+import org.onosproject.event.ListenerService;
+import org.onosproject.kafkaintegration.impl.Dispatcher;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+import org.onosproject.kafkaintegration.converter.ConversionFactory;
+import org.onosproject.kafkaintegration.converter.EventConverter;
+import org.onosproject.net.link.LinkEvent;
+import org.onosproject.net.link.LinkListener;
+import org.onosproject.net.link.LinkService;
+
+import com.google.protobuf.GeneratedMessage;
+
+/**
+ * Listens for ONOS Link Events.
+ *
+ */
+final class LinkEventsListener implements OnosEventListener {
+
+    private boolean listenerRunning = false;
+
+    private InnerListener listener = null;
+
+    // Exists to defeat instantiation
+    private LinkEventsListener() {
+    }
+
+    private static class SingletonHolder {
+        private static final LinkEventsListener INSTANCE =
+                new LinkEventsListener();
+    }
+
+    /**
+     * Returns a static reference to the Listener Factory.
+     *
+     * @return singleton object
+     */
+    public static LinkEventsListener getInstance() {
+        return SingletonHolder.INSTANCE;
+    }
+
+    @Override
+    public void startListener(Type e, ListenerService<?, ?> service) {
+        if (!listenerRunning) {
+            listener = new InnerListener();
+            LinkService linkService = (LinkService) service;
+            linkService.addListener(listener);
+            listenerRunning = true;
+        }
+    }
+
+    private class InnerListener implements LinkListener {
+
+        @Override
+        public void event(LinkEvent arg0) {
+
+            // Convert the event to GPB format
+            ConversionFactory conversionFactory =
+                    ConversionFactory.getInstance();
+            EventConverter converter = conversionFactory.getConverter(LINK);
+            GeneratedMessage message = converter.convertToProtoMessage(arg0);
+
+            // Call Dispatcher and publish event
+            Dispatcher.getInstance().publish(LINK, message);
+        }
+    }
+
+    @Override
+    public void stopListener(Type e, ListenerService<?, ?> service) {
+        if (listenerRunning) {
+            LinkService linkService = (LinkService) service;
+            linkService.removeListener(listener);
+            listenerRunning = false;
+        }
+    }
+}

+/**
+ * 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.listener;
+
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+
+/**
+ * Returns the appropriate listener object based on the ONOS event type.
+ *
+ */
+public final class ListenerFactory {
+
+    // Store listeners for all supported events
+    private Map<Type, OnosEventListener> listeners =
+            new HashMap<Type, OnosEventListener>() {
+                {
+                    put(DEVICE, DeviceEventsListener.getInstance());
+                    put(LINK, LinkEventsListener.getInstance());
+                }
+            };
+
+    // Exists to defeat instantiation
+    private ListenerFactory() {
+    }
+
+    private static class SingletonHolder {
+        private static final ListenerFactory INSTANCE = new ListenerFactory();
+    }
+
+    /**
+     * Returns a static reference to the Listener Factory.
+     *
+     * @return singleton object
+     */
+    public static ListenerFactory getInstance() {
+        return SingletonHolder.INSTANCE;
+    }
+
+    /**
+     * Returns the listener object for the specified ONOS event type.
+     *
+     * @param event ONOS Event type
+     * @return return listener object
+     */
+    public OnosEventListener getListener(Type event) {
+        return listeners.get(event);
+    }
+
+}

+/**
+ * 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.listener;
+
+import org.onosproject.event.ListenerService;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+
+/**
+ * APIs for starting and stopping a ONOS Event listener.
+ *
+ */
+public interface OnosEventListener {
+
+    /**
+     * Start the listener for the specific ONOS event type.
+     *
+     * @param event ONOS event type
+     * @param service ONOS event listener for the specific event type
+     */
+    void startListener(Type event, ListenerService<?, ?> service);
+
+    /**
+     * Stop the Listener for the specific ONOS event type.
+     *
+     * @param event ONOS event type
+     * @param service ONOS event listener for the specific event type
+     */
+    void stopListener(Type event, ListenerService<?, ?> service);
+
+}

+/**
+ * Copyright 2016-present 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.
+ */
+
+/**
+ * Listeners for listening to various ONOS events.
+ *
+ */
+package org.onosproject.kafkaintegration.listener;

@@ -14,15 +14,11 @@
  */
  */
 package org.onosproject.kafkaintegration.rest;
 package org.onosproject.kafkaintegration.rest;
 
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-import org.onosproject.codec.JsonCodec;
-import org.onosproject.kafkaintegration.api.EventExporterService;
-import org.onosproject.kafkaintegration.api.dto.EventSubscriber;
-import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId;
-import org.onosproject.rest.AbstractWebResource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static javax.ws.rs.core.Response.Status.BAD_REQUEST;
+
+import java.io.IOException;
+import java.io.InputStream;
 
 
 import javax.ws.rs.Consumes;
 import javax.ws.rs.Consumes;
 import javax.ws.rs.DELETE;
 import javax.ws.rs.DELETE;
@@ -31,11 +27,17 @@ import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
 import javax.ws.rs.Produces;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.Response;
-import java.io.IOException;
-import java.io.InputStream;
 
 
-import static com.google.common.base.Preconditions.checkNotNull;
-import static javax.ws.rs.core.Response.Status.BAD_REQUEST;
+import org.onosproject.codec.JsonCodec;
+import org.onosproject.kafkaintegration.api.EventExporterService;
+import org.onosproject.kafkaintegration.api.dto.EventSubscriber;
+import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId;
+import org.onosproject.rest.AbstractWebResource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 
 
 /**
 /**
  * Rest Interfaces for subscribing/unsubscribing to event notifications.
  * Rest Interfaces for subscribing/unsubscribing to event notifications.
@@ -44,14 +46,19 @@ import static javax.ws.rs.core.Response.Status.BAD_REQUEST;
 public class EventExporterWebResource extends AbstractWebResource {
 public class EventExporterWebResource extends AbstractWebResource {
 
 
     private final Logger log = LoggerFactory.getLogger(getClass());
     private final Logger log = LoggerFactory.getLogger(getClass());
-    public static final String JSON_NOT_NULL = "Registration Data cannot be empty";
-    public static final String REGISTRATION_SUCCESSFUL = "Registered Listener successfully";
-    public static final String DEREGISTRATION_SUCCESSFUL = "De-Registered Listener successfully";
-    public static final String EVENT_SUBSCRIPTION_SUCCESSFUL = "Event Registration successfull";
-    public static final String EVENT_SUBSCRIPTION_REMOVED = "Event De-Registration successfull";
+    public static final String JSON_NOT_NULL =
+            "Registration Data cannot be empty";
+    public static final String REGISTRATION_SUCCESSFUL =
+            "Registered Listener successfully";
+    public static final String DEREGISTRATION_SUCCESSFUL =
+            "De-Registered Listener successfully";
+    public static final String EVENT_SUBSCRIPTION_SUCCESSFUL =
+            "Event Registration successfull";
+    public static final String EVENT_SUBSCRIPTION_REMOVED =
+            "Event De-Registration successfull";
 
 
     /**
     /**
-     * Registers a listener for Onos Events.
+     * Registers a listener for ONOS Events.
      *
      *
      * @param appName The application trying to register
      * @param appName The application trying to register
      * @return 200 OK with UUID string which should be used as Kafka Consumer
      * @return 200 OK with UUID string which should be used as Kafka Consumer
@@ -76,7 +83,7 @@ public class EventExporterWebResource extends AbstractWebResource {
     }
     }
 
 
     /**
     /**
-     * Unregisters a listener for Onos Events.
+     * Unregisters a listener for ONOS Events.
      *
      *
      * @param appName The application trying to unregister
      * @param appName The application trying to unregister
      * @return 200 OK
      * @return 200 OK
@@ -93,9 +100,9 @@ public class EventExporterWebResource extends AbstractWebResource {
     }
     }
 
 
     /**
     /**
-     * Creates subscription to a specific Onos event.
+     * Creates subscription to a specific ONOS event.
      *
      *
-     * @param input Subscription Data in Json format
+     * @param input Subscription Data in JSON format
      * @return 200 OK if successful or 400 BAD REQUEST
      * @return 200 OK if successful or 400 BAD REQUEST
      * @onos.rsModel KafkaSubscription
      * @onos.rsModel KafkaSubscription
      */
      */
@@ -118,9 +125,9 @@ public class EventExporterWebResource extends AbstractWebResource {
     }
     }
 
 
     /**
     /**
-     * Parses Json Subscription Data from the external application.
+     * Parses JSON Subscription Data from the external application.
      *
      *
-     * @param node node within the parsed json tree.
+     * @param input Subscription Data in JSON format
      * @return parsed DTO object
      * @return parsed DTO object
      * @throws IOException
      * @throws IOException
      */
      */
@@ -137,9 +144,9 @@ public class EventExporterWebResource extends AbstractWebResource {
     }
     }
 
 
     /**
     /**
-     * Deletes subscription from a specific Onos event.
+     * Deletes subscription from a specific ONOS event.
      *
      *
-     * @param input data in json format
+     * @param input data in JSON format
      * @return 200 OK if successful or 400 BAD REQUEST
      * @return 200 OK if successful or 400 BAD REQUEST
      * @onos.rsModel KafkaSubscription
      * @onos.rsModel KafkaSubscription
      */
      */
@@ -152,7 +159,7 @@ public class EventExporterWebResource extends AbstractWebResource {
 
 
         try {
         try {
             EventSubscriber sub = parseSubscriptionData(input);
             EventSubscriber sub = parseSubscriptionData(input);
-            service.subscribe(sub);
+            service.unsubscribe(sub);
         } catch (Exception e) {
         } catch (Exception e) {
             log.error(e.getMessage());
             log.error(e.getMessage());
             return Response.status(BAD_REQUEST).entity(e.getMessage()).build();
             return Response.status(BAD_REQUEST).entity(e.getMessage()).build();

@@ -14,5 +14,6 @@
 
 
 /**
 /**
  * REST API Definitions.
  * REST API Definitions.
+ *
  */
  */
 package org.onosproject.kafkaintegration.rest;
 package org.onosproject.kafkaintegration.rest;

-<?sxml version="1.0" encoding="UTF-8"?>
+<?xml version="1.0" encoding="UTF-8"?>
 <!--
 <!--
-  ~ Copyright 2016 Open Networking Laboratory
+  ~ Copyright 2016-present Open Networking Laboratory
   ~
   ~
   ~ Licensed under the Apache License, Version 2.0 (the "License");
   ~ Licensed under the Apache License, Version 2.0 (the "License");
   ~ you may not use this file except in compliance with the License.
   ~ you may not use this file except in compliance with the License.
@@ -14,218 +14,26 @@
   ~ See the License for the specific language governing permissions and
   ~ See the License for the specific language governing permissions and
   ~ limitations under the License.
   ~ limitations under the License.
   -->
   -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <modelVersion>4.0.0</modelVersion>
-
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <parent>
     <parent>
         <groupId>org.onosproject</groupId>
         <groupId>org.onosproject</groupId>
         <artifactId>onos-apps</artifactId>
         <artifactId>onos-apps</artifactId>
         <version>1.7.0-SNAPSHOT</version>
         <version>1.7.0-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
         <relativePath>../pom.xml</relativePath>
     </parent>
     </parent>
+    <modelVersion>4.0.0</modelVersion>
 
 
-    <artifactId>onos-app-kafka</artifactId>
-    <packaging>bundle</packaging>
-
-    <description>
-        ONOS Kafka Integration Application.
-        This will export ONOS events to an external Kafka Server
-    </description>
-    <url>http://onosproject.org</url>
-
-    <properties>
-        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <onos.version>1.6.0-SNAPSHOT</onos.version>
-        <onos.app.name>org.onosproject.kafkaintegration</onos.app.name>
-        <onos.app.title>Kafka Integration Application</onos.app.title>
-        <onos.app.origin>Calix, Inc.</onos.app.origin>
-        <web.context>/onos/kafka</web.context>
-        <api.version>1.0.0</api.version>
-        <api.package>org.onosproject.kafkaintegration.rest</api.package>
-        <api.title>Kafka Integration Application REST API</api.title>
-        <api.description>
-               APIs for subscribing to Events generated by onos
-        </api.description>
-        <onos.app.category>Utility</onos.app.category>
-        <onos.app.url>https://wiki.onosproject.org/display/ONOS/Kafka+Integration</onos.app.url>
-        <onos.app.readme>Export onos events to a Northbound Kafka server</onos.app.readme>
-    </properties>
-
-    <dependencies>
-        <dependency>
-            <groupId>org.onosproject</groupId>
-            <artifactId>onos-api</artifactId>
-            <version>${onos.version}</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.onosproject</groupId>
-            <artifactId>onlab-osgi</artifactId>
-            <version>${onos.version}</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.onosproject</groupId>
-            <artifactId>onos-rest</artifactId>
-            <version>${onos.version}</version>
-        </dependency>
-
-        <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-            <version>4.12</version>
-            <scope>test</scope>
-        </dependency>
-
-        <dependency>
-            <groupId>org.onosproject</groupId>
-            <artifactId>onos-api</artifactId>
-            <version>${onos.version}</version>
-            <scope>test</scope>
-            <classifier>tests</classifier>
-        </dependency>
-
-        <dependency>
-            <groupId>javax.ws.rs</groupId>
-            <artifactId>javax.ws.rs-api</artifactId>
-            <version>2.0.1</version>
-        </dependency>
-
-        <dependency>
-            <groupId>com.google.protobuf</groupId>
-            <artifactId>protobuf-java</artifactId>
-            <version>2.5.0</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.codehaus.jackson</groupId>
-            <artifactId>jackson-core-asl</artifactId>
-            <version>1.9.13</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.codehaus.jackson</groupId>
-            <artifactId>jackson-mapper-asl</artifactId>
-            <version>1.9.13</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.glassfish.jersey.containers</groupId>
-            <artifactId>jersey-container-servlet</artifactId>
-            <version>2.22.2</version>
-        </dependency>
-
-        <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-annotations</artifactId>
-            <version>2.6.4</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.onosproject</groupId>
-            <artifactId>onos-core-serializers</artifactId>
-            <version>${onos.version}</version>
-        </dependency>
+    <artifactId>onos-kafka</artifactId>
 
 
-        <dependency>
-            <groupId>org.apache.felix</groupId>
-            <artifactId>org.apache.felix.scr.annotations</artifactId>
-            <version>1.9.12</version>
-            <scope>provided</scope>
-        </dependency>
-    </dependencies>
+    <packaging>pom</packaging>
+    <description>Kafka Integration Application</description>
 
 
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>org.apache.felix</groupId>
-                <artifactId>maven-bundle-plugin</artifactId>
-                <version>3.0.1</version>
-                <extensions>true</extensions>
-                <configuration>
-                    <instructions>
-                    <Bundle-SymbolicName>
-                        ${project.groupId}.${project.artifactId}
-                    </Bundle-SymbolicName>
-                    <_wab>src/main/webapp/</_wab>
-                    <Include-Resource>
-                        WEB-INF/classes/apidoc/swagger.json=target/swagger.json,
-                                    {maven-resources}
-                    </Include-Resource>
-                    <Import-Package>
-                        org.slf4j,
-                        org.osgi.framework,
-                        javax.ws.rs,
-                        javax.ws.rs.core,
-                        org.glassfish.jersey.servlet,
-                        com.fasterxml.jackson.databind,
-                        com.fasterxml.jackson.databind.node,
-                        com.fasterxml.jackson.core,
-                        org.onlab.packet.*,
-                        org.onosproject.*,
-                        com.google.common.*
-                    </Import-Package>
-                    <Web-ContextPath>${web.context}</Web-ContextPath>
-                </instructions>
-                </configuration>
-            </plugin>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-compiler-plugin</artifactId>
-                <version>2.5.1</version>
-                <configuration>
-                    <source>1.8</source>
-                    <target>1.8</target>
-                </configuration>
-            </plugin>
-            <plugin>
-                <groupId>org.apache.felix</groupId>
-                <artifactId>maven-scr-plugin</artifactId>
-                <version>1.21.0</version>
-                <executions>
-                    <execution>
-                        <id>generate-scr-srcdescriptor</id>
-                        <goals>
-                            <goal>scr</goal>
-                        </goals>
-                    </execution>
-                </executions>
-                <configuration>
-                    <supportedProjectTypes>
-                        <supportedProjectType>bundle</supportedProjectType>
-                        <supportedProjectType>war</supportedProjectType>
-                    </supportedProjectTypes>
-                </configuration>
-            </plugin>
-            <plugin>
-                <groupId>org.onosproject</groupId>
-                <artifactId>onos-maven-plugin</artifactId>
-                <version>1.9</version>
-                <executions>
-                    <execution>
-                        <id>cfg</id>
-                        <phase>generate-resources</phase>
-                        <goals>
-                            <goal>cfg</goal>
-                        </goals>
-                    </execution>
-                    <execution>
-                        <id>swagger</id>
-                        <phase>generate-sources</phase>
-                        <goals>
-                            <goal>swagger</goal>
-                        </goals>
-                    </execution>
-                    <execution>
-                        <id>app</id>
-                        <phase>package</phase>
-                        <goals>
-                            <goal>app</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
-        </plugins>
-    </build>
+    <modules>
+        <module>api</module>
+        <module>app</module>
+    </modules>
 
 
 </project>
 </project>
+

@@ -3,6 +3,23 @@ option java_package = "org.onosproject.grpc.net";
 
 
 package Device;
 package Device;
 
 
+message DeviceDescription {
+  string device_Uri = 1;
+  DeviceType type = 2;
+  string manufacturer = 3;
+  string hw_version = 4;
+  string sw_version = 5;
+  string serial_number = 6;
+  string chassis_id = 7;
+  map<string, string> annotations = 8;
+}
+
+enum MastershipRole {
+  NONE = 0;
+  MASTER = 1;
+  STANDBY = 2;
+}
+
 enum DeviceType {
 enum DeviceType {
   OTHER = 0;
   OTHER = 0;
   SWITCH = 1;
   SWITCH = 1;
@@ -20,8 +37,9 @@ enum DeviceType {
   MICROWAVE = 13;
   MICROWAVE = 13;
 }
 }
 
 
-message DeviceDescription {
-  string device_Uri = 1;
+// Corresponds to org.onosproject.net.Device.
+message DeviceCore {
+  string deviceId = 1;
   DeviceType type = 2;
   DeviceType type = 2;
   string manufacturer = 3;
   string manufacturer = 3;
   string hw_version = 4;
   string hw_version = 4;
@@ -30,9 +48,3 @@ message DeviceDescription {
   string chassis_id = 7;
   string chassis_id = 7;
   map<string, string> annotations = 8;
   map<string, string> annotations = 8;
 } 
 } 
\ No newline at end of file
-
-enum MastershipRole {
-  NONE = 0;
-  MASTER = 1;
-  STANDBY = 2;
-}

+syntax = "proto3";
+option java_package = "org.onosproject.grpc.net";
+
+
+import "Device.proto";
+import "Port.proto";
+
+package DeviceEvent;
+
+// Corresponds to org.onosproject.net.device.DeviceEvent.
+message DeviceNotification {
+    Device.DeviceCore device = 1;
+    DeviceEventType deviceEventType = 2;
+    Port.PortCore port = 3;
+}
+
+enum DeviceEventType {
+    DEVICE_ADDED = 0;
+    DEVICE_UPDATED = 1;
+    DEVICE_REMOVED = 2;
+    DEVICE_SUSPENDED = 3;
+    DEVICE_AVAILABILITY_CHANGED = 4;
+    PORT_ADDED = 5;
+    PORT_UPDATED = 6;
+    PORT_REMOVED = 7;
+    PORT_STATS_UPDATED = 8;
+}
+
+
+

@@ -38,6 +38,20 @@ message ConnectPoint {
   string port_number = 2;
   string port_number = 2;
 }
 }
 
 
+enum LinkState {
+    ACTIVE = 0;
+    INACTIVE = 1;
+}
+
+// Corresponds to org.onosproject.net.Link.
+message LinkCore {
+  LinkState state = 1;
+  ConnectPoint src = 2;
+  ConnectPoint dst = 3;
+  LinkType type = 4;
+  map<string, string> annotations = 5;
+}
+
 message LinkDescription {
 message LinkDescription {
   ConnectPoint src = 1;
   ConnectPoint src = 1;
   ConnectPoint dst = 2;
   ConnectPoint dst = 2;

+syntax = "proto3";
+option java_package = "org.onosproject.grpc.net";
+
+package LinkEvent;
+
+import "Link.proto";
+
+// Corresponds to org.onosproject.net.link.LinkEvent.
+message LinkNotification {
+    LinkEventType linkEventType = 2;
+    Link.LinkCore link = 3;
+}
+
+// Link Event Types
+enum LinkEventType {
+    LINK_ADDED = 0;
+    LINK_UPDATED = 1;
+    LINK_REMOVED = 2;
+}

@@ -31,6 +31,15 @@ message PortDescription {
   map<string, string> annotations = 8;
   map<string, string> annotations = 8;
 }
 }
 
 
+// Corresponds to org.onosproject.net.Port.
+message PortCore {
+  string port_number = 1;
+  bool is_enabled = 2;
+  PortType type = 3;
+  int64 port_speed = 4;
+  map<string, string> annotations = 5;
+}
+
 message PortStatistics {
 message PortStatistics {
   int32 port = 1;
   int32 port = 1;
   int64 packets_received = 2;
   int64 packets_received = 2;