Merge branch 'master' of ssh://gerrit.onlab.us:29418/onos-next
diff --git a/apps/ifwd/src/main/java/org/onlab/onos/ifwd/IntentReactiveForwarding.java b/apps/ifwd/src/main/java/org/onlab/onos/ifwd/IntentReactiveForwarding.java
index 15c7bc3..fded98f 100644
--- a/apps/ifwd/src/main/java/org/onlab/onos/ifwd/IntentReactiveForwarding.java
+++ b/apps/ifwd/src/main/java/org/onlab/onos/ifwd/IntentReactiveForwarding.java
@@ -1,5 +1,7 @@
package org.onlab.onos.ifwd;
+import static org.slf4j.LoggerFactory.getLogger;
+
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
@@ -26,8 +28,6 @@
import org.onlab.packet.Ethernet;
import org.slf4j.Logger;
-import static org.slf4j.LoggerFactory.getLogger;
-
/**
* WORK-IN-PROGRESS: Sample reactive forwarding application using intent framework.
*/
@@ -50,7 +50,7 @@
private ReactivePacketProcessor processor = new ReactivePacketProcessor();
- private static long intentId = 1;
+ private static long intentId = 0x123000;
@Activate
public void activate() {
diff --git a/core/net/pom.xml b/core/net/pom.xml
index c075147..6518068 100644
--- a/core/net/pom.xml
+++ b/core/net/pom.xml
@@ -36,6 +36,12 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.easymock</groupId>
+ <artifactId>easymock</artifactId>
+ <scope>test</scope>
+ </dependency>
+
<!-- TODO Consider removing store dependency.
Currently required for DistributedDeviceManagerTest. -->
<dependency>
diff --git a/core/net/src/main/java/org/onlab/onos/net/host/impl/HostMonitor.java b/core/net/src/main/java/org/onlab/onos/net/host/impl/HostMonitor.java
index 6795cad..e6e348f 100644
--- a/core/net/src/main/java/org/onlab/onos/net/host/impl/HostMonitor.java
+++ b/core/net/src/main/java/org/onlab/onos/net/host/impl/HostMonitor.java
@@ -60,14 +60,15 @@
*
* @param deviceService device service used to find edge ports
* @param packetService packet service used to send packets on the data plane
- * @param hostService host service used to look up host information
+ * @param hostManager host manager used to look up host information and
+ * probe existing hosts
*/
public HostMonitor(DeviceService deviceService, PacketService packetService,
- HostManager hostService) {
+ HostManager hostManager) {
this.deviceService = deviceService;
this.packetService = packetService;
- this.hostManager = hostService;
+ this.hostManager = hostManager;
monitoredAddresses = Collections.newSetFromMap(
new ConcurrentHashMap<IpAddress, Boolean>());
diff --git a/core/net/src/test/java/org/onlab/onos/net/host/impl/HostMonitorTest.java b/core/net/src/test/java/org/onlab/onos/net/host/impl/HostMonitorTest.java
new file mode 100644
index 0000000..d766251
--- /dev/null
+++ b/core/net/src/test/java/org/onlab/onos/net/host/impl/HostMonitorTest.java
@@ -0,0 +1,220 @@
+package org.onlab.onos.net.host.impl;
+
+import static org.easymock.EasyMock.createMock;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.expectLastCall;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.verify;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+import org.junit.Test;
+import org.onlab.onos.net.ConnectPoint;
+import org.onlab.onos.net.Device;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.Host;
+import org.onlab.onos.net.MastershipRole;
+import org.onlab.onos.net.Port;
+import org.onlab.onos.net.PortNumber;
+import org.onlab.onos.net.device.DeviceListener;
+import org.onlab.onos.net.device.DeviceService;
+import org.onlab.onos.net.flow.instructions.Instruction;
+import org.onlab.onos.net.flow.instructions.Instructions.OutputInstruction;
+import org.onlab.onos.net.host.HostProvider;
+import org.onlab.onos.net.host.PortAddresses;
+import org.onlab.onos.net.packet.OutboundPacket;
+import org.onlab.onos.net.packet.PacketProcessor;
+import org.onlab.onos.net.packet.PacketService;
+import org.onlab.onos.net.provider.ProviderId;
+import org.onlab.packet.ARP;
+import org.onlab.packet.Ethernet;
+import org.onlab.packet.IpAddress;
+import org.onlab.packet.IpPrefix;
+import org.onlab.packet.MacAddress;
+
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Multimap;
+
+public class HostMonitorTest {
+
+ private IpAddress targetIpAddress = IpAddress.valueOf("10.0.0.1");
+ private IpPrefix targetIpPrefix = IpPrefix.valueOf(targetIpAddress.toOctets());
+
+ private IpPrefix sourcePrefix = IpPrefix.valueOf("10.0.0.99/24");
+ private MacAddress sourceMac = MacAddress.valueOf(1L);
+
+ private HostMonitor hostMonitor;
+
+ @Test
+ public void testMonitorHostExists() throws Exception {
+ ProviderId id = new ProviderId("fake://", "id");
+
+ Host host = createMock(Host.class);
+ expect(host.providerId()).andReturn(id);
+ replay(host);
+
+ HostManager hostManager = createMock(HostManager.class);
+ expect(hostManager.getHostsByIp(targetIpPrefix))
+ .andReturn(Collections.singleton(host));
+ replay(hostManager);
+
+ HostProvider hostProvider = createMock(HostProvider.class);
+ expect(hostProvider.id()).andReturn(id).anyTimes();
+ hostProvider.triggerProbe(host);
+ expectLastCall().once();
+ replay(hostProvider);
+
+ hostMonitor = new HostMonitor(null, null, hostManager);
+
+ hostMonitor.registerHostProvider(hostProvider);
+ hostMonitor.addMonitoringFor(targetIpAddress);
+
+ hostMonitor.run(null);
+
+ verify(hostProvider);
+ }
+
+ @Test
+ public void testMonitorHostDoesNotExist() throws Exception {
+ HostManager hostManager = createMock(HostManager.class);
+
+ DeviceId devId = DeviceId.deviceId("fake");
+
+ Device device = createMock(Device.class);
+ expect(device.id()).andReturn(devId).anyTimes();
+ replay(device);
+
+ PortNumber portNum = PortNumber.portNumber(1L);
+
+ Port port = createMock(Port.class);
+ expect(port.number()).andReturn(portNum).anyTimes();
+ replay(port);
+
+ TestDeviceService deviceService = new TestDeviceService();
+ deviceService.addDevice(device, Collections.singleton(port));
+
+ ConnectPoint cp = new ConnectPoint(devId, portNum);
+ PortAddresses pa = new PortAddresses(cp, Collections.singleton(sourcePrefix),
+ sourceMac);
+
+ expect(hostManager.getHostsByIp(targetIpPrefix))
+ .andReturn(Collections.<Host>emptySet()).anyTimes();
+ expect(hostManager.getAddressBindingsForPort(cp))
+ .andReturn(pa).anyTimes();
+ replay(hostManager);
+
+ TestPacketService packetService = new TestPacketService();
+
+
+ // Run the test
+ hostMonitor = new HostMonitor(deviceService, packetService, hostManager);
+
+ hostMonitor.addMonitoringFor(targetIpAddress);
+ hostMonitor.run(null);
+
+
+ // Check that a packet was sent to our PacketService and that it has
+ // the properties we expect
+ assertTrue(packetService.packets.size() == 1);
+ OutboundPacket packet = packetService.packets.get(0);
+
+ // Check the output port is correct
+ assertTrue(packet.treatment().instructions().size() == 1);
+ Instruction instruction = packet.treatment().instructions().get(0);
+ assertTrue(instruction instanceof OutputInstruction);
+ OutputInstruction oi = (OutputInstruction) instruction;
+ assertTrue(oi.port().equals(portNum));
+
+ // Check the output packet is correct (well the important bits anyway)
+ Ethernet eth = new Ethernet();
+ eth.deserialize(packet.data().array(), 0, packet.data().array().length);
+ ARP arp = (ARP) eth.getPayload();
+ assertTrue(Arrays.equals(arp.getSenderProtocolAddress(), sourcePrefix.toOctets()));
+ assertTrue(Arrays.equals(arp.getSenderHardwareAddress(), sourceMac.toBytes()));
+ assertTrue(Arrays.equals(arp.getTargetProtocolAddress(), targetIpPrefix.toOctets()));
+ }
+
+ class TestPacketService implements PacketService {
+
+ List<OutboundPacket> packets = new ArrayList<>();
+
+ @Override
+ public void addProcessor(PacketProcessor processor, int priority) {
+ }
+
+ @Override
+ public void removeProcessor(PacketProcessor processor) {
+ }
+
+ @Override
+ public void emit(OutboundPacket packet) {
+ packets.add(packet);
+ }
+ }
+
+ class TestDeviceService implements DeviceService {
+
+ List<Device> devices = Lists.newArrayList();
+ Multimap<DeviceId, Port> devicePorts = HashMultimap.create();
+
+ void addDevice(Device device, Set<Port> ports) {
+ devices.add(device);
+ for (Port p : ports) {
+ devicePorts.put(device.id(), p);
+ }
+ }
+
+ @Override
+ public int getDeviceCount() {
+ return 0;
+ }
+
+ @Override
+ public Iterable<Device> getDevices() {
+ return devices;
+ }
+
+ @Override
+ public Device getDevice(DeviceId deviceId) {
+ return null;
+ }
+
+ @Override
+ public MastershipRole getRole(DeviceId deviceId) {
+ return null;
+ }
+
+ @Override
+ public List<Port> getPorts(DeviceId deviceId) {
+ List<Port> ports = Lists.newArrayList();
+ for (Port p : devicePorts.get(deviceId)) {
+ ports.add(p);
+ }
+ return ports;
+ }
+
+ @Override
+ public Port getPort(DeviceId deviceId, PortNumber portNumber) {
+ return null;
+ }
+
+ @Override
+ public boolean isAvailable(DeviceId deviceId) {
+ return false;
+ }
+
+ @Override
+ public void addListener(DeviceListener listener) {
+ }
+
+ @Override
+ public void removeListener(DeviceListener listener) {
+ }
+ }
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterCommunicationAdminService.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterCommunicationAdminService.java
index 5966f12..0bc31fa 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterCommunicationAdminService.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterCommunicationAdminService.java
@@ -3,6 +3,8 @@
import org.onlab.onos.cluster.ControllerNode;
import org.onlab.onos.store.cluster.impl.ClusterNodesDelegate;
+// TODO: This service interface can be removed, once we properly start
+// using ClusterService
/**
* Service for administering communications manager.
*/
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterMessage.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterMessage.java
index ee558dd..5d04a46 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterMessage.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterMessage.java
@@ -2,6 +2,8 @@
import org.onlab.onos.cluster.NodeId;
+// TODO: ClusterMessage should be aware about how to serialize the payload
+// TODO: Should payload type be made generic?
/**
* Base message for cluster-wide communications.
*/
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterMessageHandler.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterMessageHandler.java
index 7ec27ec..4dd7bc2 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterMessageHandler.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/ClusterMessageHandler.java
@@ -10,4 +10,4 @@
* @param message cluster message.
*/
public void handle(ClusterMessage message);
-}
\ No newline at end of file
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/MessageSubject.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/MessageSubject.java
index ee8d9c1..ee3e789 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/MessageSubject.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/MessageSubject.java
@@ -1,5 +1,9 @@
package org.onlab.onos.store.cluster.messaging;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.Objects;
+
/**
* Representation of a message subject.
* Cluster messages have associated subjects that dictate how they get handled
@@ -10,7 +14,7 @@
private final String value;
public MessageSubject(String value) {
- this.value = value;
+ this.value = checkNotNull(value);
}
public String value() {
@@ -21,4 +25,24 @@
public String toString() {
return value;
}
+
+ @Override
+ public int hashCode() {
+ return value.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null) {
+ return false;
+ }
+ if (getClass() != obj.getClass()) {
+ return false;
+ }
+ MessageSubject that = (MessageSubject) obj;
+ return Objects.equals(this.value, that.value);
+ }
}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/ClusterCommunicationManager.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/ClusterCommunicationManager.java
index d4fd9c0..2e8937c 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/ClusterCommunicationManager.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/ClusterCommunicationManager.java
@@ -39,6 +39,7 @@
private ControllerNode localNode;
private ClusterNodesDelegate nodesDelegate;
+ // FIXME: `members` should go away and should be using ClusterService
private Map<NodeId, ControllerNode> members = new HashMap<>();
private final Timer timer = new Timer("onos-controller-heatbeats");
public static final long HEART_BEAT_INTERVAL_MILLIS = 1000L;
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/ClusterMessageSubjects.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/ClusterMessageSubjects.java
index d1f75ae..11f6228 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/ClusterMessageSubjects.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/messaging/impl/ClusterMessageSubjects.java
@@ -3,6 +3,8 @@
import org.onlab.onos.store.cluster.messaging.MessageSubject;
public final class ClusterMessageSubjects {
+ // avoid instantiation
private ClusterMessageSubjects() {}
+
public static final MessageSubject CLUSTER_MEMBERSHIP_EVENT = new MessageSubject("CLUSTER_MEMBERSHIP_EVENT");
}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/common/impl/Timestamped.java b/core/store/dist/src/main/java/org/onlab/onos/store/common/impl/Timestamped.java
index 2908b16..0598d6d 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/common/impl/Timestamped.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/common/impl/Timestamped.java
@@ -6,6 +6,8 @@
import org.onlab.onos.store.Timestamp;
+import com.google.common.base.MoreObjects;
+
/**
* Wrapper class to store Timestamped value.
* @param <T>
@@ -70,6 +72,14 @@
return Objects.equals(this.timestamp, that.timestamp);
}
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(getClass())
+ .add("timestamp", timestamp)
+ .add("value", value)
+ .toString();
+ }
+
// Default constructor for serialization
@Deprecated
protected Timestamped() {
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/device/impl/GossipDeviceStore.java b/core/store/dist/src/main/java/org/onlab/onos/store/device/impl/GossipDeviceStore.java
index 85d9b07..8316769 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/device/impl/GossipDeviceStore.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/device/impl/GossipDeviceStore.java
@@ -2,7 +2,8 @@
import com.google.common.collect.FluentIterable;
import com.google.common.collect.ImmutableList;
-
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
import org.apache.commons.lang3.concurrent.ConcurrentException;
import org.apache.commons.lang3.concurrent.ConcurrentInitializer;
import org.apache.felix.scr.annotations.Activate;
@@ -59,7 +60,7 @@
import static org.onlab.onos.net.DefaultAnnotations.union;
import static com.google.common.base.Verify.verify;
-// TODO: implement remove event handling and call *Internal
+// TODO: give me a better name
/**
* Manages inventory of infrastructure devices using gossip protocol to distribute
* information.
@@ -79,14 +80,18 @@
// collection of Description given from various providers
private final ConcurrentMap<DeviceId,
ConcurrentMap<ProviderId, DeviceDescriptions>>
- deviceDescs = new ConcurrentHashMap<>();
+ deviceDescs = Maps.newConcurrentMap();
// cache of Device and Ports generated by compositing descriptions from providers
- private final ConcurrentMap<DeviceId, Device> devices = new ConcurrentHashMap<>();
- private final ConcurrentMap<DeviceId, ConcurrentMap<PortNumber, Port>> devicePorts = new ConcurrentHashMap<>();
+ private final ConcurrentMap<DeviceId, Device> devices = Maps.newConcurrentMap();
+ private final ConcurrentMap<DeviceId, ConcurrentMap<PortNumber, Port>> devicePorts = Maps.newConcurrentMap();
+
+ // to be updated under Device lock
+ private final Map<DeviceId, Timestamp> offline = Maps.newHashMap();
+ private final Map<DeviceId, Timestamp> removalRequest = Maps.newHashMap();
// available(=UP) devices
- private final Set<DeviceId> availableDevices = new HashSet<>();
+ private final Set<DeviceId> availableDevices = Sets.newConcurrentHashSet();
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected ClockService clockService;
@@ -121,7 +126,8 @@
}
@Override
- public synchronized DeviceEvent createOrUpdateDevice(ProviderId providerId, DeviceId deviceId,
+ public synchronized DeviceEvent createOrUpdateDevice(ProviderId providerId,
+ DeviceId deviceId,
DeviceDescription deviceDescription) {
Timestamp newTimestamp = clockService.getTimestamp(deviceId);
final Timestamped<DeviceDescription> deltaDesc = new Timestamped<>(deviceDescription, newTimestamp);
@@ -133,22 +139,26 @@
return event;
}
- private DeviceEvent createOrUpdateDeviceInternal(ProviderId providerId, DeviceId deviceId,
- Timestamped<DeviceDescription> deltaDesc) {
+ private DeviceEvent createOrUpdateDeviceInternal(ProviderId providerId,
+ DeviceId deviceId,
+ Timestamped<DeviceDescription> deltaDesc) {
// Collection of DeviceDescriptions for a Device
ConcurrentMap<ProviderId, DeviceDescriptions> providerDescs
= getDeviceDescriptions(deviceId);
-
- DeviceDescriptions descs
- = createIfAbsentUnchecked(providerDescs, providerId,
- new InitDeviceDescs(deltaDesc));
-
- // update description
synchronized (providerDescs) {
// locking per device
+ if (isDeviceRemoved(deviceId, deltaDesc.timestamp())) {
+ log.debug("Ignoring outdated event: {}", deltaDesc);
+ return null;
+ }
+
+ DeviceDescriptions descs
+ = createIfAbsentUnchecked(providerDescs, providerId,
+ new InitDeviceDescs(deltaDesc));
+
final Device oldDevice = devices.get(deviceId);
final Device newDevice;
@@ -163,18 +173,18 @@
}
if (oldDevice == null) {
// ADD
- return createDevice(providerId, newDevice);
+ return createDevice(providerId, newDevice, deltaDesc.timestamp());
} else {
// UPDATE or ignore (no change or stale)
- return updateDevice(providerId, oldDevice, newDevice);
+ return updateDevice(providerId, oldDevice, newDevice, deltaDesc.timestamp());
}
}
}
// Creates the device and returns the appropriate event if necessary.
- // Guarded by deviceDescs value (=locking Device)
+ // Guarded by deviceDescs value (=Device lock)
private DeviceEvent createDevice(ProviderId providerId,
- Device newDevice) {
+ Device newDevice, Timestamp timestamp) {
// update composed device cache
Device oldDevice = devices.putIfAbsent(newDevice.id(), newDevice);
@@ -183,16 +193,17 @@
providerId, oldDevice, newDevice);
if (!providerId.isAncillary()) {
- availableDevices.add(newDevice.id());
+ markOnline(newDevice.id(), timestamp);
}
return new DeviceEvent(DeviceEvent.Type.DEVICE_ADDED, newDevice, null);
}
// Updates the device and returns the appropriate event if necessary.
- // Guarded by deviceDescs value (=locking Device)
+ // Guarded by deviceDescs value (=Device lock)
private DeviceEvent updateDevice(ProviderId providerId,
- Device oldDevice, Device newDevice) {
+ Device oldDevice,
+ Device newDevice, Timestamp newTimestamp) {
// We allow only certain attributes to trigger update
if (!Objects.equals(oldDevice.hwVersion(), newDevice.hwVersion()) ||
@@ -207,14 +218,14 @@
, newDevice);
}
if (!providerId.isAncillary()) {
- availableDevices.add(newDevice.id());
+ markOnline(newDevice.id(), newTimestamp);
}
return new DeviceEvent(DeviceEvent.Type.DEVICE_UPDATED, newDevice, null);
}
// Otherwise merely attempt to change availability if primary provider
if (!providerId.isAncillary()) {
- boolean added = availableDevices.add(newDevice.id());
+ boolean added = markOnline(newDevice.id(), newTimestamp);
return !added ? null :
new DeviceEvent(DEVICE_AVAILABILITY_CHANGED, newDevice, null);
}
@@ -223,11 +234,29 @@
@Override
public DeviceEvent markOffline(DeviceId deviceId) {
- ConcurrentMap<ProviderId, DeviceDescriptions> providerDescs
+ Timestamp timestamp = clockService.getTimestamp(deviceId);
+ return markOfflineInternal(deviceId, timestamp);
+ }
+
+ private DeviceEvent markOfflineInternal(DeviceId deviceId, Timestamp timestamp) {
+
+ Map<ProviderId, DeviceDescriptions> providerDescs
= getDeviceDescriptions(deviceId);
// locking device
synchronized (providerDescs) {
+
+ // accept off-line if given timestamp is newer than
+ // the latest Timestamp from Primary provider
+ DeviceDescriptions primDescs = getPrimaryDescriptions(providerDescs);
+ Timestamp lastTimestamp = primDescs.getLatestTimestamp();
+ if (timestamp.compareTo(lastTimestamp) <= 0) {
+ // outdated event ignore
+ return null;
+ }
+
+ offline.put(deviceId, timestamp);
+
Device device = devices.get(deviceId);
if (device == null) {
return null;
@@ -236,15 +265,37 @@
if (removed) {
// TODO: broadcast ... DOWN only?
return new DeviceEvent(DEVICE_AVAILABILITY_CHANGED, device, null);
-
}
return null;
}
}
+ /**
+ * Marks the device as available if the given timestamp is not outdated,
+ * compared to the time the device has been marked offline.
+ *
+ * @param deviceId identifier of the device
+ * @param timestamp of the event triggering this change.
+ * @return true if availability change request was accepted and changed the state
+ */
+ // Guarded by deviceDescs value (=Device lock)
+ private boolean markOnline(DeviceId deviceId, Timestamp timestamp) {
+ // accept on-line if given timestamp is newer than
+ // the latest offline request Timestamp
+ Timestamp offlineTimestamp = offline.get(deviceId);
+ if (offlineTimestamp == null ||
+ offlineTimestamp.compareTo(timestamp) < 0) {
+
+ offline.remove(deviceId);
+ return availableDevices.add(deviceId);
+ }
+ return false;
+ }
+
@Override
- public synchronized List<DeviceEvent> updatePorts(ProviderId providerId, DeviceId deviceId,
- List<PortDescription> portDescriptions) {
+ public synchronized List<DeviceEvent> updatePorts(ProviderId providerId,
+ DeviceId deviceId,
+ List<PortDescription> portDescriptions) {
Timestamp newTimestamp = clockService.getTimestamp(deviceId);
List<Timestamped<PortDescription>> deltaDescs = new ArrayList<>(portDescriptions.size());
@@ -252,7 +303,8 @@
deltaDescs.add(new Timestamped<PortDescription>(e, newTimestamp));
}
- List<DeviceEvent> events = updatePortsInternal(providerId, deviceId, deltaDescs);
+ List<DeviceEvent> events = updatePortsInternal(providerId, deviceId,
+ new Timestamped<>(portDescriptions, newTimestamp));
if (!events.isEmpty()) {
// FIXME: broadcast deltaDesc, UP
log.debug("broadcast deltaDesc");
@@ -261,8 +313,9 @@
}
- private List<DeviceEvent> updatePortsInternal(ProviderId providerId, DeviceId deviceId,
- List<Timestamped<PortDescription>> deltaDescs) {
+ private List<DeviceEvent> updatePortsInternal(ProviderId providerId,
+ DeviceId deviceId,
+ Timestamped<List<PortDescription>> portDescriptions) {
Device device = devices.get(deviceId);
checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
@@ -270,30 +323,41 @@
ConcurrentMap<ProviderId, DeviceDescriptions> descsMap = deviceDescs.get(deviceId);
checkArgument(descsMap != null, DEVICE_NOT_FOUND, deviceId);
- DeviceDescriptions descs = descsMap.get(providerId);
- // every provider must provide DeviceDescription.
- checkArgument(descs != null,
- "Device description for Device ID %s from Provider %s was not found",
- deviceId, providerId);
-
List<DeviceEvent> events = new ArrayList<>();
synchronized (descsMap) {
+
+ if (isDeviceRemoved(deviceId, portDescriptions.timestamp())) {
+ log.debug("Ignoring outdated events: {}", portDescriptions);
+ return null;
+ }
+
+ DeviceDescriptions descs = descsMap.get(providerId);
+ // every provider must provide DeviceDescription.
+ checkArgument(descs != null,
+ "Device description for Device ID %s from Provider %s was not found",
+ deviceId, providerId);
+
Map<PortNumber, Port> ports = getPortMap(deviceId);
+ final Timestamp newTimestamp = portDescriptions.timestamp();
+
// Add new ports
Set<PortNumber> processed = new HashSet<>();
- for (Timestamped<PortDescription> deltaDesc : deltaDescs) {
- final PortNumber number = deltaDesc.value().portNumber();
+ for (PortDescription portDescription : portDescriptions.value()) {
+ final PortNumber number = portDescription.portNumber();
+ processed.add(number);
+
final Port oldPort = ports.get(number);
final Port newPort;
+
final Timestamped<PortDescription> existingPortDesc = descs.getPortDesc(number);
if (existingPortDesc == null ||
- deltaDesc == existingPortDesc ||
- deltaDesc.isNewer(existingPortDesc)) {
+ newTimestamp.compareTo(existingPortDesc.timestamp()) >= 0) {
// on new port or valid update
// update description
- descs.putPortDesc(deltaDesc);
+ descs.putPortDesc(new Timestamped<>(portDescription,
+ portDescriptions.timestamp()));
newPort = composePort(device, number, descsMap);
} else {
// outdated event, ignored.
@@ -303,7 +367,6 @@
events.add(oldPort == null ?
createPort(device, newPort, ports) :
updatePort(device, oldPort, newPort, ports));
- processed.add(number);
}
events.addAll(pruneOldPorts(device, ports, processed));
@@ -313,7 +376,7 @@
// Creates a new port based on the port description adds it to the map and
// Returns corresponding event.
- // Guarded by deviceDescs value (=locking Device)
+ // Guarded by deviceDescs value (=Device lock)
private DeviceEvent createPort(Device device, Port newPort,
Map<PortNumber, Port> ports) {
ports.put(newPort.number(), newPort);
@@ -322,7 +385,7 @@
// Checks if the specified port requires update and if so, it replaces the
// existing entry in the map and returns corresponding event.
- // Guarded by deviceDescs value (=locking Device)
+ // Guarded by deviceDescs value (=Device lock)
private DeviceEvent updatePort(Device device, Port oldPort,
Port newPort,
Map<PortNumber, Port> ports) {
@@ -337,7 +400,7 @@
// Prunes the specified list of ports based on which ports are in the
// processed list and returns list of corresponding events.
- // Guarded by deviceDescs value (=locking Device)
+ // Guarded by deviceDescs value (=Device lock)
private List<DeviceEvent> pruneOldPorts(Device device,
Map<PortNumber, Port> ports,
Set<PortNumber> processed) {
@@ -389,13 +452,19 @@
ConcurrentMap<ProviderId, DeviceDescriptions> descsMap = deviceDescs.get(deviceId);
checkArgument(descsMap != null, DEVICE_NOT_FOUND, deviceId);
- DeviceDescriptions descs = descsMap.get(providerId);
- // assuming all providers must to give DeviceDescription
- checkArgument(descs != null,
- "Device description for Device ID %s from Provider %s was not found",
- deviceId, providerId);
-
synchronized (descsMap) {
+
+ if (isDeviceRemoved(deviceId, deltaDesc.timestamp())) {
+ log.debug("Ignoring outdated event: {}", deltaDesc);
+ return null;
+ }
+
+ DeviceDescriptions descs = descsMap.get(providerId);
+ // assuming all providers must to give DeviceDescription
+ checkArgument(descs != null,
+ "Device description for Device ID %s from Provider %s was not found",
+ deviceId, providerId);
+
ConcurrentMap<PortNumber, Port> ports = getPortMap(deviceId);
final PortNumber number = deltaDesc.value().portNumber();
final Port oldPort = ports.get(number);
@@ -443,19 +512,51 @@
}
@Override
- public DeviceEvent removeDevice(DeviceId deviceId) {
- ConcurrentMap<ProviderId, DeviceDescriptions> descs = getDeviceDescriptions(deviceId);
+ public synchronized DeviceEvent removeDevice(DeviceId deviceId) {
+ Timestamp timestamp = clockService.getTimestamp(deviceId);
+ DeviceEvent event = removeDeviceInternal(deviceId, timestamp);
+ // TODO: broadcast removal event
+ return event;
+ }
+
+ private DeviceEvent removeDeviceInternal(DeviceId deviceId,
+ Timestamp timestamp) {
+
+ Map<ProviderId, DeviceDescriptions> descs = getDeviceDescriptions(deviceId);
synchronized (descs) {
+ // accept removal request if given timestamp is newer than
+ // the latest Timestamp from Primary provider
+ DeviceDescriptions primDescs = getPrimaryDescriptions(descs);
+ Timestamp lastTimestamp = primDescs.getLatestTimestamp();
+ if (timestamp.compareTo(lastTimestamp) <= 0) {
+ // outdated event ignore
+ return null;
+ }
+ removalRequest.put(deviceId, timestamp);
+
Device device = devices.remove(deviceId);
// should DEVICE_REMOVED carry removed ports?
- devicePorts.get(deviceId).clear();
- availableDevices.remove(deviceId);
+ Map<PortNumber, Port> ports = devicePorts.get(deviceId);
+ if (ports != null) {
+ ports.clear();
+ }
+ markOfflineInternal(deviceId, timestamp);
descs.clear();
return device == null ? null :
new DeviceEvent(DEVICE_REMOVED, device, null);
}
}
+ private boolean isDeviceRemoved(DeviceId deviceId, Timestamp timestampToCheck) {
+ Timestamp removalTimestamp = removalRequest.get(deviceId);
+ if (removalTimestamp != null &&
+ removalTimestamp.compareTo(timestampToCheck) >= 0) {
+ // removalRequest is more recent
+ return true;
+ }
+ return false;
+ }
+
/**
* Returns a Device, merging description given from multiple Providers.
*
@@ -472,7 +573,7 @@
DeviceDescriptions desc = providerDescs.get(primary);
- DeviceDescription base = desc.getDeviceDesc().value();
+ final DeviceDescription base = desc.getDeviceDesc().value();
Type type = base.type();
String manufacturer = base.manufacturer();
String hwVersion = base.hwVersion();
@@ -545,7 +646,7 @@
* @return primary ProviderID, or randomly chosen one if none exists
*/
private ProviderId pickPrimaryPID(
- ConcurrentMap<ProviderId, DeviceDescriptions> providerDescs) {
+ Map<ProviderId, DeviceDescriptions> providerDescs) {
ProviderId fallBackPrimary = null;
for (Entry<ProviderId, DeviceDescriptions> e : providerDescs.entrySet()) {
if (!e.getKey().isAncillary()) {
@@ -558,6 +659,12 @@
return fallBackPrimary;
}
+ private DeviceDescriptions getPrimaryDescriptions(
+ Map<ProviderId, DeviceDescriptions> providerDescs) {
+ ProviderId pid = pickPrimaryPID(providerDescs);
+ return providerDescs.get(pid);
+ }
+
public static final class InitDeviceDescs
implements ConcurrentInitializer<DeviceDescriptions> {
@@ -586,6 +693,16 @@
this.portDescs = new ConcurrentHashMap<>();
}
+ Timestamp getLatestTimestamp() {
+ Timestamp latest = deviceDesc.get().timestamp();
+ for (Timestamped<PortDescription> desc : portDescs.values()) {
+ if (desc.timestamp().compareTo(latest) > 0) {
+ latest = desc.timestamp();
+ }
+ }
+ return latest;
+ }
+
public Timestamped<DeviceDescription> getDeviceDesc() {
return deviceDesc.get();
}
diff --git a/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleDeviceStore.java b/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleDeviceStore.java
index 0880ac9..514a22e 100644
--- a/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleDeviceStore.java
+++ b/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleDeviceStore.java
@@ -2,6 +2,8 @@
import com.google.common.collect.FluentIterable;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
import org.apache.commons.lang3.concurrent.ConcurrentException;
import org.apache.commons.lang3.concurrent.ConcurrentInitializer;
@@ -32,7 +34,6 @@
import org.slf4j.Logger;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
@@ -48,6 +49,7 @@
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Predicates.notNull;
+import static com.google.common.base.Verify.verify;
import static org.onlab.onos.net.device.DeviceEvent.Type.*;
import static org.slf4j.LoggerFactory.getLogger;
import static org.apache.commons.lang3.concurrent.ConcurrentUtils.createIfAbsentUnchecked;
@@ -71,14 +73,14 @@
// collection of Description given from various providers
private final ConcurrentMap<DeviceId,
ConcurrentMap<ProviderId, DeviceDescriptions>>
- deviceDescs = new ConcurrentHashMap<>();
+ deviceDescs = Maps.newConcurrentMap();
// cache of Device and Ports generated by compositing descriptions from providers
- private final ConcurrentMap<DeviceId, Device> devices = new ConcurrentHashMap<>();
- private final ConcurrentMap<DeviceId, ConcurrentMap<PortNumber, Port>> devicePorts = new ConcurrentHashMap<>();
+ private final ConcurrentMap<DeviceId, Device> devices = Maps.newConcurrentMap();
+ private final ConcurrentMap<DeviceId, ConcurrentMap<PortNumber, Port>> devicePorts = Maps.newConcurrentMap();
// available(=UP) devices
- private final Set<DeviceId> availableDevices = new HashSet<>();
+ private final Set<DeviceId> availableDevices = Sets.newConcurrentHashSet();
@Activate
@@ -88,6 +90,10 @@
@Deactivate
public void deactivate() {
+ deviceDescs.clear();
+ devices.clear();
+ devicePorts.clear();
+ availableDevices.clear();
log.info("Stopped");
}
@@ -107,45 +113,54 @@
}
@Override
- public synchronized DeviceEvent createOrUpdateDevice(ProviderId providerId, DeviceId deviceId,
+ public DeviceEvent createOrUpdateDevice(ProviderId providerId,
+ DeviceId deviceId,
DeviceDescription deviceDescription) {
+
ConcurrentMap<ProviderId, DeviceDescriptions> providerDescs
= getDeviceDescriptions(deviceId);
- Device oldDevice = devices.get(deviceId);
+ synchronized (providerDescs) {
+ // locking per device
- DeviceDescriptions descs
- = createIfAbsentUnchecked(providerDescs, providerId,
- new InitDeviceDescs(deviceDescription));
+ DeviceDescriptions descs
+ = createIfAbsentUnchecked(providerDescs, providerId,
+ new InitDeviceDescs(deviceDescription));
- // update description
- descs.putDeviceDesc(deviceDescription);
- Device newDevice = composeDevice(deviceId, providerDescs);
+ Device oldDevice = devices.get(deviceId);
+ // update description
+ descs.putDeviceDesc(deviceDescription);
+ Device newDevice = composeDevice(deviceId, providerDescs);
- if (oldDevice == null) {
- // ADD
- return createDevice(providerId, newDevice);
- } else {
- // UPDATE or ignore (no change or stale)
- return updateDevice(providerId, oldDevice, newDevice);
+ if (oldDevice == null) {
+ // ADD
+ return createDevice(providerId, newDevice);
+ } else {
+ // UPDATE or ignore (no change or stale)
+ return updateDevice(providerId, oldDevice, newDevice);
+ }
}
}
// Creates the device and returns the appropriate event if necessary.
+ // Guarded by deviceDescs value (=Device lock)
private DeviceEvent createDevice(ProviderId providerId, Device newDevice) {
// update composed device cache
- synchronized (this) {
- devices.putIfAbsent(newDevice.id(), newDevice);
- if (!providerId.isAncillary()) {
- availableDevices.add(newDevice.id());
- }
+ Device oldDevice = devices.putIfAbsent(newDevice.id(), newDevice);
+ verify(oldDevice == null,
+ "Unexpected Device in cache. PID:%s [old=%s, new=%s]",
+ providerId, oldDevice, newDevice);
+
+ if (!providerId.isAncillary()) {
+ availableDevices.add(newDevice.id());
}
return new DeviceEvent(DeviceEvent.Type.DEVICE_ADDED, newDevice, null);
}
// Updates the device and returns the appropriate event if necessary.
+ // Guarded by deviceDescs value (=Device lock)
private DeviceEvent updateDevice(ProviderId providerId, Device oldDevice, Device newDevice) {
// We allow only certain attributes to trigger update
@@ -153,70 +168,87 @@
!Objects.equals(oldDevice.swVersion(), newDevice.swVersion()) ||
!AnnotationsUtil.isEqual(oldDevice.annotations(), newDevice.annotations())) {
- synchronized (this) {
- devices.replace(newDevice.id(), oldDevice, newDevice);
- if (!providerId.isAncillary()) {
- availableDevices.add(newDevice.id());
- }
+ boolean replaced = devices.replace(newDevice.id(), oldDevice, newDevice);
+ if (!replaced) {
+ verify(replaced,
+ "Replacing devices cache failed. PID:%s [expected:%s, found:%s, new=%s]",
+ providerId, oldDevice, devices.get(newDevice.id())
+ , newDevice);
+ }
+ if (!providerId.isAncillary()) {
+ availableDevices.add(newDevice.id());
}
return new DeviceEvent(DeviceEvent.Type.DEVICE_UPDATED, newDevice, null);
}
// Otherwise merely attempt to change availability if primary provider
if (!providerId.isAncillary()) {
- synchronized (this) {
boolean added = availableDevices.add(newDevice.id());
return !added ? null :
new DeviceEvent(DEVICE_AVAILABILITY_CHANGED, newDevice, null);
- }
}
return null;
}
@Override
public DeviceEvent markOffline(DeviceId deviceId) {
- synchronized (this) {
+ ConcurrentMap<ProviderId, DeviceDescriptions> providerDescs
+ = getDeviceDescriptions(deviceId);
+
+ // locking device
+ synchronized (providerDescs) {
Device device = devices.get(deviceId);
- boolean removed = (device != null) && availableDevices.remove(deviceId);
- return !removed ? null :
- new DeviceEvent(DEVICE_AVAILABILITY_CHANGED, device, null);
+ if (device == null) {
+ return null;
+ }
+ boolean removed = availableDevices.remove(deviceId);
+ if (removed) {
+ // TODO: broadcast ... DOWN only?
+ return new DeviceEvent(DEVICE_AVAILABILITY_CHANGED, device, null);
+ }
+ return null;
}
}
@Override
- public synchronized List<DeviceEvent> updatePorts(ProviderId providerId, DeviceId deviceId,
- List<PortDescription> portDescriptions) {
+ public List<DeviceEvent> updatePorts(ProviderId providerId,
+ DeviceId deviceId,
+ List<PortDescription> portDescriptions) {
- // TODO: implement multi-provider
Device device = devices.get(deviceId);
checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
ConcurrentMap<ProviderId, DeviceDescriptions> descsMap = deviceDescs.get(deviceId);
checkArgument(descsMap != null, DEVICE_NOT_FOUND, deviceId);
- DeviceDescriptions descs = descsMap.get(providerId);
- checkArgument(descs != null,
- "Device description for Device ID %s from Provider %s was not found",
- deviceId, providerId);
-
-
List<DeviceEvent> events = new ArrayList<>();
- synchronized (this) {
- ConcurrentMap<PortNumber, Port> ports = getPortMap(deviceId);
+ synchronized (descsMap) {
+ DeviceDescriptions descs = descsMap.get(providerId);
+ // every provider must provide DeviceDescription.
+ checkArgument(descs != null,
+ "Device description for Device ID %s from Provider %s was not found",
+ deviceId, providerId);
+
+ Map<PortNumber, Port> ports = getPortMap(deviceId);
// Add new ports
Set<PortNumber> processed = new HashSet<>();
for (PortDescription portDescription : portDescriptions) {
- PortNumber number = portDescription.portNumber();
- Port oldPort = ports.get(number);
+ final PortNumber number = portDescription.portNumber();
+ processed.add(portDescription.portNumber());
+
+ final Port oldPort = ports.get(number);
+ final Port newPort;
+
+// event suppression hook?
+
// update description
descs.putPortDesc(portDescription);
- Port newPort = composePort(device, number, descsMap);
+ newPort = composePort(device, number, descsMap);
events.add(oldPort == null ?
- createPort(device, newPort, ports) :
- updatePort(device, oldPort, newPort, ports));
- processed.add(portDescription.portNumber());
+ createPort(device, newPort, ports) :
+ updatePort(device, oldPort, newPort, ports));
}
events.addAll(pruneOldPorts(device, ports, processed));
@@ -226,17 +258,19 @@
// Creates a new port based on the port description adds it to the map and
// Returns corresponding event.
+ // Guarded by deviceDescs value (=Device lock)
private DeviceEvent createPort(Device device, Port newPort,
- ConcurrentMap<PortNumber, Port> ports) {
+ Map<PortNumber, Port> ports) {
ports.put(newPort.number(), newPort);
return new DeviceEvent(PORT_ADDED, device, newPort);
}
// Checks if the specified port requires update and if so, it replaces the
// existing entry in the map and returns corresponding event.
+ // Guarded by deviceDescs value (=Device lock)
private DeviceEvent updatePort(Device device, Port oldPort,
Port newPort,
- ConcurrentMap<PortNumber, Port> ports) {
+ Map<PortNumber, Port> ports) {
if (oldPort.isEnabled() != newPort.isEnabled() ||
!AnnotationsUtil.isEqual(oldPort.annotations(), newPort.annotations())) {
@@ -248,6 +282,7 @@
// Prunes the specified list of ports based on which ports are in the
// processed list and returns list of corresponding events.
+ // Guarded by deviceDescs value (=Device lock)
private List<DeviceEvent> pruneOldPorts(Device device,
Map<PortNumber, Port> ports,
Set<PortNumber> processed) {
@@ -264,12 +299,6 @@
return events;
}
- private ConcurrentMap<ProviderId, DeviceDescriptions> getDeviceDescriptions(
- DeviceId deviceId) {
- return createIfAbsentUnchecked(deviceDescs, deviceId,
- NewConcurrentHashMap.<ProviderId, DeviceDescriptions>ifNeeded());
- }
-
// Gets the map of ports for the specified device; if one does not already
// exist, it creates and registers a new one.
private ConcurrentMap<PortNumber, Port> getPortMap(DeviceId deviceId) {
@@ -277,8 +306,14 @@
NewConcurrentHashMap.<PortNumber, Port>ifNeeded());
}
+ private ConcurrentMap<ProviderId, DeviceDescriptions> getDeviceDescriptions(
+ DeviceId deviceId) {
+ return createIfAbsentUnchecked(deviceDescs, deviceId,
+ NewConcurrentHashMap.<ProviderId, DeviceDescriptions>ifNeeded());
+ }
+
@Override
- public synchronized DeviceEvent updatePortStatus(ProviderId providerId, DeviceId deviceId,
+ public DeviceEvent updatePortStatus(ProviderId providerId, DeviceId deviceId,
PortDescription portDescription) {
Device device = devices.get(deviceId);
checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
@@ -286,19 +321,22 @@
ConcurrentMap<ProviderId, DeviceDescriptions> descsMap = deviceDescs.get(deviceId);
checkArgument(descsMap != null, DEVICE_NOT_FOUND, deviceId);
- DeviceDescriptions descs = descsMap.get(providerId);
- // assuming all providers must to give DeviceDescription
- checkArgument(descs != null,
- "Device description for Device ID %s from Provider %s was not found",
- deviceId, providerId);
+ synchronized (descsMap) {
+ DeviceDescriptions descs = descsMap.get(providerId);
+ // assuming all providers must to give DeviceDescription
+ checkArgument(descs != null,
+ "Device description for Device ID %s from Provider %s was not found",
+ deviceId, providerId);
- synchronized (this) {
ConcurrentMap<PortNumber, Port> ports = getPortMap(deviceId);
final PortNumber number = portDescription.portNumber();
- Port oldPort = ports.get(number);
+ final Port oldPort = ports.get(number);
+ final Port newPort;
+
// update description
descs.putPortDesc(portDescription);
- Port newPort = composePort(device, number, descsMap);
+ newPort = composePort(device, number, descsMap);
+
if (oldPort == null) {
return createPort(device, newPort, ports);
} else {
@@ -333,7 +371,7 @@
synchronized (descs) {
Device device = devices.remove(deviceId);
// should DEVICE_REMOVED carry removed ports?
- ConcurrentMap<PortNumber, Port> ports = devicePorts.get(deviceId);
+ Map<PortNumber, Port> ports = devicePorts.get(deviceId);
if (ports != null) {
ports.clear();
}
@@ -360,14 +398,14 @@
DeviceDescriptions desc = providerDescs.get(primary);
- // base
- Type type = desc.getDeviceDesc().type();
- String manufacturer = desc.getDeviceDesc().manufacturer();
- String hwVersion = desc.getDeviceDesc().hwVersion();
- String swVersion = desc.getDeviceDesc().swVersion();
- String serialNumber = desc.getDeviceDesc().serialNumber();
+ final DeviceDescription base = desc.getDeviceDesc();
+ Type type = base.type();
+ String manufacturer = base.manufacturer();
+ String hwVersion = base.hwVersion();
+ String swVersion = base.swVersion();
+ String serialNumber = base.serialNumber();
DefaultAnnotations annotations = DefaultAnnotations.builder().build();
- annotations = merge(annotations, desc.getDeviceDesc().annotations());
+ annotations = merge(annotations, base.annotations());
for (Entry<ProviderId, DeviceDescriptions> e : providerDescs.entrySet()) {
if (e.getKey().equals(primary)) {
@@ -386,7 +424,14 @@
hwVersion, swVersion, serialNumber, annotations);
}
- // probably want composePort"s" also
+ /**
+ * Returns a Port, merging description given from multiple Providers.
+ *
+ * @param device device the port is on
+ * @param number port number
+ * @param providerDescs Collection of Descriptions from multiple providers
+ * @return Port instance
+ */
private Port composePort(Device device, PortNumber number,
ConcurrentMap<ProviderId, DeviceDescriptions> providerDescs) {
@@ -441,7 +486,9 @@
public static final class InitDeviceDescs
implements ConcurrentInitializer<DeviceDescriptions> {
+
private final DeviceDescription deviceDesc;
+
public InitDeviceDescs(DeviceDescription deviceDesc) {
this.deviceDesc = checkNotNull(deviceDesc);
}
@@ -456,8 +503,6 @@
* Collection of Description of a Device and it's Ports given from a Provider.
*/
private static class DeviceDescriptions {
- // private final DeviceId id;
- // private final ProviderId pid;
private final AtomicReference<DeviceDescription> deviceDesc;
private final ConcurrentMap<PortNumber, PortDescription> portDescs;
@@ -475,10 +520,6 @@
return portDescs.get(number);
}
- public Collection<PortDescription> getPortDescs() {
- return Collections.unmodifiableCollection(portDescs.values());
- }
-
/**
* Puts DeviceDescription, merging annotations as necessary.
*
diff --git a/pom.xml b/pom.xml
index 8248c68..ad4ddcb 100644
--- a/pom.xml
+++ b/pom.xml
@@ -129,6 +129,12 @@
<version>1.9.13</version>
</dependency>
+ <dependency>
+ <groupId>org.easymock</groupId>
+ <artifactId>easymock</artifactId>
+ <version>3.2</version>
+ <scope>test</scope>
+ </dependency>
<!-- Web related -->
<dependency>
diff --git a/utils/netty/src/main/java/org/onlab/netty/Endpoint.java b/utils/netty/src/main/java/org/onlab/netty/Endpoint.java
index 8ed86ae..d6a87b5 100644
--- a/utils/netty/src/main/java/org/onlab/netty/Endpoint.java
+++ b/utils/netty/src/main/java/org/onlab/netty/Endpoint.java
@@ -68,4 +68,4 @@
}
return true;
}
-}
\ No newline at end of file
+}
diff --git a/utils/netty/src/main/java/org/onlab/netty/InternalMessage.java b/utils/netty/src/main/java/org/onlab/netty/InternalMessage.java
index e6c027e..96cbe79 100644
--- a/utils/netty/src/main/java/org/onlab/netty/InternalMessage.java
+++ b/utils/netty/src/main/java/org/onlab/netty/InternalMessage.java
@@ -86,4 +86,4 @@
return message;
}
}
-}
\ No newline at end of file
+}
diff --git a/utils/netty/src/main/java/org/onlab/netty/KryoSerializer.java b/utils/netty/src/main/java/org/onlab/netty/KryoSerializer.java
index 8a90c07..6df0b23 100644
--- a/utils/netty/src/main/java/org/onlab/netty/KryoSerializer.java
+++ b/utils/netty/src/main/java/org/onlab/netty/KryoSerializer.java
@@ -45,12 +45,12 @@
}
@Override
- public <T> T deserialize(ByteBuffer buffer) {
+ public <T> T decode(ByteBuffer buffer) {
return serializerPool.deserialize(buffer);
}
@Override
- public void serialize(Object obj, ByteBuffer buffer) {
+ public void encode(Object obj, ByteBuffer buffer) {
serializerPool.serialize(obj, buffer);
}
-}
\ No newline at end of file
+}
diff --git a/utils/netty/src/main/java/org/onlab/netty/MessageDecoder.java b/utils/netty/src/main/java/org/onlab/netty/MessageDecoder.java
index 3ed3216..a0d34a5 100644
--- a/utils/netty/src/main/java/org/onlab/netty/MessageDecoder.java
+++ b/utils/netty/src/main/java/org/onlab/netty/MessageDecoder.java
@@ -48,7 +48,7 @@
checkState(serializerVersion == MessageEncoder.SERIALIZER_VERSION, "Unexpected serializer version");
checkpoint(DecoderState.READ_CONTENT);
case READ_CONTENT:
- InternalMessage message = serializer.deserialize(buffer.readBytes(contentLength).nioBuffer());
+ InternalMessage message = serializer.decode(buffer.readBytes(contentLength).nioBuffer());
message.setMessagingService(messagingService);
out.add(message);
checkpoint(DecoderState.READ_HEADER_VERSION);
diff --git a/utils/netty/src/main/java/org/onlab/netty/Serializer.java b/utils/netty/src/main/java/org/onlab/netty/Serializer.java
index 56494b2..46550d4 100644
--- a/utils/netty/src/main/java/org/onlab/netty/Serializer.java
+++ b/utils/netty/src/main/java/org/onlab/netty/Serializer.java
@@ -24,20 +24,18 @@
public byte[] encode(Object data);
/**
- * Serializes the specified object into bytes using one of the
- * pre-registered serializers.
+ * Encodes the specified POJO into a byte buffer.
*
- * @param obj object to be serialized
+ * @param data POJO to be encoded
* @param buffer to write serialized bytes
*/
- public void serialize(final Object obj, ByteBuffer buffer);
+ public void encode(final Object data, ByteBuffer buffer);
/**
- * Deserializes the specified bytes into an object using one of the
- * pre-registered serializers.
+ * Decodes the specified byte buffer to a POJO.
*
- * @param buffer bytes to be deserialized
- * @return deserialized object
+ * @param buffer bytes to be decoded
+ * @return POJO
*/
- public <T> T deserialize(final ByteBuffer buffer);
+ public <T> T decode(final ByteBuffer buffer);
}
diff --git a/utils/netty/src/main/java/org/onlab/netty/package-info.java b/utils/netty/src/main/java/org/onlab/netty/package-info.java
index b1b90a3..fee7b04 100644
--- a/utils/netty/src/main/java/org/onlab/netty/package-info.java
+++ b/utils/netty/src/main/java/org/onlab/netty/package-info.java
@@ -1,4 +1,4 @@
/**
* Asynchronous messaging APIs implemented using the Netty framework.
*/
-package org.onlab.netty;
\ No newline at end of file
+package org.onlab.netty;