Distributed Link store implementation that takes event versioning into consideration

@@ -86,46 +86,48 @@ public class OnosDistributedDeviceStore
 
     @Override
     public Iterable<Device> getDevices() {
-        // TODO builder v.s. copyOf. Guava semms to be using copyOf?
-        // FIXME: synchronize.
         Builder<Device> builder = ImmutableSet.builder();
-        for (VersionedValue<? extends Device> device : devices.values()) {
-            builder.add(device.entity());
+        synchronized (this) {
+            for (VersionedValue<Device> device : devices.values()) {
+                builder.add(device.entity());
+            }
+            return builder.build();
         }
-        return builder.build();
     }
 
     @Override
     public Device getDevice(DeviceId deviceId) {
-        return devices.get(deviceId).entity();
+        VersionedValue<Device> device = devices.get(deviceId);
+        checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
+        return device.entity();
     }
 
     @Override
     public DeviceEvent createOrUpdateDevice(ProviderId providerId, DeviceId deviceId,
                                             DeviceDescription deviceDescription) {
-        Timestamp now = clockService.getTimestamp(deviceId);
+        Timestamp newTimestamp = clockService.getTimestamp(deviceId);
         VersionedValue<Device> device = devices.get(deviceId);
 
         if (device == null) {
-            return createDevice(providerId, deviceId, deviceDescription, now);
+            return createDevice(providerId, deviceId, deviceDescription, newTimestamp);
         }
 
-        checkState(now.compareTo(device.timestamp()) > 0,
+        checkState(newTimestamp.compareTo(device.timestamp()) > 0,
                 "Existing device has a timestamp in the future!");
 
-        return updateDevice(providerId, device.entity(), deviceDescription, now);
+        return updateDevice(providerId, device.entity(), deviceDescription, newTimestamp);
     }
 
     // Creates the device and returns the appropriate event if necessary.
     private DeviceEvent createDevice(ProviderId providerId, DeviceId deviceId,
                                      DeviceDescription desc, Timestamp timestamp) {
-        DefaultDevice device = new DefaultDevice(providerId, deviceId, desc.type(),
+        Device device = new DefaultDevice(providerId, deviceId, desc.type(),
                                                  desc.manufacturer(),
                                                  desc.hwVersion(), desc.swVersion(),
                                                  desc.serialNumber());
 
-        devices.put(deviceId, new VersionedValue<Device>(device, true, timestamp));
-        // FIXME: broadcast a message telling peers of a device event.
+        devices.put(deviceId, new VersionedValue<>(device, true, timestamp));
+        // TODO,FIXME: broadcast a message telling peers of a device event.
         return new DeviceEvent(DEVICE_ADDED, device, null);
     }
 
@@ -148,7 +150,7 @@ public class OnosDistributedDeviceStore
         }
 
         // Otherwise merely attempt to change availability
-        DefaultDevice updated = new DefaultDevice(providerId, device.id(),
+        Device updated = new DefaultDevice(providerId, device.id(),
                 desc.type(),
                 desc.manufacturer(),
                 desc.hwVersion(),
@@ -196,18 +198,18 @@ public class OnosDistributedDeviceStore
             VersionedValue<Device> device = devices.get(deviceId);
             checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
             Map<PortNumber, VersionedValue<Port>> ports = getPortMap(deviceId);
-            Timestamp timestamp = clockService.getTimestamp(deviceId);
+            Timestamp newTimestamp = clockService.getTimestamp(deviceId);
 
             // Add new ports
             Set<PortNumber> processed = new HashSet<>();
             for (PortDescription portDescription : portDescriptions) {
                 VersionedValue<Port> port = ports.get(portDescription.portNumber());
                 if (port == null) {
-                    events.add(createPort(device, portDescription, ports, timestamp));
+                    events.add(createPort(device, portDescription, ports, newTimestamp));
                 }
-                checkState(timestamp.compareTo(port.timestamp()) > 0,
+                checkState(newTimestamp.compareTo(port.timestamp()) > 0,
                         "Existing port state has a timestamp in the future!");
-                events.add(updatePort(device, port, portDescription, ports, timestamp));
+                events.add(updatePort(device.entity(), port.entity(), portDescription, ports, newTimestamp));
                 processed.add(portDescription.portNumber());
             }
 
@@ -233,19 +235,19 @@ public class OnosDistributedDeviceStore
     // Checks if the specified port requires update and if so, it replaces the
     // existing entry in the map and returns corresponding event.
     //@GuardedBy("this")
-    private DeviceEvent updatePort(VersionedValue<Device> device, VersionedValue<Port> port,
+    private DeviceEvent updatePort(Device device, Port port,
                                    PortDescription portDescription,
                                    Map<PortNumber, VersionedValue<Port>> ports,
                                    Timestamp timestamp) {
-        if (port.entity().isEnabled() != portDescription.isEnabled()) {
+        if (port.isEnabled() != portDescription.isEnabled()) {
             VersionedValue<Port> updatedPort = new VersionedValue<Port>(
-                    new DefaultPort(device.entity(), portDescription.portNumber(),
+                    new DefaultPort(device, portDescription.portNumber(),
                                     portDescription.isEnabled()),
                     portDescription.isEnabled(),
                     timestamp);
-            ports.put(port.entity().number(), updatedPort);
-            updatePortMap(device.entity().id(), ports);
-            return new DeviceEvent(PORT_UPDATED, device.entity(), updatedPort.entity());
+            ports.put(port.number(), updatedPort);
+            updatePortMap(device.id(), ports);
+            return new DeviceEvent(PORT_UPDATED, device, updatedPort.entity());
         }
         return null;
     }
@@ -300,7 +302,7 @@ public class OnosDistributedDeviceStore
         Map<PortNumber, VersionedValue<Port>> ports = getPortMap(deviceId);
         VersionedValue<Port> port = ports.get(portDescription.portNumber());
         Timestamp timestamp = clockService.getTimestamp(deviceId);
-        return updatePort(device, port, portDescription, ports, timestamp);
+        return updatePort(device.entity(), port.entity(), portDescription, ports, timestamp);
     }
 
     @Override

+package org.onlab.onos.store.link.impl;
+
+import static org.onlab.onos.net.Link.Type.DIRECT;
+import static org.onlab.onos.net.Link.Type.INDIRECT;
+import static org.onlab.onos.net.link.LinkEvent.Type.LINK_ADDED;
+import static org.onlab.onos.net.link.LinkEvent.Type.LINK_REMOVED;
+import static org.onlab.onos.net.link.LinkEvent.Type.LINK_UPDATED;
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.onos.net.ConnectPoint;
+import org.onlab.onos.net.DefaultLink;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.Link;
+import org.onlab.onos.net.LinkKey;
+import org.onlab.onos.net.link.LinkDescription;
+import org.onlab.onos.net.link.LinkEvent;
+import org.onlab.onos.net.link.LinkStore;
+import org.onlab.onos.net.link.LinkStoreDelegate;
+import org.onlab.onos.net.provider.ProviderId;
+import org.onlab.onos.store.AbstractStore;
+import org.onlab.onos.store.ClockService;
+import org.onlab.onos.store.Timestamp;
+import org.onlab.onos.store.device.impl.VersionedValue;
+import org.slf4j.Logger;
+
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Multimap;
+import com.google.common.collect.ImmutableSet.Builder;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+
+/**
+ * Manages inventory of infrastructure links using a protocol that takes into consideration
+ * the order in which events occur.
+ */
+// FIXME: This does not yet implement the full protocol.
+// The full protocol requires the sender of LLDP message to include the
+// version information of src device/port and the receiver to
+// take that into account when figuring out if a more recent src
+// device/port down event renders the link discovery obsolete.
+@Component(immediate = true)
+@Service
+public class OnosDistributedLinkStore
+    extends AbstractStore<LinkEvent, LinkStoreDelegate>
+    implements LinkStore {
+
+    private final Logger log = getLogger(getClass());
+
+    // Link inventory
+    private ConcurrentMap<LinkKey, VersionedValue<Link>> links;
+
+    public static final String LINK_NOT_FOUND = "Link between %s and %s not found";
+
+    // TODO synchronize?
+    // Egress and ingress link sets
+    private final Multimap<DeviceId, VersionedValue<Link>> srcLinks = HashMultimap.create();
+    private final Multimap<DeviceId, VersionedValue<Link>> dstLinks = HashMultimap.create();
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected ClockService clockService;
+
+    @Activate
+    public void activate() {
+
+        links = new ConcurrentHashMap<>();
+
+        log.info("Started");
+    }
+
+    @Deactivate
+    public void deactivate() {
+        log.info("Stopped");
+    }
+
+    @Override
+    public int getLinkCount() {
+        return links.size();
+    }
+
+    @Override
+    public Iterable<Link> getLinks() {
+        Builder<Link> builder = ImmutableSet.builder();
+        synchronized (this) {
+            for (VersionedValue<Link> link : links.values()) {
+                builder.add(link.entity());
+            }
+            return builder.build();
+        }
+    }
+
+    @Override
+    public Set<Link> getDeviceEgressLinks(DeviceId deviceId) {
+        Set<VersionedValue<Link>> egressLinks = ImmutableSet.copyOf(srcLinks.get(deviceId));
+        Set<Link> rawEgressLinks = new HashSet<>();
+        for (VersionedValue<Link> link : egressLinks) {
+            rawEgressLinks.add(link.entity());
+        }
+        return rawEgressLinks;
+    }
+
+    @Override
+    public Set<Link> getDeviceIngressLinks(DeviceId deviceId) {
+        Set<VersionedValue<Link>> ingressLinks = ImmutableSet.copyOf(dstLinks.get(deviceId));
+        Set<Link> rawIngressLinks = new HashSet<>();
+        for (VersionedValue<Link> link : ingressLinks) {
+            rawIngressLinks.add(link.entity());
+        }
+        return rawIngressLinks;
+    }
+
+    @Override
+    public Link getLink(ConnectPoint src, ConnectPoint dst) {
+        VersionedValue<Link> link = links.get(new LinkKey(src, dst));
+        checkArgument(link != null, "LINK_NOT_FOUND", src, dst);
+        return link.entity();
+    }
+
+    @Override
+    public Set<Link> getEgressLinks(ConnectPoint src) {
+        Set<Link> egressLinks = new HashSet<>();
+        for (VersionedValue<Link> link : srcLinks.get(src.deviceId())) {
+            if (link.entity().src().equals(src)) {
+                egressLinks.add(link.entity());
+            }
+        }
+        return egressLinks;
+    }
+
+    @Override
+    public Set<Link> getIngressLinks(ConnectPoint dst) {
+        Set<Link> ingressLinks = new HashSet<>();
+        for (VersionedValue<Link> link : dstLinks.get(dst.deviceId())) {
+            if (link.entity().dst().equals(dst)) {
+                ingressLinks.add(link.entity());
+            }
+        }
+        return ingressLinks;
+    }
+
+    @Override
+    public LinkEvent createOrUpdateLink(ProviderId providerId,
+                                        LinkDescription linkDescription) {
+
+        final DeviceId destinationDeviceId = linkDescription.dst().deviceId();
+        final Timestamp newTimestamp = clockService.getTimestamp(destinationDeviceId);
+
+        LinkKey key = new LinkKey(linkDescription.src(), linkDescription.dst());
+        VersionedValue<Link> link = links.get(key);
+        if (link == null) {
+            return createLink(providerId, key, linkDescription, newTimestamp);
+        }
+
+        checkState(newTimestamp.compareTo(link.timestamp()) > 0,
+                "Existing Link has a timestamp in the future!");
+
+        return updateLink(providerId, link, key, linkDescription, newTimestamp);
+    }
+
+    // Creates and stores the link and returns the appropriate event.
+    private LinkEvent createLink(ProviderId providerId, LinkKey key,
+            LinkDescription linkDescription, Timestamp timestamp) {
+        VersionedValue<Link> link = new VersionedValue<Link>(new DefaultLink(providerId, key.src(), key.dst(),
+                linkDescription.type()), true, timestamp);
+        synchronized (this) {
+            links.put(key, link);
+            addNewLink(link, timestamp);
+        }
+        // FIXME: notify peers.
+        return new LinkEvent(LINK_ADDED, link.entity());
+    }
+
+    // update Egress and ingress link sets
+    private void addNewLink(VersionedValue<Link> link, Timestamp timestamp) {
+        Link rawLink = link.entity();
+        synchronized (this) {
+            srcLinks.put(rawLink.src().deviceId(), link);
+            dstLinks.put(rawLink.dst().deviceId(), link);
+        }
+    }
+
+    // Updates, if necessary the specified link and returns the appropriate event.
+    private LinkEvent updateLink(ProviderId providerId, VersionedValue<Link> existingLink,
+                                 LinkKey key, LinkDescription linkDescription, Timestamp timestamp) {
+        // FIXME confirm Link update condition is OK
+        if (existingLink.entity().type() == INDIRECT && linkDescription.type() == DIRECT) {
+            synchronized (this) {
+
+                VersionedValue<Link> updatedLink = new VersionedValue<Link>(
+                        new DefaultLink(providerId, existingLink.entity().src(), existingLink.entity().dst(),
+                                        linkDescription.type()), true, timestamp);
+                links.replace(key, existingLink, updatedLink);
+
+                replaceLink(existingLink, updatedLink);
+                // FIXME: notify peers.
+                return new LinkEvent(LINK_UPDATED, updatedLink.entity());
+            }
+        }
+        return null;
+    }
+
+    // update Egress and ingress link sets
+    private void replaceLink(VersionedValue<Link> current, VersionedValue<Link> updated) {
+        synchronized (this) {
+            srcLinks.remove(current.entity().src().deviceId(), current);
+            dstLinks.remove(current.entity().dst().deviceId(), current);
+
+            srcLinks.put(current.entity().src().deviceId(), updated);
+            dstLinks.put(current.entity().dst().deviceId(), updated);
+        }
+    }
+
+    @Override
+    public LinkEvent removeLink(ConnectPoint src, ConnectPoint dst) {
+        synchronized (this) {
+            LinkKey key = new LinkKey(src, dst);
+            VersionedValue<Link> link = links.remove(key);
+            if (link != null) {
+                removeLink(link);
+                // notify peers
+                return new LinkEvent(LINK_REMOVED, link.entity());
+            }
+            return null;
+        }
+    }
+
+    // update Egress and ingress link sets
+    private void removeLink(VersionedValue<Link> link) {
+        synchronized (this) {
+            srcLinks.remove(link.entity().src().deviceId(), link);
+            dstLinks.remove(link.entity().dst().deviceId(), link);
+        }
+    }
+}