Added support for replicating device removed topology events

@@ -120,6 +120,7 @@ public class GossipDeviceStore
                     .register(KryoPoolUtil.API)
                     .register(KryoPoolUtil.API)
                     .register(InternalDeviceEvent.class, new InternalDeviceEventSerializer())
                     .register(InternalDeviceEvent.class, new InternalDeviceEventSerializer())
                     .register(InternalDeviceOfflineEvent.class, new InternalDeviceOfflineEventSerializer())
                     .register(InternalDeviceOfflineEvent.class, new InternalDeviceOfflineEventSerializer())
+                    .register(InternalDeviceRemovedEvent.class)
                     .register(InternalPortEvent.class, new InternalPortEventSerializer())
                     .register(InternalPortEvent.class, new InternalPortEventSerializer())
                     .register(InternalPortStatusEvent.class, new InternalPortStatusEventSerializer())
                     .register(InternalPortStatusEvent.class, new InternalPortStatusEventSerializer())
                     .register(Timestamp.class)
                     .register(Timestamp.class)
@@ -138,6 +139,8 @@ public class GossipDeviceStore
         clusterCommunicator.addSubscriber(
         clusterCommunicator.addSubscriber(
                 GossipDeviceStoreMessageSubjects.DEVICE_OFFLINE, new InternalDeviceOfflineEventListener());
                 GossipDeviceStoreMessageSubjects.DEVICE_OFFLINE, new InternalDeviceOfflineEventListener());
         clusterCommunicator.addSubscriber(
         clusterCommunicator.addSubscriber(
+                GossipDeviceStoreMessageSubjects.DEVICE_REMOVED, new InternalDeviceRemovedEventListener());
+        clusterCommunicator.addSubscriber(
                 GossipDeviceStoreMessageSubjects.PORT_UPDATE, new InternalPortEventListener());
                 GossipDeviceStoreMessageSubjects.PORT_UPDATE, new InternalPortEventListener());
         clusterCommunicator.addSubscriber(
         clusterCommunicator.addSubscriber(
                 GossipDeviceStoreMessageSubjects.PORT_STATUS_UPDATE, new InternalPortStatusEventListener());
                 GossipDeviceStoreMessageSubjects.PORT_STATUS_UPDATE, new InternalPortStatusEventListener());
@@ -583,7 +586,16 @@ public class GossipDeviceStore
     public synchronized DeviceEvent removeDevice(DeviceId deviceId) {
     public synchronized DeviceEvent removeDevice(DeviceId deviceId) {
         Timestamp timestamp = clockService.getTimestamp(deviceId);
         Timestamp timestamp = clockService.getTimestamp(deviceId);
         DeviceEvent event = removeDeviceInternal(deviceId, timestamp);
         DeviceEvent event = removeDeviceInternal(deviceId, timestamp);
-        // TODO: broadcast removal event
+        if (event != null) {
+            log.info("Notifying peers of a device removed topology event for deviceId: {}",
+                    deviceId);
+            try {
+                notifyPeers(new InternalDeviceRemovedEvent(deviceId, timestamp));
+            } catch (IOException e) {
+                log.error("Failed to notify peers of a device removed topology event for deviceId: {}",
+                     deviceId);
+            }
+        }
         return event;
         return event;
     }
     }
 
 
@@ -834,6 +846,14 @@ public class GossipDeviceStore
         clusterCommunicator.broadcast(message);
         clusterCommunicator.broadcast(message);
     }
     }
 
 
+    private void notifyPeers(InternalDeviceRemovedEvent event) throws IOException {
+        ClusterMessage message = new ClusterMessage(
+                clusterService.getLocalNode().id(),
+                GossipDeviceStoreMessageSubjects.DEVICE_REMOVED,
+                SERIALIZER.encode(event));
+        clusterCommunicator.broadcast(message);
+    }
+
     private void notifyPeers(InternalPortEvent event) throws IOException {
     private void notifyPeers(InternalPortEvent event) throws IOException {
         ClusterMessage message = new ClusterMessage(
         ClusterMessage message = new ClusterMessage(
                 clusterService.getLocalNode().id(),
                 clusterService.getLocalNode().id(),
@@ -879,6 +899,20 @@ public class GossipDeviceStore
         }
         }
     }
     }
 
 
+    private class InternalDeviceRemovedEventListener implements ClusterMessageHandler {
+        @Override
+        public void handle(ClusterMessage message) {
+
+            log.info("Received device removed event from peer: {}", message.sender());
+            InternalDeviceRemovedEvent event = (InternalDeviceRemovedEvent) SERIALIZER.decode(message.payload());
+
+            DeviceId deviceId = event.deviceId();
+            Timestamp timestamp = event.timestamp();
+
+            removeDeviceInternal(deviceId, timestamp);
+        }
+    }
+
     private class InternalPortEventListener implements ClusterMessageHandler {
     private class InternalPortEventListener implements ClusterMessageHandler {
         @Override
         @Override
         public void handle(ClusterMessage message) {
         public void handle(ClusterMessage message) {

@@ -11,6 +11,7 @@ public final class GossipDeviceStoreMessageSubjects {
 
 
     public static final MessageSubject DEVICE_UPDATE = new MessageSubject("peer-device-update");
     public static final MessageSubject DEVICE_UPDATE = new MessageSubject("peer-device-update");
     public static final MessageSubject DEVICE_OFFLINE = new MessageSubject("peer-device-offline");
     public static final MessageSubject DEVICE_OFFLINE = new MessageSubject("peer-device-offline");
+    public static final MessageSubject DEVICE_REMOVED = new MessageSubject("peer-device-removed");
     public static final MessageSubject PORT_UPDATE = new MessageSubject("peer-port-update");
     public static final MessageSubject PORT_UPDATE = new MessageSubject("peer-port-update");
     public static final MessageSubject PORT_STATUS_UPDATE = new MessageSubject("peer-port-status-update");
     public static final MessageSubject PORT_STATUS_UPDATE = new MessageSubject("peer-port-status-update");
 }
 }

+package org.onlab.onos.store.device.impl;
+
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.store.Timestamp;
+
+/**
+ * Information published by GossipDeviceStore to notify peers of a device
+ * being administratively removed.
+ */
+public class InternalDeviceRemovedEvent {
+
+    private final DeviceId deviceId;
+    private final Timestamp timestamp;
+
+    /**
+     * Creates a InternalDeviceRemovedEvent.
+     * @param deviceId identifier of the removed device.
+     * @param timestamp timestamp of when the device was administratively removed.
+     */
+    public InternalDeviceRemovedEvent(DeviceId deviceId, Timestamp timestamp) {
+        this.deviceId = deviceId;
+        this.timestamp = timestamp;
+    }
+
+    public DeviceId deviceId() {
+        return deviceId;
+    }
+
+    public Timestamp timestamp() {
+        return timestamp;
+    }
+
+    // for serializer
+    @SuppressWarnings("unused")
+    private InternalDeviceRemovedEvent() {
+        deviceId = null;
+        timestamp = null;
+    }
+}