Added support for replicating device offline topology events

@@ -119,8 +119,10 @@ public class GossipDeviceStore
             serializerPool = KryoPool.newBuilder()
             serializerPool = KryoPool.newBuilder()
                     .register(KryoPoolUtil.API)
                     .register(KryoPoolUtil.API)
                     .register(InternalDeviceEvent.class, new InternalDeviceEventSerializer())
                     .register(InternalDeviceEvent.class, new InternalDeviceEventSerializer())
+                    .register(InternalDeviceOfflineEvent.class, new InternalDeviceOfflineEventSerializer())
                     .register(InternalPortEvent.class, new InternalPortEventSerializer())
                     .register(InternalPortEvent.class, new InternalPortEventSerializer())
                     .register(InternalPortStatusEvent.class, new InternalPortStatusEventSerializer())
                     .register(InternalPortStatusEvent.class, new InternalPortStatusEventSerializer())
+                    .register(Timestamp.class)
                     .register(Timestamped.class)
                     .register(Timestamped.class)
                     .register(MastershipBasedTimestamp.class, new MastershipBasedTimestampSerializer())
                     .register(MastershipBasedTimestamp.class, new MastershipBasedTimestampSerializer())
                     .build()
                     .build()
@@ -134,6 +136,8 @@ public class GossipDeviceStore
         clusterCommunicator.addSubscriber(
         clusterCommunicator.addSubscriber(
                 GossipDeviceStoreMessageSubjects.DEVICE_UPDATE, new InternalDeviceEventListener());
                 GossipDeviceStoreMessageSubjects.DEVICE_UPDATE, new InternalDeviceEventListener());
         clusterCommunicator.addSubscriber(
         clusterCommunicator.addSubscriber(
+                GossipDeviceStoreMessageSubjects.DEVICE_OFFLINE, new InternalDeviceOfflineEventListener());
+        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());
@@ -177,7 +181,7 @@ public class GossipDeviceStore
             try {
             try {
                 notifyPeers(new InternalDeviceEvent(providerId, deviceId, deltaDesc));
                 notifyPeers(new InternalDeviceEvent(providerId, deviceId, deltaDesc));
             } catch (IOException e) {
             } catch (IOException e) {
-                log.error("Failed to notify peers of a device update topology event or providerId: "
+                log.error("Failed to notify peers of a device update topology event for providerId: "
                         + providerId + " and deviceId: " + deviceId, e);
                         + providerId + " and deviceId: " + deviceId, e);
             }
             }
         }
         }
@@ -280,7 +284,18 @@ public class GossipDeviceStore
     @Override
     @Override
     public DeviceEvent markOffline(DeviceId deviceId) {
     public DeviceEvent markOffline(DeviceId deviceId) {
         Timestamp timestamp = clockService.getTimestamp(deviceId);
         Timestamp timestamp = clockService.getTimestamp(deviceId);
-        return markOfflineInternal(deviceId, timestamp);
+        DeviceEvent event = markOfflineInternal(deviceId, timestamp);
+        if (event != null) {
+            log.info("Notifying peers of a device offline topology event for deviceId: {}",
+                    deviceId);
+            try {
+                notifyPeers(new InternalDeviceOfflineEvent(deviceId, timestamp));
+            } catch (IOException e) {
+                log.error("Failed to notify peers of a device offline topology event for deviceId: {}",
+                     deviceId);
+            }
+        }
+        return event;
     }
     }
 
 
     private DeviceEvent markOfflineInternal(DeviceId deviceId, Timestamp timestamp) {
     private DeviceEvent markOfflineInternal(DeviceId deviceId, Timestamp timestamp) {
@@ -811,6 +826,14 @@ public class GossipDeviceStore
         clusterCommunicator.broadcast(message);
         clusterCommunicator.broadcast(message);
     }
     }
 
 
+    private void notifyPeers(InternalDeviceOfflineEvent event) throws IOException {
+        ClusterMessage message = new ClusterMessage(
+                clusterService.getLocalNode().id(),
+                GossipDeviceStoreMessageSubjects.DEVICE_OFFLINE,
+                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(),
@@ -830,15 +853,32 @@ public class GossipDeviceStore
     private class InternalDeviceEventListener implements ClusterMessageHandler {
     private class InternalDeviceEventListener implements ClusterMessageHandler {
         @Override
         @Override
         public void handle(ClusterMessage message) {
         public void handle(ClusterMessage message) {
+
             log.info("Received device update event from peer: {}", message.sender());
             log.info("Received device update event from peer: {}", message.sender());
             InternalDeviceEvent event = (InternalDeviceEvent) SERIALIZER.decode(message.payload());
             InternalDeviceEvent event = (InternalDeviceEvent) SERIALIZER.decode(message.payload());
+
             ProviderId providerId = event.providerId();
             ProviderId providerId = event.providerId();
             DeviceId deviceId = event.deviceId();
             DeviceId deviceId = event.deviceId();
             Timestamped<DeviceDescription> deviceDescription = event.deviceDescription();
             Timestamped<DeviceDescription> deviceDescription = event.deviceDescription();
+
             createOrUpdateDeviceInternal(providerId, deviceId, deviceDescription);
             createOrUpdateDeviceInternal(providerId, deviceId, deviceDescription);
         }
         }
     }
     }
 
 
+    private class InternalDeviceOfflineEventListener implements ClusterMessageHandler {
+        @Override
+        public void handle(ClusterMessage message) {
+
+            log.info("Received device offline event from peer: {}", message.sender());
+            InternalDeviceOfflineEvent event = (InternalDeviceOfflineEvent) SERIALIZER.decode(message.payload());
+
+            DeviceId deviceId = event.deviceId();
+            Timestamp timestamp = event.timestamp();
+
+            markOfflineInternal(deviceId, timestamp);
+        }
+    }
+
     private class InternalPortEventListener implements ClusterMessageHandler {
     private class InternalPortEventListener implements ClusterMessageHandler {
         @Override
         @Override
         public void handle(ClusterMessage message) {
         public void handle(ClusterMessage message) {

@@ -3,13 +3,14 @@ package org.onlab.onos.store.device.impl;
 import org.onlab.onos.store.cluster.messaging.MessageSubject;
 import org.onlab.onos.store.cluster.messaging.MessageSubject;
 
 
 /**
 /**
- * MessageSubjects used by GossipDeviceStore.
+ * MessageSubjects used by GossipDeviceStore peer-peer communication.
  */
  */
 public final class GossipDeviceStoreMessageSubjects {
 public final class GossipDeviceStoreMessageSubjects {
 
 
     private GossipDeviceStoreMessageSubjects() {}
     private 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 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");
 }
 }