Merge remote-tracking branch 'origin/master'

@@ -119,8 +119,10 @@ public class GossipDeviceStore
             serializerPool = KryoPool.newBuilder()
                     .register(KryoPoolUtil.API)
                     .register(InternalDeviceEvent.class, new InternalDeviceEventSerializer())
+                    .register(InternalDeviceOfflineEvent.class, new InternalDeviceOfflineEventSerializer())
                     .register(InternalPortEvent.class, new InternalPortEventSerializer())
                     .register(InternalPortStatusEvent.class, new InternalPortStatusEventSerializer())
+                    .register(Timestamp.class)
                     .register(Timestamped.class)
                     .register(MastershipBasedTimestamp.class, new MastershipBasedTimestampSerializer())
                     .build()
@@ -134,6 +136,8 @@ public class GossipDeviceStore
         clusterCommunicator.addSubscriber(
                 GossipDeviceStoreMessageSubjects.DEVICE_UPDATE, new InternalDeviceEventListener());
         clusterCommunicator.addSubscriber(
+                GossipDeviceStoreMessageSubjects.DEVICE_OFFLINE, new InternalDeviceOfflineEventListener());
+        clusterCommunicator.addSubscriber(
                 GossipDeviceStoreMessageSubjects.PORT_UPDATE, new InternalPortEventListener());
         clusterCommunicator.addSubscriber(
                 GossipDeviceStoreMessageSubjects.PORT_STATUS_UPDATE, new InternalPortStatusEventListener());
@@ -177,7 +181,7 @@ public class GossipDeviceStore
             try {
                 notifyPeers(new InternalDeviceEvent(providerId, deviceId, deltaDesc));
             } 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);
             }
         }
@@ -280,7 +284,18 @@ public class GossipDeviceStore
     @Override
     public DeviceEvent markOffline(DeviceId 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) {
@@ -811,6 +826,14 @@ public class GossipDeviceStore
         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 {
         ClusterMessage message = new ClusterMessage(
                 clusterService.getLocalNode().id(),
@@ -830,15 +853,32 @@ public class GossipDeviceStore
     private class InternalDeviceEventListener implements ClusterMessageHandler {
         @Override
         public void handle(ClusterMessage message) {
+
             log.info("Received device update event from peer: {}", message.sender());
             InternalDeviceEvent event = (InternalDeviceEvent) SERIALIZER.decode(message.payload());
+
             ProviderId providerId = event.providerId();
             DeviceId deviceId = event.deviceId();
             Timestamped<DeviceDescription> deviceDescription = event.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 {
         @Override
         public void handle(ClusterMessage message) {

@@ -3,13 +3,14 @@ package org.onlab.onos.store.device.impl;
 import org.onlab.onos.store.cluster.messaging.MessageSubject;
 
 /**
- * MessageSubjects used by GossipDeviceStore.
+ * MessageSubjects used by GossipDeviceStore peer-peer communication.
  */
 public final class GossipDeviceStoreMessageSubjects {
 
     private GossipDeviceStoreMessageSubjects() {}
 
     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_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
+ * going offline.
+ */
+public class InternalDeviceOfflineEvent {
+
+    private final DeviceId deviceId;
+    private final Timestamp timestamp;
+
+    /**
+     * Creates a InternalDeviceOfflineEvent.
+     * @param deviceId identifier of device going offline.
+     * @param timestamp timestamp of when the device went offline.
+     */
+    public InternalDeviceOfflineEvent(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 InternalDeviceOfflineEvent() {
+        deviceId = null;
+        timestamp = null;
+    }
+}

+package org.onlab.onos.store.device.impl;
+
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.store.Timestamp;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+/**
+ * Kryo Serializer for {@link InternalDeviceOfflineEvent}.
+ */
+public class InternalDeviceOfflineEventSerializer extends Serializer<InternalDeviceOfflineEvent> {
+
+    /**
+     * Creates a serializer for {@link InternalDeviceOfflineEvent}.
+     */
+    public InternalDeviceOfflineEventSerializer() {
+        // does not accept null
+        super(false);
+    }
+
+    @Override
+    public void write(Kryo kryo, Output output, InternalDeviceOfflineEvent event) {
+        kryo.writeClassAndObject(output, event.deviceId());
+        kryo.writeClassAndObject(output, event.timestamp());
+    }
+
+    @Override
+    public InternalDeviceOfflineEvent read(Kryo kryo, Input input,
+                               Class<InternalDeviceOfflineEvent> type) {
+        DeviceId deviceId = (DeviceId) kryo.readClassAndObject(input);
+        Timestamp timestamp = (Timestamp) kryo.readClassAndObject(input);
+
+        return new InternalDeviceOfflineEvent(deviceId, timestamp);
+    }
+}